From 3c06ae97e838897490cecc0c01d002ee7f26be70 Mon Sep 17 00:00:00 2001 From: Zyy Date: Tue, 4 Aug 2026 15:11:54 +0800 Subject: [PATCH] fix: reconcile quota, backup, and session behavior - preserve confirmed quota exhaustion until its recorded reset advances - normalize account quota presentation and subagent session affinity - repair S3-compatible backup handling and related admin UI details Co-authored-by: Codex --- .../crates/gateway-admin/src/backup/task.rs | 36 ++- .../gateway-admin/tests/backup/support.rs | 18 +- .../crates/gateway-admin/tests/backup/task.rs | 76 ++++- .../gateway-core/src/engine/coordinator.rs | 38 +++ .../gateway-core/tests/engine/coordinator.rs | 84 +++++- backend/crates/providers/openai/src/admin.rs | 72 +++-- .../openai/src/credential/affinity.rs | 26 +- .../openai/src/credential/quota/mod.rs | 223 +++++++++++--- .../openai/src/credential/quota/recovery.rs | 45 +-- .../openai/src/credential/quota/snapshot.rs | 63 ++-- .../openai/src/credential/selector.rs | 112 ++++--- .../crates/providers/openai/src/provider.rs | 16 + .../openai/src/transport/client_sse.rs | 7 +- .../openai/src/transport/diagnostics.rs | 37 +++ .../src/transport/protocol/responses.rs | 20 ++ .../crates/providers/openai/tests/admin.rs | 49 +++- .../openai/tests/credential/contract.rs | 132 ++++++++- .../openai/tests/credential/quota/mod.rs | 71 ++++- .../openai/tests/credential/quota/snapshot.rs | 30 +- .../openai/tests/provider/contract.rs | 142 +++++++-- .../openai/tests/transport/diagnostics.rs | 19 +- .../openai/tests/transport/headers.rs | 64 ++++ .../openai/tests/transport/request.rs | 35 +++ frontend/src/components/base/BasePopover.vue | 18 +- .../components/AccountStatusBadge.vue | 276 ++++++++++-------- frontend/src/views/accounts/constants.ts | 34 ++- .../components/backup/BackupRecordsCard.vue | 12 +- .../components/UsageRecordDetailModal.vue | 6 +- 28 files changed, 1360 insertions(+), 401 deletions(-) diff --git a/backend/crates/gateway-admin/src/backup/task.rs b/backend/crates/gateway-admin/src/backup/task.rs index bd2992026..34ae8eaf7 100644 --- a/backend/crates/gateway-admin/src/backup/task.rs +++ b/backend/crates/gateway-admin/src/backup/task.rs @@ -220,7 +220,8 @@ impl BackupTask { sha256: Some(artifact.sha256.clone()), ..Default::default() }; - self.repository + let Some(uploading_record) = self + .repository .transition_status( &record.id, BackupStatus::Dumping, @@ -229,9 +230,12 @@ impl BackupTask { now, ) .await - .map_err(repo_error)?; + .map_err(repo_error)? + else { + return Ok(()); + }; self.upload_and_verify( - record, + &uploading_record, &artifact.path, artifact.size_bytes, &artifact.sha256, @@ -394,7 +398,8 @@ impl BackupTask { sha256: Some(artifact.sha256.clone()), ..Default::default() }; - self.repository + let Some(uploading_record) = self + .repository .transition_status( &record.id, BackupStatus::Dumping, @@ -403,10 +408,13 @@ impl BackupTask { now, ) .await - .map_err(repo_error)?; + .map_err(repo_error)? + else { + return Ok(()); + }; self.upload_and_verify( - record, + &uploading_record, &artifact.path, artifact.size_bytes, &artifact.sha256, @@ -448,8 +456,12 @@ impl BackupTask { .await { let _ = store.delete_object(&config, &record.object_key).await; - self.fail_task(record, error.code(), "上传到对象存储失败") - .await?; + self.fail_task( + record, + error.code(), + format!("上传到对象存储失败：{}", error.message()), + ) + .await?; return Ok(()); } @@ -486,8 +498,12 @@ impl BackupTask { } Err(error) => { let _ = store.delete_object(&config, &record.object_key).await; - self.fail_task(record, error.code(), "远端对象校验失败") - .await?; + self.fail_task( + record, + error.code(), + format!("远端对象校验失败：{}", error.message()), + ) + .await?; } } Ok(()) diff --git a/backend/crates/gateway-admin/tests/backup/support.rs b/backend/crates/gateway-admin/tests/backup/support.rs index 1ba63b5c4..547c6a272 100644 --- a/backend/crates/gateway-admin/tests/backup/support.rs +++ b/backend/crates/gateway-admin/tests/backup/support.rs @@ -13,9 +13,10 @@ use gateway_admin::{ MutationActor, MutationContext, Revision, auth::AdminAuditEvent, backup::{ - BackupObjectMetadata, BackupRecord, BackupRecordListQuery, BackupRecordPage, - BackupRecordSeed, BackupSettings, BackupStatus, BackupStorageConfig, BackupTriggerKind, - ConnectionTestResult, UpdateBackupScheduleCommand, UpdateBackupStorageCommand, + BackupError, BackupObjectMetadata, BackupRecord, BackupRecordListQuery, + BackupRecordPage, BackupRecordSeed, BackupSettings, BackupStatus, BackupStorageConfig, + BackupTriggerKind, ConnectionTestResult, UpdateBackupScheduleCommand, + UpdateBackupStorageCommand, }, }, ports::{ @@ -497,15 +498,23 @@ impl DatabaseDumpPort for FakeDumpPort { /// 内存版对象存储：把上传内容按 key 保存并支持 Head/Delete/预签名。 pub(crate) struct FakeObjectStore { pub(crate) objects: Mutex>>, + upload_error: Option, } impl FakeObjectStore { pub(crate) fn new() -> Self { Self { objects: Mutex::new(std::collections::HashMap::new()), + upload_error: None, } } + /// 让上传返回指定的脱敏 S3 错误。 + pub(crate) fn fail_upload(mut self, code: &'static str, message: impl Into) -> Self { + self.upload_error = Some(BackupError::new(code, message.into())); + self + } + pub(crate) fn object(&self, key: &str) -> Option> { self.objects.lock().expect("objects").get(key).cloned() } @@ -530,6 +539,9 @@ impl BackupObjectStorePort for FakeObjectStore { _config: &BackupStorageConfig, request: UploadObjectRequest, ) -> Result<(), gateway_admin::model::backup::BackupError> { + if let Some(error) = &self.upload_error { + return Err(error.clone()); + } let content = std::fs::read(&request.source).map_err(|_| { gateway_admin::model::backup::BackupError::new( gateway_admin::model::backup::code::S3_UPLOAD_FAILED, diff --git a/backend/crates/gateway-admin/tests/backup/task.rs b/backend/crates/gateway-admin/tests/backup/task.rs index 8915d0acb..2174711d4 100644 --- a/backend/crates/gateway-admin/tests/backup/task.rs +++ b/backend/crates/gateway-admin/tests/backup/task.rs @@ -6,9 +6,9 @@ use chrono::{Duration, Utc}; use gateway_admin::{ backup::task::BackupTask, model::backup::{ - BackupRecordSeed, BackupStatus, BackupTriggerKind, UpdateBackupScheduleCommand, + BackupRecordSeed, BackupStatus, BackupTriggerKind, UpdateBackupScheduleCommand, code, }, - ports::backup::BackupRepository, + ports::backup::{BackupRepository, DatabaseDumpPort, DumpRequest}, }; use gateway_core::engine::CancellationToken; @@ -89,6 +89,78 @@ async fn backup_task_marks_failed_when_dump_port_fails() { assert!(object_store.object(&seed.object_key).is_none()); } +#[tokio::test] +async fn backup_task_marks_failed_with_the_upload_error() { + let repository = Arc::new(FakeBackupRepository::new(configured_settings())); + let dump = Arc::new(FakeDumpPort::new()); + let object_store = + Arc::new(FakeObjectStore::new().fail_upload(code::S3_UPLOAD_FAILED, "对象存储拒绝上传")); + let seed = BackupRecordSeed { + id: backup_id("upload-failure"), + trigger_kind: BackupTriggerKind::Manual, + scheduled_at: None, + object_key: "codex/2026/08/01/upload-failure.dump".to_owned(), + expires_at: None, + }; + repository + .insert_backup_record(seed.clone()) + .await + .expect("insert queued"); + + let task = BackupTask::new(repository.clone(), dump, object_store); + task.run_cycle(&CancellationToken::new()) + .await + .expect("run cycle"); + + let records = repository.all_records(); + let failed = &records[0]; + assert_eq!(failed.status, BackupStatus::Failed); + assert_eq!(failed.error_code.as_deref(), Some(code::S3_UPLOAD_FAILED)); + assert_eq!( + failed.error_message.as_deref(), + Some("上传到对象存储失败：对象存储拒绝上传") + ); +} + +#[tokio::test] +async fn backup_task_recovery_marks_failed_with_the_upload_error() { + let repository = Arc::new(FakeBackupRepository::new(configured_settings())); + let dump = Arc::new(FakeDumpPort::new()); + let object_store = + Arc::new(FakeObjectStore::new().fail_upload(code::S3_UPLOAD_FAILED, "对象存储拒绝上传")); + let seed = BackupRecordSeed { + id: backup_id("recovered-upload-failure"), + trigger_kind: BackupTriggerKind::Manual, + scheduled_at: None, + object_key: "codex/2026/08/01/recovered-upload-failure.dump".to_owned(), + expires_at: None, + }; + repository + .insert_backup_record(seed.clone()) + .await + .expect("insert queued"); + let cancellation = CancellationToken::new(); + dump.dump(DumpRequest { + backup_id: seed.id.clone(), + cancellation: cancellation.clone(), + }) + .await + .expect("stage dump"); + repository.force_status(&seed.id, BackupStatus::Dumping); + + let task = BackupTask::new(repository.clone(), dump, object_store); + task.run_cycle(&cancellation).await.expect("run cycle"); + + let records = repository.all_records(); + let failed = &records[0]; + assert_eq!(failed.status, BackupStatus::Failed); + assert_eq!(failed.error_code.as_deref(), Some(code::S3_UPLOAD_FAILED)); + assert_eq!( + failed.error_message.as_deref(), + Some("上传到对象存储失败：对象存储拒绝上传") + ); +} + #[tokio::test] async fn retention_cleans_expired_scheduled_backups() { let repository = Arc::new(FakeBackupRepository::new(configured_settings())); diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 08e99b733..6bddd6f83 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -660,6 +660,9 @@ where ProviderBoundary::Result(result) => match *result { Ok(stream) => stream, Err(error) => { + if self.prepare_unavailable_native_continuation_replay(&error) { + return Ok(Some(PullOutcome::AttemptDiscarded)); + } if matches!( error.kind(), ProviderErrorKind::AccountCapacityUnavailable @@ -1096,6 +1099,41 @@ where true } + /// 原生续写的原账号在本地调度阶段已不可用时，改用完整输入重放。 + /// + /// 这不是强制 Smart：下一次 attempt 仍使用请求配置的调度策略。只有网关持有 + /// 对应 Provider 的可携带会话状态时，才可安全移除 native handle 并换号。 + fn prepare_unavailable_native_continuation_replay(&mut self, error: &ProviderError) -> bool { + if self.account_selection.required_account().is_some() + || self.continuation_attempt != ContinuationAttempt::Native + || !matches!( + error.kind(), + ProviderErrorKind::NoEligibleAccount + | ProviderErrorKind::AccountCapacityUnavailable + ) + { + return false; + } + let Some(pin) = self + .continuation + .as_ref() + .and_then(ContinuationBinding::pinned) + else { + return false; + }; + if self + .operation + .provider_session_state(pin.provider().as_str()) + .is_none() + { + return false; + } + + self.continuation_attempt = ContinuationAttempt::ReplayAny; + self.excluded_accounts.insert(pin.account().clone()); + true + } + fn reset_uncommitted_observations(&mut self) { self.usage = Usage::new(); self.cost = CostEstimate::unavailable(); diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 015ad3338..0e6ac914f 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -513,17 +513,27 @@ fn image_stream(image_output_tokens: Option) -> Vec RoutingPlan { + plan_with_policy( + operation, + AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(2).expect("account concurrency"), + Duration::from_millis(50), + ), + ) +} + +fn plan_with_policy( + operation: &Operation, + account_selection_policy: AccountSelectionPolicy, +) -> RoutingPlan { let provider = ProviderKind::new("openai").expect("provider"); let public_model = PublicModelId::new("gpt-5").expect("public model"); let capabilities = ModelCapabilities::new(BTreeSet::from([operation.kind()]), Some(32_000)) .with_upstream_feature_validation(); RuntimeSnapshot::new( ConfigRevision::new(1).expect("config revision"), - AccountSelectionPolicy::new( - RotationStrategy::Smart, - NonZeroU32::new(2).expect("account concurrency"), - Duration::from_millis(50), - ), + account_selection_policy, vec![provider.clone()], vec![ProviderModel::new( provider.clone(), @@ -1514,6 +1524,70 @@ fn native_continuation_replays_owner_before_safely_switching_account() { ); } +#[test] +fn unavailable_native_continuation_replays_with_the_configured_account_policy() { + let Operation::Generate(generate) = generate_operation() else { + panic!("generate operation"); + }; + let operation = Operation::Generate(generate.with_provider_session_state( + ProviderSessionState::new("openai", Map::new()).expect("provider session state"), + )); + let policy = AccountSelectionPolicy::new( + RotationStrategy::RoundRobin, + NonZeroU32::new(2).expect("account concurrency"), + Duration::from_millis(50), + ); + let route_plan = plan_with_policy(&operation, policy); + let (coordinator, store, provider) = coordinator(vec![ + Script::Error(ProviderError::new( + ProviderErrorKind::NoEligibleAccount, + UpstreamSendState::NotSent, + )), + Script::Stream { + account_id: "acct_two", + items: complete_stream(None), + }, + ]); + let original = ProviderAccountId::new("acct_one").expect("account"); + let continuation = NativeContinuationPin::new( + PreviousResponseId::new("previous-response"), + PreviousResponseId::new("upstream-response"), + ProviderKind::new("openai").expect("provider"), + original.clone(), + ); + + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + Some(ContinuationBinding::Pinned(continuation)), + CancellationToken::new(), + )) + .expect("start execution"); + block_on(session.collect_uncommitted()).expect("portable replay succeeds"); + block_on(session.commit_downstream(Some(200))).expect("commit response"); + + let contexts = provider.contexts.lock().expect("contexts lock"); + assert_eq!(contexts.len(), 2); + assert_eq!( + contexts[0].continuation_attempt(), + ContinuationAttempt::Native + ); + assert_eq!( + contexts[1].continuation_attempt(), + ContinuationAttempt::ReplayAny + ); + assert_eq!( + contexts[1].account_selection_policy().strategy(), + RotationStrategy::RoundRobin + ); + assert!(contexts[1].excluded_accounts().contains(&original)); + let state = store.state.lock().expect("store lock"); + assert_eq!(state.attempts.len(), 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); +} + #[test] fn diagnostic_account_reaches_provider_and_matching_metadata_succeeds() { let operation = operation(RetrySafety::Idempotent); diff --git a/backend/crates/providers/openai/src/admin.rs b/backend/crates/providers/openai/src/admin.rs index d5466ee05..1f1aaf85a 100644 --- a/backend/crates/providers/openai/src/admin.rs +++ b/backend/crates/providers/openai/src/admin.rs @@ -50,7 +50,7 @@ use crate::credential::{ CodexCredentialAdminService, CodexCredentialCatalogError, CodexCredentialCatalogService, CodexCredentialQuotaError, CodexCredentialQuotaService, CodexOAuthAdmin, CodexOAuthAdminError, CodexOAuthPendingClaimOutcome, CodexOAuthPendingStore, CodexOAuthPendingStoreError, - CodexPendingAuthorization, CodexQuotaWindowKind, CodexQuotaWindowRole, + CodexPendingAuthorization, CodexQuotaWindow, CodexQuotaWindowKind, CodexQuotaWindowRole, CompleteCodexOAuthAuthorization, CompletedCodexOAuthCredential, ExportManagedCodexCredential, RotateManagedCodexCredential, StartCodexOAuthAuthorization, StoredCodexPendingAuthorization, refresh_time, @@ -782,6 +782,7 @@ fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota let windows: Vec = snapshot .windows() .iter() + .filter(|window| should_project_quota_window(window)) .map(|window| { let mut data = Map::new(); data.insert( @@ -798,6 +799,7 @@ fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota label: codex_quota_window_label( window.limit_name(), window.kind(), + window.role(), window.source(), window.window_seconds(), ), @@ -826,6 +828,17 @@ fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota } } +/// `secondary_window` 有时只是上游的空占位：没有时长、重置时间，也没有用量。 +/// 它不能被可靠翻译成 5 小时或周额度，因此不应占用账号面板；带有实际事实的 +/// 次级窗口（时长、重置、非零用量或触顶）仍完整保留。 +fn should_project_quota_window(window: &CodexQuotaWindow) -> bool { + window.role() != CodexQuotaWindowRole::Secondary + || window.window_seconds().is_some() + || window.reset_at().is_some() + || window.used_percent().is_some_and(|used| used > 0.0) + || window.limit_reached() +} + fn quota_windows_limit_reached(windows: &[ProviderQuotaWindow]) -> bool { windows.iter().any(|window| window.limit_reached) } @@ -891,45 +904,66 @@ const fn quota_role(role: CodexQuotaWindowRole) -> &'static str { } } -/// 窗口展示名，优先使用上游 `limit_name`（原文，`_` 转空格）； -/// 无 limit_name 时按窗口时长猜（±5% 容差，汉化输出），猜不中兜底「额度」。 +/// 按窗口时长显示汉化额度名；非核心桶补上上游桶名称作为前缀。 fn codex_quota_window_label( limit_name: Option<&str>, kind: CodexQuotaWindowKind, + role: CodexQuotaWindowRole, source: &str, window_seconds: Option, ) -> String { - if let Some(limit_name) = limit_name { - return limit_name.replace('_', " "); - } let base = match kind { - CodexQuotaWindowKind::Monthly => "月限额".to_owned(), - CodexQuotaWindowKind::Weekly => "周限额".to_owned(), + CodexQuotaWindowKind::Monthly => "月额度".to_owned(), + CodexQuotaWindowKind::Weekly => "周额度".to_owned(), CodexQuotaWindowKind::ShortTerm => { if window_seconds.is_some_and(|seconds| seconds > 86_400) { - "周限额".to_owned() + "周额度".to_owned() } else { - "5小时限额".to_owned() + "5小时额度".to_owned() } } - CodexQuotaWindowKind::Other => custom_quota_window_label(window_seconds), + CodexQuotaWindowKind::Other => custom_quota_window_label(window_seconds, role), }; - if matches!(source, "core" | "codex" | "code_review" | "spend_control") { - return base; + if let Some(prefix) = codex_quota_window_prefix(limit_name, source) { + return format!("{prefix} · {base}"); } - format!("{source} · {base}") + base } -fn custom_quota_window_label(window_seconds: Option) -> String { +fn codex_quota_window_prefix(limit_name: Option<&str>, source: &str) -> Option { + let source_identifier = quota_identifier(source); + if matches!(source_identifier.as_str(), "core" | "codex") { + return None; + } + let label = limit_name.unwrap_or(source); + match quota_identifier(label).as_str() { + "premium" => Some("Premium".to_owned()), + "code_review" => Some("代码审查".to_owned()), + "spend_control" => Some("消费控制".to_owned()), + _ => Some(label.replace('_', " ")), + } +} + +fn quota_identifier(value: &str) -> String { + value.trim().to_ascii_lowercase().replace(['-', ' '], "_") +} + +fn custom_quota_window_label(window_seconds: Option, role: CodexQuotaWindowRole) -> String { let Some(seconds) = window_seconds.filter(|seconds| *seconds > 0) else { - return "额度".to_owned(); + // 与官方客户端一致：没有时长时不臆测为 5 小时或周额度；保留 + // primary/secondary 语义，让用户知道这是上游未标明时长的独立窗口。 + return match role { + CodexQuotaWindowRole::Primary => "主额度".to_owned(), + CodexQuotaWindowRole::Secondary => "次级额度".to_owned(), + CodexQuotaWindowRole::Monthly => "月额度".to_owned(), + }; }; if seconds % 86_400 == 0 { - format!("{}日限额", seconds / 86_400) + format!("{}日额度", seconds / 86_400) } else if seconds % 3_600 == 0 { - format!("{}小时限额", seconds / 3_600) + format!("{}小时额度", seconds / 3_600) } else { - format!("{}分钟限额", seconds.div_ceil(60)) + format!("{}分钟额度", seconds.div_ceil(60)) } } diff --git a/backend/crates/providers/openai/src/credential/affinity.rs b/backend/crates/providers/openai/src/credential/affinity.rs index 97cfc5fdb..d7a8def2b 100644 --- a/backend/crates/providers/openai/src/credential/affinity.rs +++ b/backend/crates/providers/openai/src/credential/affinity.rs @@ -10,19 +10,33 @@ use crate::transport::request::derive_conversation_anchor; pub(crate) fn derive_codex_session_affinity_key( request: &CodexResponsesRequest, ) -> Option { - if let Some(conversation_id) = request + let session_key = if let Some(conversation_id) = request .local_conversation_id .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) { if let Ok(key) = ProviderSessionAffinityKey::try_new(conversation_id.to_owned()) { - return Some(key); + Some(key) + } else { + opaque_affinity_key("local-conversation", conversation_id) } - return opaque_affinity_key("local-conversation", conversation_id); - } - let (domain, value) = derive_conversation_anchor(request)?; - opaque_affinity_key(domain, &value) + } else { + let (domain, value) = derive_conversation_anchor(request)?; + opaque_affinity_key(domain, &value) + }; + + let session_discriminator = request.subagent_kind(); + session_key.and_then(|session_key| match session_discriminator.as_deref() { + None => Some(session_key), + Some(discriminator) => { + // 仅扩展原会话锚点的键空间；后续查找、绑定和调度仍使用同一套亲和逻辑。 + opaque_affinity_key( + "subagent-session", + &format!("{discriminator}\0{}", session_key.expose_to_store()), + ) + } + }) } fn opaque_affinity_key(domain: &str, value: &str) -> Option { diff --git a/backend/crates/providers/openai/src/credential/quota/mod.rs b/backend/crates/providers/openai/src/credential/quota/mod.rs index 9a94d3480..b8b77f1e3 100644 --- a/backend/crates/providers/openai/src/credential/quota/mod.rs +++ b/backend/crates/providers/openai/src/credential/quota/mod.rs @@ -13,7 +13,7 @@ pub use snapshot::{ use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, RwLock}; -use std::time::{Duration, Instant, SystemTime}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use chrono::Utc; use gateway_core::engine::credential::{ @@ -36,7 +36,7 @@ use crate::transport::{CodexBackendClient, CodexClientError, CodexRequestContext use super::agent_identity::{CodexAgentIdentityTaskService, PreparedCodexRuntimeCredential}; use super::repository::{CodexCredentialRepository, CredentialRepositoryError}; -use recovery::{QuotaRecoveryEvidence, quota_state_transition, quota_success_state}; +use recovery::{quota_state_transition, quota_success_state}; use snapshot::{ parse_account_quota_snapshot, quota_projection_ttl, quota_snapshot_from_observation, scheduling_signals_from_snapshot, @@ -395,7 +395,7 @@ impl CodexCredentialQuotaService { } } - /// 写入新快照前读取旧确认 reset（402 恢复用「旧 reset 已到期」判定）。 + /// 写入新快照前读取已确认耗尽窗口的 reset，用于判断窗口是否真的推进。 async fn previous_confirmed_reset(&self, account: &ProviderAccount) -> Option { let observations = self .store @@ -410,6 +410,52 @@ impl CodexCredentialQuotaService { snapshot.fact().resets_at().map(SystemTime::from) } + /// 将已确认的额度拒绝投影为 100%，使展示、调度和恢复共用同一窗口事实。 + /// + /// `reset_at` 来自官方 `usage_limit_reached` 错误体；缺失时保留当前快照的 reset。 + pub(crate) async fn record_confirmed_exhaustion( + &self, + account: &ProviderAccount, + reset_at: Option, + ) -> Result<(), CodexCredentialQuotaError> { + let existing = self + .store + .get_quotas(std::slice::from_ref(account.id())) + .await? + .into_iter() + .find(|observation| { + observation.account_id == *account.id() + && observation.expected_revision == account.revision() + }) + .and_then(|observation| observation.quota) + .map(OpaqueProviderData::into_inner) + .unwrap_or_default(); + let quota = confirmed_exhaustion_projection(existing, reset_at); + let observed_at = SystemTime::now(); + let outcome = self + .store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: Some(OpaqueProviderData::new(quota.clone())), + observed_at: Some(observed_at), + limit_reached: Some(true), + }) + .await?; + if outcome == QuotaWriteOutcome::Conflict { + return Err(CodexCredentialQuotaError::RevisionConflict); + } + if let Ok(snapshot) = parse_account_quota_snapshot( + account.id().clone(), + account.revision(), + observed_at, + &Value::Object(quota), + ) { + self.scheduling.observe(&snapshot); + } + Ok(()) + } + /// 批量预热请求级额度投影；持久层或 Provider JSON 异常只退化为未知额度。 pub async fn prepare_scheduling(&self, accounts: &[ProviderAccount]) { if self.scheduling.hydration_targets(accounts).is_empty() { @@ -539,18 +585,42 @@ impl CodexCredentialQuotaService { observed_at: SystemTime, summary: &mut CodexQuotaSyncSummary, ) -> Result<(), CodexCredentialQuotaError> { - let snapshot = parse_account_quota_snapshot( + let refreshed_snapshot = parse_account_quota_snapshot( account.id().clone(), account.revision(), observed_at, value, )?; - let fact = snapshot.fact(); + let refreshed_fact = refreshed_snapshot.fact(); let previous_reset_at = self.previous_confirmed_reset(account).await; + let now = SystemTime::now(); let object = value .as_object() .cloned() .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; + let object = if account.availability() == AccountAvailability::QuotaExhausted + && quota_success_state( + account.availability(), + refreshed_fact, + previous_reset_at, + now, + ) + .is_none() + { + confirmed_exhaustion_projection( + object, + confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + ) + } else { + object + }; + let snapshot = parse_account_quota_snapshot( + account.id().clone(), + account.revision(), + observed_at, + &Value::Object(object.clone()), + )?; + let fact = snapshot.fact(); let outcome = self .store .compare_and_swap_quota(QuotaObservation { @@ -581,10 +651,9 @@ impl CodexCredentialQuotaService { } if let Some(availability) = quota_success_state( current.availability(), - fact, + refreshed_fact, previous_reset_at, - SystemTime::now(), - QuotaRecoveryEvidence::RefreshedSnapshot, + now, ) { let _ = self .repository @@ -607,6 +676,11 @@ impl CodexCredentialQuotaService { .limits .values() .any(|details| passive_rate_limit_snapshot(details).is_some()); + let previous_reset_at = if account.availability() == AccountAvailability::QuotaExhausted { + self.previous_confirmed_reset(account).await + } else { + None + }; let existing = self .store .get_quotas(std::slice::from_ref(account.id())) @@ -646,6 +720,17 @@ impl CodexCredentialQuotaService { } let quota = merge_passive_quota(existing, &rate_limits); let observed_at = SystemTime::now(); + let refreshed_fact = parse_codex_quota_usage(&Value::Object(quota.clone()))?; + let quota = if account.availability() == AccountAvailability::QuotaExhausted { + // 被动成功响应不能解除确认额度耗尽；继续展示 100%，直到权威 usage 快照 + // 证明旧 reset 已跨过且新窗口已经推进。 + confirmed_exhaustion_projection( + quota, + confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + ) + } else { + quota + }; let fact = parse_codex_quota_usage(&Value::Object(quota.clone()))?; let snapshot = parse_account_quota_snapshot( account.id().clone(), @@ -666,28 +751,9 @@ impl CodexCredentialQuotaService { if outcome == QuotaWriteOutcome::Conflict { return Ok(false); } - // 可用性写回必须基于最新账号状态：revision fence 只覆盖凭据轮换， - // 不覆盖其他路径的仅状态变化（Cooldown/QuotaExhausted 转换）。 - let Some(current) = self.store.get_account(account.id()).await? else { - return Ok(false); - }; - if current.revision() != account.revision() { - return Ok(false); - } if let Ok(snapshot) = snapshot { self.scheduling.observe(&snapshot); } - if let Some(availability) = quota_success_state( - current.availability(), - fact, - None, - observed_at, - QuotaRecoveryEvidence::SuccessfulResponse, - ) { - self.repository - .apply_state(¤t, availability, observed_at) - .await?; - } Ok(true) } @@ -852,17 +918,41 @@ impl CodexCredentialQuotaService { }); } }; - let snapshot = parse_account_quota_snapshot( + let refreshed_snapshot = parse_account_quota_snapshot( account.id().clone(), account.revision(), observed_at, &value, )?; + let refreshed_fact = refreshed_snapshot.fact(); let previous_reset_at = self.previous_confirmed_reset(&account).await; + let now = SystemTime::now(); let object = value .as_object() .cloned() .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; + let object = if account.availability() == AccountAvailability::QuotaExhausted + && quota_success_state( + account.availability(), + refreshed_fact, + previous_reset_at, + now, + ) + .is_none() + { + confirmed_exhaustion_projection( + object, + confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact), + ) + } else { + object + }; + let snapshot = parse_account_quota_snapshot( + account.id().clone(), + account.revision(), + observed_at, + &Value::Object(object.clone()), + )?; if self .store .compare_and_swap_quota(QuotaObservation { @@ -888,10 +978,9 @@ impl CodexCredentialQuotaService { self.scheduling.observe(&snapshot); if let Some(availability) = quota_success_state( current.availability(), - snapshot.fact(), + refreshed_fact, previous_reset_at, - SystemTime::now(), - QuotaRecoveryEvidence::RefreshedSnapshot, + now, ) { self.repository .apply_state(¤t, availability, observed_at) @@ -1053,14 +1142,6 @@ fn merge_passive_quota( .remove("additional_rate_limits") .and_then(|value| value.as_array().cloned()) .unwrap_or_default(); - let replaces_core_limit = active_limit - .and_then(|limit_id| rate_limits.limits.get(limit_id)) - .is_some_and(|details| passive_rate_limit_snapshot(details).is_some()); - if replaces_core_limit { - // `codex` additional 与顶层 `rate_limit` 归为同一个 core 源。保留旧 - // additional 会在解析时反向覆盖当前 active limit，导致页面显示旧百分比。 - additional.retain(|item| !is_codex_core_alias(item)); - } for (limit_id, details) in &rate_limits.limits { let Some(rate_limit) = passive_rate_limit_snapshot(details) else { continue; @@ -1100,14 +1181,64 @@ fn merge_passive_quota( quota } -fn is_codex_core_alias(value: &Value) -> bool { +/// 用已确认的上游额度拒绝覆盖 core 限额投影。 +/// +/// 上游 usage 快照可能在结算期间短暂滞后为 98/99% 或 `allowed=true`；保留其他 +/// 原始 metadata，但把 core 窗口固定为 100%，直到权威快照证明进入下一窗口。 +fn confirmed_exhaustion_projection( + mut quota: Map, + reset_at: Option, +) -> Map { + let mut rate_limit = quota + .remove("rate_limit") + .and_then(|value| value.as_object().cloned()) + .unwrap_or_default(); + rate_limit.insert("allowed".to_owned(), Value::Bool(false)); + rate_limit.insert("limit_reached".to_owned(), Value::Bool(true)); + for field in ["primary_window", "secondary_window"] { + let existed = rate_limit.contains_key(field); + let mut window = rate_limit + .remove(field) + .and_then(|value| value.as_object().cloned()) + .unwrap_or_default(); + if field == "primary_window" || existed { + window.insert("used_percent".to_owned(), Value::from(100_u64)); + if field == "primary_window" + && let Some(reset_at) = reset_at + { + window.insert("reset_at".to_owned(), Value::from(reset_at)); + } + rate_limit.insert(field.to_owned(), Value::Object(window)); + } + } + quota.insert("rate_limit".to_owned(), Value::Object(rate_limit)); + quota +} + +/// 选择应继续锁定的 reset：新鲜快照仍确认耗尽时可把锁推进到更晚窗口； +/// 否则保留已确认失败的旧窗口，避免 98/99% 快照反向解锁。 +fn confirmed_exhaustion_reset_at( + previous_reset_at: Option, + refreshed_fact: CodexQuotaFact, +) -> Option { + let refreshed_reset_at = refreshed_fact.resets_at().map(SystemTime::from); + let reset_at = match (previous_reset_at, refreshed_reset_at) { + (Some(previous), Some(refreshed)) if refreshed_fact.exhausted() && refreshed > previous => { + refreshed + } + (Some(previous), _) => previous, + (None, refreshed) => refreshed?, + }; + system_time_to_unix_seconds(reset_at) +} + +fn system_time_to_unix_seconds(value: SystemTime) -> Option { value - .get("metered_feature") - .or_else(|| value.get("limit_name")) - .and_then(Value::as_str) - .is_some_and(|source| { - source.trim().to_ascii_lowercase().replace(['-', ' '], "_") == "codex" - }) + .duration_since(UNIX_EPOCH) + .ok()? + .as_secs() + .try_into() + .ok() } fn passive_rate_limit_snapshot(details: &RateLimitDetails) -> Option> { diff --git a/backend/crates/providers/openai/src/credential/quota/recovery.rs b/backend/crates/providers/openai/src/credential/quota/recovery.rs index 224142504..d8b4625ea 100644 --- a/backend/crates/providers/openai/src/credential/quota/recovery.rs +++ b/backend/crates/providers/openai/src/credential/quota/recovery.rs @@ -1,7 +1,7 @@ -//! 402 恢复证据与 availability 转换。 +//! 确认额度耗尽后的恢复证据与 availability 转换。 //! -//! `QuotaExhausted -> Ready` 只接受同一 credential revision 的真实成功响应、 -//! 新鲜快照显式 `allowed=true` 且无 exhaustion、或已到期旧 reset 配合未耗尽新快照。 +//! `QuotaExhausted -> Ready` 只能由同一 credential revision 的权威 usage 快照证明： +//! 已确认的旧窗口 reset 已到期，且新快照的 reset 已推进到下一窗口。 use std::time::SystemTime; @@ -12,47 +12,24 @@ use serde_json::Value; use super::snapshot::CodexQuotaFact; use crate::transport::CodexClientError; -/// 402 恢复可接受的证据强度。 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum QuotaRecoveryEvidence { - /// 一次真实成功推理响应（最强证据）。 - SuccessfulResponse, - /// 周期 worker 刷新出的新快照（需旧 reset 已到期佐证）。 - RefreshedSnapshot, -} - -/// 只处理 `QuotaExhausted` 的恢复——429/触顶不再进入该状态，由 quota 数据 -/// 驱动调度排除（限流不改变账号可用性）。`Invalid/Expired/Banned` 等终态由 selector 的 -/// 成功响应恢复路径处理。 +/// 只处理 `QuotaExhausted` 的恢复。`allowed=true`、成功响应和百分比回落都不能单独 +/// 覆盖一次已确认的额度拒绝；它们在 reset 未推进时可能只是上游滞后的观测。 /// -/// `previous_reset_at` 是写入新快照前读取的旧确认 reset；`RefreshedSnapshot` -/// 恢复要求它已到期（窗口真实滚动后新快照的 reset 指向下一个未来周期， -/// 不能用它判定旧耗尽窗口是否已结束）。 +/// `previous_reset_at` 是写入新快照前读取的已确认耗尽窗口。恢复必须同时满足： +/// 旧窗口已经到期，且权威快照的 reset 指向比旧窗口更晚的新周期。 pub(crate) fn quota_success_state( current: AccountAvailability, fact: CodexQuotaFact, previous_reset_at: Option, now: SystemTime, - recovery_evidence: QuotaRecoveryEvidence, ) -> Option { if current != AccountAvailability::QuotaExhausted || fact.exhausted() { return None; } - // 新鲜快照显式 `allowed=true` 且无 exhaustion，直接恢复。 - if fact.explicit_allowed() { - return Some(AccountAvailability::Ready); - } - // 一次真实成功响应是最强证据；周期刷新需要旧确认 reset 已到期 - // （used_percent 是可滞后、会取整的观测值，其回落不能覆盖仍未到期的旧 reset）。 - match recovery_evidence { - QuotaRecoveryEvidence::SuccessfulResponse => Some(AccountAvailability::Ready), - QuotaRecoveryEvidence::RefreshedSnapshot - if previous_reset_at.is_some_and(|reset_at| reset_at <= now) => - { - Some(AccountAvailability::Ready) - } - QuotaRecoveryEvidence::RefreshedSnapshot => None, - } + let previous_reset_at = previous_reset_at?; + let refreshed_reset_at = fact.resets_at().map(SystemTime::from)?; + (previous_reset_at <= now && refreshed_reset_at > previous_reset_at) + .then_some(AccountAvailability::Ready) } pub(crate) fn quota_state_transition(error: &CodexClientError) -> Option { diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index 65af94e52..74073d67e 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -18,8 +18,6 @@ pub struct CodexQuotaFact { remaining_percent: Option, resets_at: Option>, exhausted: bool, - /// 上游显式 `allowed=true`（恢复证据：新鲜快照明确放行）。 - explicit_allowed: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -41,7 +39,7 @@ pub enum CodexQuotaWindowRole { pub struct CodexQuotaWindow { key: String, source: String, - /// 上游提供的限流名称（`metered_feature`/`limit_name`/`limitId`）；显示优先用它。 + /// 上游提供的限流桶名称（`metered_feature`/`limit_name`/`limitId`）；用于非核心桶的展示前缀。 limit_name: Option, kind: CodexQuotaWindowKind, role: CodexQuotaWindowRole, @@ -189,12 +187,6 @@ impl CodexQuotaFact { pub const fn exhausted(&self) -> bool { self.exhausted } - - /// 上游是否显式 `allowed=true`（新鲜快照的恢复证据）。 - #[must_use] - pub const fn explicit_allowed(&self) -> bool { - self.explicit_allowed - } } pub fn parse_codex_quota_usage(usage: &Value) -> Result { @@ -220,7 +212,6 @@ pub fn parse_codex_quota_usage(usage: &Value) -> Result { + key: String, source: String, limit_name: Option, rate_limit: &'a Value, @@ -346,9 +339,15 @@ fn canonical_rate_limits( object: &Map, ) -> Result>, CodexCredentialQuotaError> { let mut limits = Vec::new(); + let has_primary_codex_rate_limit = object + .get("rate_limit") + .is_some_and(|value| !value.is_null()); if let Some(rate_limit) = object.get("rate_limit") { + // 官方 Codex 将顶层 `rate_limit` 固定识别为 `codex`，而不把 + // `active_limit` 投影成另一个展示桶。后者只是当前活动套餐的元数据。 limits.push(CanonicalRateLimit { - source: "core".to_owned(), + key: "core".to_owned(), + source: "codex".to_owned(), limit_name: None, rate_limit, }); @@ -358,6 +357,7 @@ fn canonical_rate_limits( .filter(|value| !value.is_null()) { limits.push(CanonicalRateLimit { + key: "code-review".to_owned(), source: "code_review".to_owned(), limit_name: Some("code_review".to_owned()), rate_limit, @@ -379,23 +379,29 @@ fn canonical_rate_limits( .as_object() .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?; let (source, limit_name) = canonical_rate_limit_source(item, index); + // 部分上游响应会在 `additional_rate_limits` 中重复默认的 `codex` + // 桶。它没有独立的 limit id，不能在账号面板再次展示。 + if has_primary_codex_rate_limit && is_codex_limit_id(&source) { + continue; + } if let Some(rate_limit) = item.get("rate_limit") { - // additional 里的 codex 桶覆盖顶层 rate_limit（同源替换）。 - if let Some(existing) = limits.iter_mut().find(|existing| existing.source == source) { - existing.rate_limit = rate_limit; - existing.limit_name = limit_name; - } else { - limits.push(CanonicalRateLimit { - source, - limit_name, - rate_limit, - }); - } + // 非 Codex additional 桶保留为独立快照；额外用索引生成窗口键， + // 避免同名 additional 产生重复前端 key。 + limits.push(CanonicalRateLimit { + key: format!("additional-{index}-{source}"), + source, + limit_name, + rate_limit, + }); } } Ok(limits) } +fn is_codex_limit_id(value: &str) -> bool { + value.trim().eq_ignore_ascii_case("codex") +} + fn canonical_rate_limit_source( item: &Map, index: usize, @@ -417,14 +423,11 @@ fn canonical_rate_limit_source( !value.is_empty() && value.len() <= 128 && !value.chars().any(char::is_control) }) .map(str::to_owned); - if source.trim().to_ascii_lowercase().replace(['-', ' '], "_") == "codex" { - ("core".to_owned(), limit_name) - } else { - (source, limit_name) - } + (source, limit_name) } fn parse_rate_limit_windows( + key_source: &str, source: &str, limit_name: Option<&str>, value: &Value, @@ -482,7 +485,7 @@ fn parse_rate_limit_windows( || used_percent.is_some_and(|used| used >= 100.0); let kind = quota_window_kind(window_seconds); output.push(CodexQuotaWindow { - key: format!("{}-{}", quota_key(source), quota_role_name(role, kind)), + key: format!("{}-{}", quota_key(key_source), quota_role_name(role, kind)), source: source.to_owned(), limit_name: limit_name.map(str::to_owned), kind, @@ -559,7 +562,6 @@ const fn quota_role_name(role: CodexQuotaWindowRole, kind: CodexQuotaWindowKind) struct QuotaAggregate { recognized: bool, exhausted: bool, - explicit_allowed: bool, remaining_percent: Option, resets_at: Option>, } @@ -572,7 +574,6 @@ impl QuotaAggregate { self.recognized = true; self.exhausted |= optional_bool(object, "limit_reached")?.unwrap_or(false); self.exhausted |= optional_bool(object, "allowed")?.is_some_and(|allowed| !allowed); - self.explicit_allowed |= optional_bool(object, "allowed")?.is_some_and(|allowed| allowed); for key in ["primary_window", "secondary_window"] { if let Some(window) = object.get(key) { self.observe_window(window)?; diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 93134339f..32ac60175 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -345,49 +345,6 @@ impl CodexCredentialSelector { AccountCandidate { account, signals } }) .collect::>(); - // skip_exhausted 语义：true 时排除任一窗口触顶（limit_reached）的账号， - // 仅由上游 429 兜底；false 时保留全部候选（含 QuotaExhausted 的放宽投影）。 - // 保留原始账号供 revision-fenced 凭据加载，用影子候选承载资格放宽。 - let scheduling_candidates = if diagnostic { - // 连接测试必须实际探测指定账号；额度快照只影响常规调度。 - candidates.clone() - } else if self.skip_exhausted { - candidates - .iter() - .filter(|candidate| !candidate.signals.quota_limit_reached) - .cloned() - .collect::>() - } else { - candidates - .iter() - .cloned() - .map(|mut candidate| { - candidate.account = quota_exhausted_scheduling_projection(&candidate.account); - candidate - }) - .collect::>() - }; - let scheduling_candidates = scheduling_candidates.as_slice(); - let mut affinity = if diagnostic { - AffinitySelection::default() - } else { - self.resolve_session_affinity( - request.session_affinity_key, - &candidates, - SystemTime::now(), - ) - .await - }; - let cyber_policy_scope = self - .prepare_cyber_policy_scope(cyber_policy_session_key) - .await; - let mut excluded = request.attempt.excluded_accounts().clone(); - if let Some(state) = cyber_policy_scope - .as_ref() - .and_then(|scope| scope.state.as_ref()) - { - excluded.extend(state.excluded_accounts().iter().cloned()); - } let continuation_account = match request.attempt.continuation_attempt() { ContinuationAttempt::Native => request .attempt @@ -409,7 +366,74 @@ impl CodexCredentialSelector { { return Err(CredentialSelectionError::NoEligibleCredential); } - let pinned_account = required_account.or(continuation_account); + let pinned_account = required_account.or_else(|| continuation_account.clone()); + let preserve_native_continuation_state = matches!( + request.attempt.continuation_attempt(), + ContinuationAttempt::Native + ) && continuation_account.is_some(); + // skip_exhausted 语义：true 时排除任一窗口触顶（limit_reached）的账号， + // 仅由上游 429 兜底；false 时保留全部候选（含 QuotaExhausted 的放宽投影）。 + // 保留原始账号供 revision-fenced 凭据加载，用影子候选承载资格放宽。 + let scheduling_candidates = if diagnostic { + // 连接测试必须实际探测指定账号；额度快照只影响常规调度。 + candidates.clone() + } else if self.skip_exhausted { + candidates + .iter() + .filter(|candidate| !candidate.signals.quota_limit_reached) + .cloned() + .collect::>() + } else { + candidates + .iter() + .cloned() + .map(|mut candidate| { + // 已命中 previous_response_id 的 native continuation 先检查原账号。 + // 即使常规调度允许探测额度耗尽账号，也必须保留原账号真实状态， + // 由 Coordinator 据此切换到可携带重放，而非伪装为 Ready。 + if !preserve_native_continuation_state + || pinned_account.as_ref() != Some(candidate.account.id()) + { + candidate.account = + quota_exhausted_scheduling_projection(&candidate.account); + } + candidate + }) + .collect::>() + }; + let scheduling_candidates = scheduling_candidates.as_slice(); + let mut affinity = if diagnostic { + AffinitySelection::default() + } else { + self.resolve_session_affinity( + request.session_affinity_key, + &candidates, + SystemTime::now(), + ) + .await + }; + let cyber_policy_scope = self + .prepare_cyber_policy_scope(cyber_policy_session_key) + .await; + let mut excluded = request.attempt.excluded_accounts().clone(); + if !diagnostic + && preserve_native_continuation_state + && let Some(pinned) = pinned_account.as_ref() + && candidates.iter().any(|candidate| { + candidate.account.id() == pinned && candidate.signals.quota_limit_reached + }) + { + // `skip_exhausted = false` 只允许普通新请求探测可疑账号；已绑定的 + // native response handle 先将真实触顶状态交给 Coordinator，随后才可 + // 通过 ReplayAny 按当前调度策略换号。 + excluded.insert(pinned.clone()); + } + if let Some(state) = cyber_policy_scope + .as_ref() + .and_then(|scope| scope.state.as_ref()) + { + excluded.extend(state.excluded_accounts().iter().cloned()); + } if let Some(required) = pinned_account.as_ref() { for candidate in &candidates { if candidate.account.id() != required { diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index 25d36825c..f6956b31e 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -1702,6 +1702,8 @@ struct MappedProviderFailure { cyber_policy_failure: bool, set_cookie_headers: Vec, rate_limit_headers: Vec<(String, String)>, + /// 结构化 usage-limit 错误给出的窗口 reset（Unix 秒）。 + confirmed_exhaustion_reset_at: Option, observation: Option, capture_response_cookies: bool, } @@ -1715,6 +1717,7 @@ impl MappedProviderFailure { cyber_policy_failure: false, set_cookie_headers: Vec::new(), rate_limit_headers: Vec::new(), + confirmed_exhaustion_reset_at: None, observation: None, capture_response_cookies: false, } @@ -1872,6 +1875,18 @@ async fn apply_failure( failure.account_failure, Some(CodexAccountFailure::QuotaExhausted | CodexAccountFailure::UsageLimitExhausted { .. }) ); + if needs_authoritative_quota_refresh + && let Err(error) = context + .quota + .record_confirmed_exhaustion(account, failure.confirmed_exhaustion_reset_at) + .await + { + tracing::warn!( + account_id = %account.id(), + error = %error, + "Failed to project confirmed OpenAI quota exhaustion" + ); + } if failure.cyber_policy_failure { context .selector @@ -2169,6 +2184,7 @@ fn map_upstream_failure( cyber_policy_failure, set_cookie_headers: failure.set_cookie_headers, rate_limit_headers: failure.rate_limit_headers, + confirmed_exhaustion_reset_at: failure.usage_limit_resets_at, observation, capture_response_cookies: !matches!( category, diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 2ab3e43d6..891cf7068 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -87,8 +87,13 @@ impl CodexBackendClient { let headers = self.request_headers_for_http_response(upstream_request, context)?; let headers_started_at = Instant::now(); // 与官方 Codex app-server 一致：/v1/responses 请求体默认 zstd 压缩。 + // Codex 上游只交付 SSE；即使下游请求 `stream: false`，也要上游流式执行， + // 再由 API 层收集 canonical events 并返回完整 JSON。不能把下游的传输偏好 + // 直接透传给 Codex，否则上游会以 400 拒绝非流式请求。 + let mut upstream_body = upstream_request.body().clone(); + upstream_body.insert("stream".to_owned(), serde_json::Value::Bool(true)); let body = - serde_json::to_vec(&upstream_request).map_err(CodexClientError::RequestBodyEncode)?; + serde_json::to_vec(&upstream_body).map_err(CodexClientError::RequestBodyEncode)?; let body = zstd::stream::encode_all(std::io::Cursor::new(body), 3) .map_err(CodexClientError::RequestCompression)?; let response = self diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 19835e04f..8d2d6ce16 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/backend/crates/providers/openai/src/transport/diagnostics.rs @@ -3,6 +3,7 @@ use super::client::CodexClientVisibleUpstreamResponse; use super::protocol::responses::ResponsesSseFailure; use base64::{Engine as _, engine::general_purpose::STANDARD}; +use chrono::{DateTime, Utc}; use reqwest::{StatusCode, header::HeaderMap}; use serde_json::Value; @@ -113,6 +114,8 @@ pub struct CodexUpstreamFailure { pub(crate) client_error_type: Option, pub(crate) client_response: Option>, pub(crate) identity_error_code: Option, + /// 官方 `usage_limit_reached` 错误体携带的窗口重置时间（Unix 秒）。 + pub(crate) usage_limit_resets_at: Option, pub(crate) retry_after_seconds: Option, pub(crate) request_id: Option, pub(crate) set_cookie_headers: Vec, @@ -139,6 +142,7 @@ impl CodexUpstreamFailure { ) -> Self { let fields = ParsedUpstreamError::from_body(body); let category = classify_upstream_failure( + UpstreamFailureSource::HttpResponse, Some(status), body, &fields, @@ -153,6 +157,9 @@ impl CodexUpstreamFailure { client_error_type: fields.error_type, client_response: client_response.cloned().map(Box::new), identity_error_code: diagnostics.identity_error_code.clone(), + usage_limit_resets_at: (category == CodexFailureCategory::UsageLimitExhausted) + .then_some(fields.resets_at) + .flatten(), retry_after_seconds, request_id: diagnostics.request_id.clone(), set_cookie_headers: set_cookie_headers.to_vec(), @@ -174,11 +181,13 @@ impl CodexUpstreamFailure { error_type: failure.upstream_type.clone(), message: failure.message.clone(), client_message: Some(failure.message.clone()), + resets_at: None, }; let status = failure .explicit_status_code .and_then(|code| StatusCode::from_u16(code).ok()); let category = classify_upstream_failure( + UpstreamFailureSource::SseFailure, status, "", &fields, @@ -193,6 +202,7 @@ impl CodexUpstreamFailure { client_error_type: fields.error_type, client_response: None, identity_error_code: diagnostics.identity_error_code.clone(), + usage_limit_resets_at: None, retry_after_seconds: failure.retry_after_seconds, request_id: diagnostics.request_id.clone(), set_cookie_headers: set_cookie_headers.to_vec(), @@ -246,6 +256,7 @@ struct ParsedUpstreamError { error_type: Option, message: String, client_message: Option, + resets_at: Option, } impl ParsedUpstreamError { @@ -256,6 +267,7 @@ impl ParsedUpstreamError { error_type: None, message: body.to_owned(), client_message: None, + resets_at: None, }; }; let error = value @@ -282,15 +294,26 @@ impl ParsedUpstreamError { .clone() .or_else(|| error.as_str().and_then(non_empty_owned)) .unwrap_or_else(|| body.to_owned()); + let resets_at = error + .get("resets_at") + .and_then(Value::as_i64) + .filter(|seconds| DateTime::::from_timestamp(*seconds, 0).is_some()); Self { code, error_type, message, client_message, + resets_at, } } } +#[derive(Clone, Copy)] +enum UpstreamFailureSource { + HttpResponse, + SseFailure, +} + #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct CodexUpstreamDiagnostics { pub status_code: Option, @@ -363,6 +386,7 @@ fn decode_identity_error_code(encoded: &str) -> Option { } fn classify_upstream_failure( + source: UpstreamFailureSource, status: Option, body: &str, fields: &ParsedUpstreamError, @@ -376,6 +400,19 @@ fn classify_upstream_failure( let message = fields.message.to_ascii_lowercase(); let body = body.to_ascii_lowercase(); + // 与官方 Codex HTTP/WS 路径一致：429 只有结构化 + // `error.type=usage_limit_reached` 才能确认额度窗口耗尽；其余 429 都是临时限流。 + // SSE `response.failed` 的结构化字段形态不同，保留其 code/type 语义单独分类。 + if matches!(source, UpstreamFailureSource::HttpResponse) + && status == Some(StatusCode::TOO_MANY_REQUESTS) + { + return if error_type == "usage_limit_reached" { + CodexFailureCategory::UsageLimitExhausted + } else { + CodexFailureCategory::RateLimited + }; + } + if [code.as_str(), message.as_str(), body.as_str()] .into_iter() .any(is_model_unsupported) diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 3dcb9ee90..b88497485 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -606,6 +606,26 @@ impl CodexResponsesRequest { ) } + /// 返回会话亲和所用的子代理区分值。 + /// + /// Codex 原生请求在 turn metadata 中声明 `subagent_kind`；网关的 + /// `/responses/review` 等入口则通过 `client_metadata.x-openai-subagent` + /// 传递给上游。该可选值仅参与会话亲和键派生。 + pub fn subagent_kind(&self) -> Option { + self.semantics() + .subagent_kind + .filter(|value| !value.trim().is_empty()) + .or_else(|| { + self.client_metadata() + .and_then(Value::as_object) + .and_then(|metadata| metadata.get("x-openai-subagent")) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }) + } + /// 设置 / 合并 client metadata。 pub fn set_client_metadata(&mut self, client_metadata: Option) { match client_metadata { diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 21d5b6294..ea759b8e3 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -428,7 +428,8 @@ async fn openai_admin_provider_projects_cached_quota_models_and_canonical_export } #[tokio::test] -async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_quota() { +async fn openai_admin_provider_projects_official_codex_quota_and_independent_buckets_with_chinese_labels() + { let store = Arc::new(MemoryAccountStore::default()); store .seed_oauth_credential(ImportCodexOAuthCredential { @@ -444,11 +445,15 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu .account("acct_admin_canonical_quota") .expect("stored account"); let raw = json!({ + "active_limit": "premium", "rate_limit": { "primary_window": { "used_percent": 91, "reset_at": 1_900_000_000, "limit_window_seconds": 2_592_000 + }, + "secondary_window": { + "used_percent": 88 } }, "additional_rate_limits": [{ @@ -459,6 +464,19 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu "used_percent": 2, "reset_at": 1_900_000_000, "limit_window_seconds": 2_592_000 + }, + "secondary_window": { + "used_percent": 0 + } + } + }, { + "limit_name": "code_review", + "metered_feature": "code_review", + "rate_limit": { + "primary_window": { + "used_percent": 12, + "reset_at": 1_900_000_000, + "limit_window_seconds": 604_800 } } }] @@ -499,10 +517,31 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu .collect::>(); assert_eq!(monthly.len(), 1); - // limit_name 优先显示，不再猜成「月限额」。 - assert_eq!(monthly[0].label, "custom codex label"); - assert_eq!(monthly[0].source.as_deref(), Some("core")); - assert_eq!(monthly[0].used_percent, Some(2.0)); + assert!(monthly.iter().any(|window| { + window.label == "月额度" + && window.source.as_deref() == Some("codex") + && window.used_percent == Some(91.0) + })); + assert!( + !quota + .windows + .iter() + .any(|window| window.key.starts_with("additional-0-codex")), + "the additional codex alias should not become a second display bucket" + ); + let secondary = quota + .windows + .iter() + .find(|window| window.key == "core-secondary") + .expect("core secondary quota"); + assert_eq!(secondary.label, "次级额度"); + assert_eq!(secondary.used_percent, Some(88.0)); + let review = quota + .windows + .iter() + .find(|window| window.source.as_deref() == Some("code_review")) + .expect("code review quota"); + assert_eq!(review.label, "代码审查 · 周额度"); } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/credential/contract.rs b/backend/crates/providers/openai/tests/credential/contract.rs index 53e9fc360..dc5838b11 100644 --- a/backend/crates/providers/openai/tests/credential/contract.rs +++ b/backend/crates/providers/openai/tests/credential/contract.rs @@ -4,9 +4,13 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use futures::executor::block_on; +use gateway_core::engine::continuation::{ + ContinuationBinding, NativeContinuationPin, PreviousResponseId, +}; use gateway_core::engine::credential::{ AccountAttemptFeedback, AccountAvailability, AccountFeedbackStats, AccountSelectionPolicy, - AccountStateChange, ProviderAccountId, ProviderAccountStore as _, RotationStrategy, + AccountStateChange, OpaqueProviderData, ProviderAccountId, ProviderAccountStore as _, + QuotaObservation, QuotaWriteOutcome, RotationStrategy, }; use gateway_core::engine::{ AccountAttemptContext, AttemptContext, CancellationToken, ModelRequestId, RequestAttemptContext, @@ -24,6 +28,7 @@ use provider_openai::credential::{ }; use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState}; use secrecy::ExposeSecret; +use serde_json::json; use url::Url; use crate::support::{ @@ -985,6 +990,131 @@ fn selector_keeps_a_quota_exhausted_account_eligible_when_configured_not_to_skip ); } +#[test] +fn native_continuation_surfaces_the_original_accounts_quota_status_to_the_coordinator() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_original", "at-original"); + create_account(&store, "acct_fallback", "at-fallback"); + let strict_selector = selector(&store, Arc::new(TestLeaseCoordinator::default())); + let original = store.account("acct_original").expect("original account"); + block_on(strict_selector.record_failure(&original, CodexAccountFailure::QuotaExhausted, None)) + .expect("mark original account exhausted"); + + let selector = selector_with_runtime( + &store, + Arc::new(TestLeaseCoordinator::default()), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + false, + Arc::new(MemoryCooldownPort::new()), + ); + let continuation = NativeContinuationPin::new( + PreviousResponseId::new("previous-response"), + PreviousResponseId::new("upstream-response"), + ProviderKind::new("openai").expect("provider"), + original.id().clone(), + ); + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_native_continuation").expect("request id"), + ClientApiKeyId::new("key_codex_contract").expect("client key id"), + ), + NonZeroU32::new(1).expect("attempt"), + SystemTime::now() + Duration::from_secs(30), + account_policy(), + AccountAttemptContext::new(BTreeSet::new(), None, None), + Some(ContinuationBinding::Pinned(continuation)), + CancellationToken::new(), + ); + let error = + block_on( + selector.select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &Url::parse("https://chatgpt.com/backend-api/codex/responses") + .expect("request URL"), + attempt: &attempt, + session_affinity_key: None, + }), + ) + .expect_err("the selector must surface the unavailable native account"); + + assert!(matches!( + error, + CredentialSelectionError::NoEligibleCredential + )); +} + +#[test] +fn native_continuation_surfaces_the_original_accounts_quota_signal_to_the_coordinator() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_original_signal", "at-original-signal"); + create_account(&store, "acct_fallback_signal", "at-fallback-signal"); + let original = store + .account("acct_original_signal") + .expect("original account"); + let quota = json!({ + "rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": {"used_percent": 98} + } + }); + let outcome = block_on(store.compare_and_swap_quota(QuotaObservation { + account_id: original.id().clone(), + expected_revision: original.revision(), + quota: Some(OpaqueProviderData::new( + quota.as_object().expect("quota object").clone(), + )), + observed_at: Some(SystemTime::now()), + limit_reached: None, + })) + .expect("persist quota signal"); + assert!(matches!(outcome, QuotaWriteOutcome::Updated)); + + let selector = selector_with_runtime( + &store, + Arc::new(TestLeaseCoordinator::default()), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + false, + Arc::new(MemoryCooldownPort::new()), + ); + let continuation = NativeContinuationPin::new( + PreviousResponseId::new("previous-response"), + PreviousResponseId::new("upstream-response"), + ProviderKind::new("openai").expect("provider"), + original.id().clone(), + ); + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_native_continuation_signal").expect("request id"), + ClientApiKeyId::new("key_codex_contract").expect("client key id"), + ), + NonZeroU32::new(1).expect("attempt"), + SystemTime::now() + Duration::from_secs(30), + account_policy(), + AccountAttemptContext::new(BTreeSet::new(), None, None), + Some(ContinuationBinding::Pinned(continuation)), + CancellationToken::new(), + ); + let error = + block_on( + selector.select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &Url::parse("https://chatgpt.com/backend-api/codex/responses") + .expect("request URL"), + attempt: &attempt, + session_affinity_key: None, + }), + ) + .expect_err("the selector must surface the quota-limited native account"); + + assert!(matches!( + error, + CredentialSelectionError::NoEligibleCredential + )); +} + #[test] fn identity_verification_failure_isolates_only_selected_account() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index cf6e26860..baf786083 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -435,7 +435,7 @@ async fn passive_rate_limit_headers_replace_stale_secondary_window() { } #[tokio::test] -async fn passive_active_limit_replaces_stale_codex_alias() { +async fn passive_active_limit_keeps_the_top_level_codex_bucket_canonical() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_passive_active_limit").await; let account = store.account("acct_passive_active_limit").expect("account"); @@ -491,7 +491,12 @@ async fn passive_active_limit_replaces_stale_codex_alias() { .expect("quota snapshot"); assert_eq!(snapshot.windows().len(), 1); - assert_eq!(snapshot.windows()[0].used_percent(), Some(55.0)); + assert!( + snapshot + .windows() + .iter() + .any(|window| { window.source() == "codex" && window.used_percent() == Some(55.0) }) + ); } #[tokio::test] @@ -961,7 +966,7 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor .await .expect("lower percent quota recheck"); - assert_eq!(lower_percent.updated, 1); + assert_eq!(lower_percent.exhausted, 1); assert_eq!( store .account(account_id) @@ -970,6 +975,12 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor AccountAvailability::QuotaExhausted, "a provider percentage change is not sufficient to release a confirmed exhaustion" ); + let snapshot = next_cycle + .read_account(account.id()) + .await + .expect("read held exhaustion projection") + .expect("held exhaustion projection"); + assert_eq!(snapshot.fact().remaining_percent(), Some(0)); } #[tokio::test] @@ -978,6 +989,28 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { let account_id = "acct_periodic_reset_elapsed"; create_account(&store, account_id).await; let account = store.account(account_id).expect("created account"); + let elapsed_reset = Utc::now().timestamp() - 60; + store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: Some(OpaqueProviderData::new( + json!({ + "rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": {"used_percent": 100, "reset_at": elapsed_reset} + } + }) + .as_object() + .expect("recorded exhausted quota") + .clone(), + )), + observed_at: Some(SystemTime::now()), + limit_reached: Some(true), + }) + .await + .expect("record elapsed exhausted reset"); store .apply_state_change(AccountStateChange { message: None, @@ -989,14 +1022,14 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() { .await .expect("mark account exhausted with an elapsed reset"); let server = MockServer::start().await; - let elapsed_reset = Utc::now().timestamp() - 1; + let next_reset = Utc::now().timestamp() + 3600; 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": 1, "reset_at": elapsed_reset} + "primary_window": {"used_percent": 1, "reset_at": next_reset} } }))) .expect(1) @@ -1118,7 +1151,33 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() { .await .expect("mark account exhausted"); } + let good = store + .account("acct_quota_batch_good") + .expect("good account"); + let elapsed_reset = Utc::now().timestamp() - 60; + store + .compare_and_swap_quota(QuotaObservation { + account_id: good.id().clone(), + expected_revision: good.revision(), + quota: Some(OpaqueProviderData::new( + json!({ + "rate_limit": { + "allowed": false, + "limit_reached": true, + "primary_window": {"used_percent": 100, "reset_at": elapsed_reset} + } + }) + .as_object() + .expect("recorded exhausted quota") + .clone(), + )), + observed_at: Some(SystemTime::now()), + limit_reached: Some(true), + }) + .await + .expect("seed good account old reset"); let server = MockServer::start().await; + let next_reset = Utc::now().timestamp() + 3600; Mock::given(method("GET")) .and(path("/api/codex/usage")) .and(header("authorization", "Bearer token-acct_quota_batch_bad")) @@ -1137,7 +1196,7 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() { .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "rate_limit": { "allowed": true, - "primary_window": {"used_percent": 25.0, "reset_at": 1_900_000_000} + "primary_window": {"used_percent": 25.0, "reset_at": next_reset} } }))) .mount(&server) diff --git a/backend/crates/providers/openai/tests/credential/quota/snapshot.rs b/backend/crates/providers/openai/tests/credential/quota/snapshot.rs index 28ae196d1..a18a136e7 100644 --- a/backend/crates/providers/openai/tests/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/tests/credential/quota/snapshot.rs @@ -305,7 +305,7 @@ async fn scheduling_signal_survives_limit_reached_without_percent_or_reset() { } #[tokio::test] -async fn persisted_codex_additional_limit_replaces_the_top_level_rate_limit() { +async fn persisted_codex_alias_keeps_the_top_level_rate_limit_canonical() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_canonical_codex_limit").await; let account = store @@ -323,7 +323,7 @@ async fn persisted_codex_additional_limit_replaces_the_top_level_rate_limit() { "metered_feature": "codex", "rate_limit": { "primary_window": { - "used_percent": 2, + "used_percent": 98, "reset_at": 1_900_000_000, "limit_window_seconds": 2_592_000 } @@ -355,10 +355,13 @@ async fn persisted_codex_additional_limit_replaces_the_top_level_rate_limit() { .filter(|window| window.kind() == CodexQuotaWindowKind::Monthly) .collect::>(); - assert_eq!(snapshot.fact().remaining_percent(), Some(98)); + assert_eq!(snapshot.fact().remaining_percent(), Some(9)); assert_eq!(monthly.len(), 1); - assert_eq!(monthly[0].source(), "core"); - assert_eq!(monthly[0].used_percent(), Some(2.0)); + assert!( + monthly + .iter() + .any(|window| { window.source() == "codex" && window.used_percent() == Some(91.0) }) + ); } #[tokio::test] @@ -419,16 +422,21 @@ async fn code_review_limit_projects_as_independent_window_with_limit_name() { .expect("read quota") .expect("quota snapshot"); - // spend_control 不再生成窗口（只作 exhaustion 信号）， - // 顶层 code_review_rate_limit 与 additional 的 code_review 桶同源替换后保留一个。 + // spend_control 不再生成窗口（只作 exhaustion 信号）；顶层和 additional + // 的 code_review 桶分别保留，避免一个快照覆盖另一个。 let review = snapshot .windows() .iter() .filter(|window| window.source() == "code_review") .collect::>(); - assert_eq!(review.len(), 1); - assert_eq!(review[0].limit_name(), Some("code review")); - assert_eq!(review[0].used_percent(), Some(55.0)); - assert_eq!(snapshot.windows().len(), 2); + assert_eq!(review.len(), 2); + assert!(review.iter().any(|window| { + window.limit_name() == Some("code_review") && window.used_percent() == Some(80.0) + })); + assert!(review.iter().any(|window| { + window.limit_name() == Some("code review") && window.used_percent() == Some(55.0) + })); + assert_ne!(review[0].key(), review[1].key()); + assert_eq!(snapshot.windows().len(), 3); assert!(!snapshot.fact().exhausted()); } diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 03ef8a4b3..92d4f9c03 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -1071,6 +1071,66 @@ async fn prompt_cache_key_should_become_an_opaque_session_affinity_lookup_key() assert!(keys[0].bytes().all(|byte| byte.is_ascii_hexdigit())); } +#[tokio::test] +async fn subagent_requests_should_use_an_affinity_key_isolated_from_the_root_session() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .mount(&server) + .await; + let provider = provider_with_affinity_and_base_url(&store, Arc::clone(&affinity), server.uri()); + + for (request_id, subagent_kind) in [ + ("req_root_affinity", None), + ("req_subagent_affinity_first", Some("review")), + ("req_subagent_affinity_second", Some("review")), + ] { + let mut body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!("new task")), + ("prompt_cache_key".to_owned(), json!("root-session-key")), + ]); + if let Some(subagent_kind) = subagent_kind { + body.insert( + "client_metadata".to_owned(), + json!({"x-openai-subagent": subagent_kind}), + ); + } + let generation = GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body) + .expect("OpenAI payload") + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + ); + let mut stream = provider + .execute( + planned_request("openai", Operation::Generate(generation)), + context(request_id, CancellationToken::new()), + ) + .await + .expect("prepare subagent provider stream"); + while let Some(event) = stream.next().await { + event.expect("subagent response"); + } + drop(stream); + } + + let keys = affinity.lookup_keys(); + assert_eq!(keys.len(), 3); + assert_ne!(keys[0], keys[1], "subagent must not inherit the root key"); + assert_eq!( + keys[1], keys[2], + "the same subagent should retain its own key" + ); +} + #[tokio::test] async fn explicit_session_id_should_override_turn_specific_prompt_cache_keys_for_affinity() { let store = Arc::new(MemoryAccountStore::default()); @@ -1289,9 +1349,10 @@ async fn response_failed_before_semantic_output_is_atomic_and_persists_quota_loc } #[tokio::test] -async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_quota_snapshot() { +async fn official_usage_limit_failure_projects_100_percent_and_holds_until_reset_advances() { let store = Arc::new(MemoryAccountStore::default()); let account_id = "acct_usage_limit_request_path"; + let reset_at = 1_900_000_000; create_account(&store, account_id).await; let account = store.account(account_id).expect("created account"); store @@ -1303,7 +1364,7 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu "rate_limit": { "allowed": true, "limit_reached": false, - "primary_window": {"used_percent": 99, "reset_at": 1_900_000_000} + "primary_window": {"used_percent": 99, "reset_at": reset_at} } }) .as_object() @@ -1319,8 +1380,8 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu Mock::given(method("GET")) .and(path("/api/codex/usage")) .and(header("authorization", format!("Bearer at-{account_id}"))) - // 生产环境会短暂返回互相矛盾的快照：数字已到 100%，但布尔位仍声称可用。 - // 后台同步应覆盖展示数据，不能据此撤销真实请求已经确认的额度耗尽。 + // 生产环境会短暂返回互相矛盾的快照：98/99% 且布尔位仍声称可用。 + // 后台同步不能覆盖真实请求已确认的额度耗尽。 .respond_with( ResponseTemplate::new(200) // 验证后台 usage 同步不能把原始的额度错误响应拖到查询完成之后。 @@ -1329,7 +1390,7 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu "rate_limit": { "allowed": true, "limit_reached": false, - "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000} + "primary_window": {"used_percent": 99, "reset_at": reset_at} } })), ) @@ -1338,16 +1399,13 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu .await; Mock::given(method("POST")) .and(path("/codex/responses")) - .respond_with( - ResponseTemplate::new(200) - .insert_header("content-type", "text/event-stream") - .set_body_string(concat!( - "event: response.created\n", - "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_usage_limit_confirm\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n", - "event: response.failed\n", - "data: {\"type\":\"response.failed\",\"status_code\":429,\"retry_after_seconds\":86400,\"response\":{\"id\":\"resp_usage_limit_confirm\",\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"message\":\"usage limit reached\"}}}\n\n" - )), - ) + .respond_with(ResponseTemplate::new(429).set_body_json(json!({ + "error": { + "type": "usage_limit_reached", + "message": "usage limit reached", + "resets_at": reset_at + } + }))) .mount(&server) .await; @@ -1373,6 +1431,24 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted); let account = store.account(account_id).expect("usage-limit account"); assert_eq!(account.availability(), AccountAvailability::QuotaExhausted); + let projected = store + .get_quotas(&[account.id().clone()]) + .await + .expect("read immediate quota projection") + .into_iter() + .next() + .expect("confirmed quota observation"); + let projected_observed_at = projected.observed_at.expect("projection observed at"); + let projected = projected + .quota + .map(OpaqueProviderData::into_inner) + .expect("confirmed quota projection"); + assert_eq!( + Value::Object(projected) + .pointer("/rate_limit/primary_window/used_percent") + .and_then(Value::as_u64), + Some(100) + ); tokio::time::sleep(Duration::from_millis(200)).await; let requests = server.received_requests().await.expect("received requests"); assert_eq!( @@ -1384,35 +1460,45 @@ async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_qu "usage refresh must wait for the upstream quota settlement delay" ); - let quota = timeout(Duration::from_secs(5), async { + timeout(Duration::from_secs(5), async { loop { - let quota = store + let observed_at = store .get_quotas(&[account.id().clone()]) .await - .expect("read refreshed quota") + .expect("read refreshed quota observation") .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; + .and_then(|observation| observation.observed_at); + if observed_at.is_some_and(|observed_at| observed_at > projected_observed_at) { + break; } tokio::time::sleep(Duration::from_millis(10)).await; } }) .await - .expect("background usage refresh must replace the stale quota snapshot"); + .expect("background usage refresh must run"); + let quota = store + .get_quotas(&[account.id().clone()]) + .await + .expect("read refreshed quota") + .into_iter() + .next() + .and_then(|observation| observation.quota) + .map(OpaqueProviderData::into_inner) + .map(Value::Object) + .expect("authoritative quota projection"); assert_eq!( quota .pointer("/rate_limit/primary_window/used_percent") .and_then(Value::as_u64), Some(100) ); + assert_eq!( + quota + .pointer("/rate_limit/primary_window/reset_at") + .and_then(Value::as_i64), + Some(reset_at) + ); let requests = server.received_requests().await.expect("received requests"); assert_eq!( requests diff --git a/backend/crates/providers/openai/tests/transport/diagnostics.rs b/backend/crates/providers/openai/tests/transport/diagnostics.rs index 1b95fe122..6b2b44660 100644 --- a/backend/crates/providers/openai/tests/transport/diagnostics.rs +++ b/backend/crates/providers/openai/tests/transport/diagnostics.rs @@ -146,13 +146,13 @@ fn structured_expiry_code_classifies_without_an_auth_status() { } #[test] -fn structured_quota_signal_takes_precedence_over_http_429() { +fn http_429_without_usage_limit_type_remains_a_temporary_rate_limit() { assert_eq!( classify( 429, r#"{"error":{"code":"insufficient_quota","message":"rate limit reached"}}"# ), - CodexFailureCategory::QuotaExhausted + CodexFailureCategory::RateLimited ); } @@ -168,13 +168,24 @@ fn structured_usage_limit_type_is_resettable_quota_exhaustion() { } #[test] -fn usage_limit_message_without_a_type_is_resettable_quota_exhaustion() { +fn http_429_usage_limit_message_without_a_type_remains_a_temporary_rate_limit() { assert_eq!( classify( 429, r#"{"error":{"message":"The usage limit has been reached"}}"# ), - CodexFailureCategory::UsageLimitExhausted + CodexFailureCategory::RateLimited + ); +} + +#[test] +fn http_429_usage_limit_code_without_a_type_remains_a_temporary_rate_limit() { + assert_eq!( + classify( + 429, + r#"{"error":{"code":"usage_limit_reached","message":"The usage limit has been reached"}}"# + ), + CodexFailureCategory::RateLimited ); } diff --git a/backend/crates/providers/openai/tests/transport/headers.rs b/backend/crates/providers/openai/tests/transport/headers.rs index eb16e82e2..f43a8ee0f 100644 --- a/backend/crates/providers/openai/tests/transport/headers.rs +++ b/backend/crates/providers/openai/tests/transport/headers.rs @@ -789,6 +789,70 @@ async fn backend_http_should_zstd_compress_codex_responses_request_body() { assert_eq!(json["instructions"], "compress me"); } +#[tokio::test] +async fn backend_http_should_force_upstream_sse_for_non_streaming_client_request() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind non-streaming HTTP server"); + let address = listener + .local_addr() + .expect("non-streaming HTTP server address"); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept HTTP client"); + let request = read_http_request_with_body(&mut stream).await; + write_completed_sse_response(&mut stream).await; + request + }); + let mut request = CodexResponsesRequest::from_body( + json!({ + "model": "gpt-test", + "instructions": "collect this response", + "input": [], + "stream": false + }) + .as_object() + .expect("request body") + .clone(), + ); + request.force_http_sse = true; + assert!( + !request.stream(), + "client-facing preference remains non-streaming" + ); + let client = CodexBackendClient::new( + reqwest::Client::builder() + .no_proxy() + .build() + .expect("HTTP client"), + format!("http://{address}"), + test_wire_profile(), + ); + + client + .create_response( + &request, + request_context("req_non_streaming_client", Some("acct-non-streaming")), + ) + .await + .expect("upstream SSE response"); + + let raw = server.await.expect("non-streaming HTTP server task"); + let separator = raw + .windows(4) + .position(|window| window == b"\r\n\r\n") + .expect("HTTP head/body separator"); + let compressed_body = &raw[separator + 4..]; + let decompressed = zstd::stream::decode_all(std::io::Cursor::new(compressed_body)) + .expect("zstd body should decode"); + let json: serde_json::Value = + serde_json::from_slice(&decompressed).expect("decompressed body should be valid JSON"); + assert_eq!(json["stream"], true); + assert!( + !request.stream(), + "upstream normalization must not mutate local metadata" + ); +} + #[tokio::test] async fn websocket_should_keep_an_exact_chain_while_new_connections_adopt_the_latest_codex_core_wire_profile() { diff --git a/backend/crates/providers/openai/tests/transport/request.rs b/backend/crates/providers/openai/tests/transport/request.rs index 60564b0d1..d858368cd 100644 --- a/backend/crates/providers/openai/tests/transport/request.rs +++ b/backend/crates/providers/openai/tests/transport/request.rs @@ -401,6 +401,41 @@ fn observability_semantics_should_reuse_codex_turn_metadata() { assert!(semantics.compact); } +#[test] +fn encoder_should_extract_subagent_kind_from_wire_or_turn_metadata() { + let metadata_request = request(Map::from_iter([ + ("model".to_owned(), json!("client-model")), + ("input".to_owned(), json!("prompt")), + ( + "client_metadata".to_owned(), + json!({"x-openai-subagent": "review"}), + ), + ])); + let metadata_encoded = + encode_generate_request(&metadata_request, "gpt-test").expect("encode metadata request"); + assert_eq!(metadata_encoded.subagent_kind().as_deref(), Some("review")); + + let turn_metadata_payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("client-model")), + ("input".to_owned(), json!("prompt")), + ]), + ) + .expect("OpenAI payload") + .with_context(Map::from_iter([( + "turn_metadata".to_owned(), + Value::String(r#"{"subagent_kind":"worker"}"#.to_owned()), + )])); + let turn_metadata_request = GenerateRequest::from_protocol_payload(turn_metadata_payload); + let turn_metadata_encoded = encode_generate_request(&turn_metadata_request, "gpt-test") + .expect("encode turn metadata request"); + assert_eq!( + turn_metadata_encoded.subagent_kind().as_deref(), + Some("worker") + ); +} + #[test] fn observability_semantics_should_use_the_transparent_openai_payload() { let payload = ProtocolPayload::json_object( diff --git a/frontend/src/components/base/BasePopover.vue b/frontend/src/components/base/BasePopover.vue index 261bb13e5..c6f309158 100644 --- a/frontend/src/components/base/BasePopover.vue +++ b/frontend/src/components/base/BasePopover.vue @@ -6,7 +6,7 @@ import { computed, nextTick, onBeforeUnmount, ref, shallowRef, useAttrs, watch } type PopoverPlacement = 'top' | 'top-start' | 'top-end' | 'right' | 'bottom' | 'bottom-start' | 'bottom-end' | 'left' -type PopoverTrigger = 'click' | 'hover' +type PopoverTrigger = 'click' | 'hover' | 'hover-click' interface PopoverPoint { left: number @@ -29,7 +29,7 @@ const props = withDefaults( offset?: number width?: number | string disabled?: boolean - /** hover 触发时打开弹窗的延迟（毫秒）；click 触发不生效。 */ + /** 包含 hover 的触发方式打开弹窗的延迟（毫秒）；click 触发不生效。 */ hoverDelay?: number panelClass?: string triggerClass?: string @@ -277,7 +277,7 @@ function closePopover() { } function togglePopover() { - if (props.trigger !== 'click') { + if (!supportsClickTrigger()) { return } @@ -289,6 +289,14 @@ function togglePopover() { void openPopover() } +function supportsClickTrigger() { + return props.trigger === 'click' || props.trigger === 'hover-click' +} + +function supportsHoverTrigger() { + return props.trigger === 'hover' || props.trigger === 'hover-click' +} + function clearHoverCloseTimer() { if (hoverCloseTimer.value === undefined) { return @@ -308,7 +316,7 @@ function clearHoverOpenTimer() { } function handleHoverEnter() { - if (props.trigger !== 'hover') { + if (!supportsHoverTrigger()) { return } @@ -323,7 +331,7 @@ function handleHoverEnter() { } function handleHoverLeave() { - if (props.trigger !== 'hover') { + if (!supportsHoverTrigger()) { return } diff --git a/frontend/src/views/accounts/components/AccountStatusBadge.vue b/frontend/src/views/accounts/components/AccountStatusBadge.vue index 5617269f5..36779b71e 100644 --- a/frontend/src/views/accounts/components/AccountStatusBadge.vue +++ b/frontend/src/views/accounts/components/AccountStatusBadge.vue @@ -3,16 +3,14 @@ import { AlertTriangle, CircleCheck, Gauge, Power, Timer } from '@lucide/vue' import { computed } from 'vue' import BasePopover from '@/components/base/BasePopover.vue' +import BaseScrollbar from '@/components/base/BaseScrollbar.vue' import { errorReasonLabels, statusLabels, statusTones } from '../constants' const props = withDefaults( defineProps<{ status: string - /** `status === 'error'` 时的具体原因（对应后端 `errorReason`）。 */ errorReason?: string | null - /** 最近一次失败的上游错误原文（对应后端 `errorMessage`），作为辅助证据展示。 */ errorMessage?: string | null - // 429 临时限流到期时间（对应后端 `quota.rateLimitedUntil`）；`status === 'rate_limited'` 时有值。 rateLimitedUntil?: string | null variant?: 'inline' | 'pill' }>(), @@ -24,25 +22,58 @@ const props = withDefaults( }, ) -const tone = computed(() => statusTones[props.status]) +const statusStyles = { + success: { + text: 'text-(--cp-success-text)', + dot: 'bg-(--cp-success)', + badge: 'border-(--cp-success-border) bg-(--cp-success-bg) text-(--cp-success-text)', + icon: 'bg-(--cp-success-bg) text-(--cp-success-text) shadow-[inset_0_0_0_1px_var(--cp-success-border)]', + }, + danger: { + text: 'text-(--cp-danger-text)', + dot: 'bg-(--cp-danger)', + badge: 'border-(--cp-danger-border) bg-(--cp-danger-bg) text-(--cp-danger-text)', + icon: 'bg-(--cp-danger-bg) text-(--cp-danger-text) shadow-[inset_0_0_0_1px_var(--cp-danger-border)]', + }, + warning: { + text: 'text-(--cp-warning-text)', + dot: 'bg-(--cp-warning)', + badge: 'border-(--cp-warning-border) bg-(--cp-warning-bg) text-(--cp-warning-text)', + icon: 'bg-(--cp-warning-bg) text-(--cp-warning-text) shadow-[inset_0_0_0_1px_var(--cp-warning-border)]', + }, + info: { + text: 'text-(--cp-info-text)', + dot: 'bg-(--cp-info)', + badge: 'border-(--cp-info-border) bg-(--cp-info-bg) text-(--cp-info-text)', + icon: 'bg-(--cp-info-bg) text-(--cp-info-text) shadow-[inset_0_0_0_1px_var(--cp-info-border)]', + }, + normal: { + text: 'text-(--cp-text-secondary)', + dot: 'bg-(--cp-text-muted)', + badge: 'border-(--cp-divider-subtle) bg-(--cp-bg-subtle) text-(--cp-text-secondary)', + icon: 'bg-(--cp-bg-subtle) text-(--cp-text-secondary)', + }, +} as const + +const tone = computed(() => statusTones[props.status] || 'normal') +const statusStyle = computed(() => statusStyles[tone.value]) const label = computed(() => statusLabels[props.status] || props.status) -/** 分类原因（受控枚举映射文案）；`status === 'error'` 时有值。 */ -const reasonLabel = computed(() => { - if (props.errorReason) - return errorReasonLabels[props.errorReason] || props.errorReason - return null -}) -/** 上游错误原文（辅助证据）；有则显、无则隐。 */ +const reasonLabel = computed(() => + props.errorReason + ? errorReasonLabels[props.errorReason] || props.errorReason + : null, +) const errorText = computed(() => props.errorMessage || null) -// 429 临时限流中且带到期时间。 -const hasRateLimit = computed(() => +const detailTitle = computed(() => reasonLabel.value || label.value) +const isRateLimited = computed(() => props.status === 'rate_limited' && Boolean(props.rateLimitedUntil), ) -/** 限流到期相对时间：如「剩余 12 分钟」。到点返回 null。 */ const rateLimitedRelative = computed(() => { - if (!hasRateLimit.value) + const rateLimitedUntil = props.rateLimitedUntil + if (!rateLimitedUntil) return null - const until = new Date(props.rateLimitedUntil!.replace(' ', 'T')).getTime() + + const until = new Date(rateLimitedUntil.replace(' ', 'T')).getTime() const diffMinutes = Math.round((until - Date.now()) / 60_000) if (Number.isNaN(diffMinutes) || diffMinutes < 1) return null @@ -54,49 +85,39 @@ const rateLimitedRelative = computed(() => { return `剩余 ${hours} 小时` return `剩余 ${hours} 小时 ${mins} 分` }) -// `error` 状态始终显示弹窗（即使 reason/message 为空也展示状态卡片）， -// `rate_limited` 需要带到期时间才显示；其余状态不弹窗。 const hasDetail = computed(() => - props.status === 'error' || hasRateLimit.value, -) -/** 弹窗明细区是否渲染（错误原因 / 限流到期 / 错误反馈至少一项）。 */ -const hasDetailRows = computed(() => - hasRateLimit.value || Boolean(reasonLabel.value) || Boolean(errorText.value), + props.status === 'error' || isRateLimited.value, ) -const textClass = computed(() => { - if (tone.value === 'success') { - return 'text-(--cp-success-text)' - } - if (tone.value === 'danger') { - return 'text-(--cp-danger-text)' - } - if (tone.value === 'warning') { - return 'text-(--cp-warning-text)' - } - if (tone.value === 'info') { - return 'text-(--cp-info-text)' - } - return 'text-(--cp-text-secondary)' +const detailDescription = computed(() => { + if (props.status === 'rate_limited') + return '上游暂时限制了请求频率，系统正在冷却该账号。' + if (props.status === 'error') + return '该账号暂不参与调度，处理凭据后可重新测试连接。' + return '该账号当前状态需要关注。' }) -const dotClass = computed(() => { - if (tone.value === 'success') { - return 'bg-(--cp-success)' +const recoveryHint = computed(() => { + if (props.status === 'rate_limited') + return '冷却结束后，系统会自动恢复调度。' + + switch (props.errorReason) { + case 'expired': + case 'token_expired': + return '重新授权后，账号会重新参与调度。' + case 'invalid': + return '请更新或重新导入凭据，再次测试连接。' + case 'banned': + return '请确认上游账号状态，解除限制后再启用。' + default: + return '重新测试连接以获取最新状态。' } - if (tone.value === 'danger') { - return 'bg-(--cp-danger)' - } - if (tone.value === 'warning') { - return 'bg-(--cp-warning)' - } - if (tone.value === 'info') { - return 'bg-(--cp-info)' - } - return 'bg-(--cp-text-muted)' }) -/** 状态卡片左侧图标块：状态色 soft-bg + 语义图标。 */ +const triggerLabel = computed(() => + `${detailTitle.value}。${detailDescription.value} 点击或聚焦查看详情。`, +) + const cardIcon = computed(() => { switch (props.status) { case 'error': return AlertTriangle @@ -106,88 +127,107 @@ const cardIcon = computed(() => { default: return CircleCheck } }) -const iconBlockClass = computed(() => { - if (tone.value === 'success') { - return 'bg-(--cp-success-bg) text-(--cp-success-text)' - } - if (tone.value === 'danger') { - return 'bg-(--cp-danger-bg) text-(--cp-danger-text)' - } - if (tone.value === 'warning') { - return 'bg-(--cp-warning-bg) text-(--cp-warning-text)' - } - if (tone.value === 'info') { - return 'bg-(--cp-info-bg) text-(--cp-info-text)' - } - return 'bg-(--cp-bg-subtle) text-(--cp-text-secondary)' -}) diff --git a/frontend/src/views/accounts/constants.ts b/frontend/src/views/accounts/constants.ts index 9caa34512..6102c737f 100644 --- a/frontend/src/views/accounts/constants.ts +++ b/frontend/src/views/accounts/constants.ts @@ -19,14 +19,9 @@ export interface AccountLocalUsage { requestBuckets?: AccountRequestBucket[] } -const summaryGroupOrder = new Map([ +const quotaGroupOrder = new Map([ ['shortTerm', 0], ['monthly', 1], -]) - -const panelGroupOrder = new Map([ - ['monthly', 0], - ['shortTerm', 1], ['other', 2], ]) @@ -166,15 +161,13 @@ export function derivedAccountStatus(row: AccountRow): string { export function visibleSummaryQuotaWindows(windows: AccountQuotaWindow[]) { const known = [...windows] - .filter(window => summaryGroupOrder.has(window.group)) - .sort((left, right) => groupOrder(left, summaryGroupOrder) - groupOrder(right, summaryGroupOrder)) - return known.length > 0 ? known : windows + .filter(window => window.group !== 'other') + .sort(compareQuotaWindows) + return known.length > 0 ? known : [...windows].sort(compareQuotaWindows) } export function orderedPanelQuotaWindows(windows: AccountQuotaWindow[]) { - return [...windows].sort( - (left, right) => groupOrder(left, panelGroupOrder) - groupOrder(right, panelGroupOrder), - ) + return [...windows].sort(compareQuotaWindows) } function quotaWindowPercent(window: AccountQuotaWindow) { @@ -209,8 +202,21 @@ export function quotaWindowPercentTextClass(window: AccountQuotaWindow) { return 'text-(--cp-success-text)' } -function groupOrder(window: AccountQuotaWindow, order: Map) { - return order.get(window.group) ?? order.size +function compareQuotaWindows(left: AccountQuotaWindow, right: AccountQuotaWindow) { + const groupDifference = groupOrder(left) - groupOrder(right) + if (groupDifference !== 0) + return groupDifference + + const secondsDifference = (left.windowSeconds ?? Number.MAX_SAFE_INTEGER) + - (right.windowSeconds ?? Number.MAX_SAFE_INTEGER) + if (secondsDifference !== 0) + return secondsDifference + + return left.key.localeCompare(right.key) +} + +function groupOrder(window: AccountQuotaWindow) { + return quotaGroupOrder.get(window.group) ?? quotaGroupOrder.size } function accountProviderLabel(value?: string | null) { diff --git a/frontend/src/views/settings/components/backup/BackupRecordsCard.vue b/frontend/src/views/settings/components/backup/BackupRecordsCard.vue index f247aeece..841883a3e 100644 --- a/frontend/src/views/settings/components/backup/BackupRecordsCard.vue +++ b/frontend/src/views/settings/components/backup/BackupRecordsCard.vue @@ -133,7 +133,7 @@ function canDelete(record: BackupRecord): boolean { @page-size-change="emit('pageSizeChange', $event)" > diff --git a/frontend/src/views/usage/components/UsageRecordDetailModal.vue b/frontend/src/views/usage/components/UsageRecordDetailModal.vue index 803eb7b4d..d560e908b 100644 --- a/frontend/src/views/usage/components/UsageRecordDetailModal.vue +++ b/frontend/src/views/usage/components/UsageRecordDetailModal.vue @@ -1,6 +1,7 @@