From be0715e96bb586e1a7a63db8887794050fe6ef55 Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:45:59 +0800 Subject: [PATCH] fix: preserve model metadata and terminal state, improve quota reset (#224) Preserve versioned API key model metadata and terminal request state across clock rollback. Close reset credit dialogs after confirmed consumption and refresh account quota in place. Fixes #217 Fixes #216 Co-authored-by: Codex --- .../gateway-store/src/postgres/execution.rs | 3 +- .../provider_accounts/admin_adapter.rs | 2 +- .../provider_accounts/core_adapter.rs | 2 +- .../postgres/provider_accounts/repository.rs | 18 +- .../gateway-store/tests/postgres/execution.rs | 109 +++++++ .../tests/postgres/provider_accounts/mod.rs | 1 + .../postgres/provider_accounts/timestamps.rs | 261 +++++++++++++++++ .../openai/src/credential/catalog.rs | 224 ++++++++------- .../providers/openai/src/transport/catalog.rs | 26 +- .../openai/src/transport/client_sse.rs | 11 +- .../openai/tests/credential/catalog.rs | 271 ++++++++++++++++++ .../openai/tests/provider/contract.rs | 58 ++-- docs/api.md | 10 +- docs/architecture.md | 7 +- .../AccountQuotaPanel/ResetCredits.vue | 10 +- .../components/AccountQuotaPanel/index.vue | 4 +- .../composables/useAccountMutations.ts | 14 + .../composables/useAccountResetCredits.ts | 30 +- frontend/src/views/accounts/index.vue | 3 +- 19 files changed, 892 insertions(+), 172 deletions(-) create mode 100644 backend/crates/gateway-store/tests/postgres/provider_accounts/timestamps.rs diff --git a/backend/crates/gateway-store/src/postgres/execution.rs b/backend/crates/gateway-store/src/postgres/execution.rs index cbba02fde..09725f420 100644 --- a/backend/crates/gateway-store/src/postgres/execution.rs +++ b/backend/crates/gateway-store/src/postgres/execution.rs @@ -717,6 +717,7 @@ impl ModelRequestRepository for PgExecutionStore { finalization: ModelRequestFinalization, ) -> StoreResult { finalization.validate()?; + // 墙上时间可能回拨；终态必须可落盘，实际耗时仍保留 Core 的单调时钟观测。 let finalized = sqlx::query_scalar::<_, i64>( "with finalized as ( update model_requests @@ -734,7 +735,7 @@ impl ModelRequestRepository for PgExecutionStore { transport_decision_wait_ms = $27, connect_ms = $28, headers_ms = $29, first_event_ms = $30, first_reasoning_ms = $31, first_text_ms = $32, first_token_ms = $33, provider_processing_ms = $34, - latency_ms = $35, completed_at = $36, + latency_ms = $35, completed_at = greatest($36, started_at), upstream_transport = coalesce($37, upstream_transport), http_version = coalesce($38, http_version), websocket_pool = $39, service_tier = $40, provider_observation_json = $41, diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs index c20fdd77c..ac7b695a8 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs @@ -754,7 +754,7 @@ impl AccountStore for PgAdminAccountStore { ).fetch_one(&mut *transaction).await .map_err(|_| postgres_unavailable("lock default concurrency"))?; let changed = sqlx::query_scalar::<_, String>( - "update provider_accounts set concurrency_limit = $2, updated_at = now() + "update provider_accounts set concurrency_limit = $2, updated_at = greatest(now(), updated_at) where id = $1 and enabled = true and coalesce(concurrency_limit, $3) > $2 returning id", ) diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/core_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/core_adapter.rs index 457f530c9..86058d32f 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/core_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/core_adapter.rs @@ -198,7 +198,7 @@ impl ProviderAccountStore for PgProviderAccountRepository { last_error_message = case when $10::text is not null and enabled then $13 else last_error_message end, - updated_at = greatest(now(), coalesce($11, now())) + updated_at = greatest(now(), updated_at, $11) where id = $1 and credential_revision = $2 returning credential_revision", ) diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs index 7e5a33f84..d264a9696 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs @@ -192,7 +192,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { require_nonempty(ENTITY, "name", &account.name)?; let result = sqlx::query( "update provider_accounts - set name = $2, email = $3, plan_type = $4, updated_at = now() + set name = $2, email = $3, plan_type = $4, updated_at = greatest(now(), updated_at) where id = $1", ) .bind(account.id) @@ -225,7 +225,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { has_refresh_token = $4, access_token_expires_at = $5, next_refresh_at = $6, - updated_at = now() + updated_at = greatest(now(), updated_at) where id = $1 and credential_revision = $2 returning credential_revision", ) @@ -269,7 +269,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { when enabled then $6 else last_error_message end, - updated_at = case when enabled then greatest(now(), $4) else updated_at end + updated_at = case when enabled then greatest(now(), updated_at, $4) else updated_at end where id = $1 and credential_revision = $2 and (credential_observed_at is null or credential_observed_at <= $4)", ) @@ -288,7 +288,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { async fn set_provider_account_enabled(&self, id: &str, enabled: bool) -> StoreResult { require_nonempty(ENTITY, "id", id)?; let result = sqlx::query( - "update provider_accounts set enabled = $2, updated_at = now() where id = $1", + "update provider_accounts set enabled = $2, updated_at = greatest(now(), updated_at) where id = $1", ) .bind(id) .bind(enabled) @@ -330,7 +330,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { when $7::timestamptz is not null and (quota_access_observed_at is null or quota_access_observed_at <= $7) then $8 else quota_reset_at end, - updated_at = greatest(now(), $4) + updated_at = greatest(now(), updated_at, $4, $7) where id = $1 and credential_revision = $2 and (quota_observed_at is null or quota_observed_at <= $4)", ) @@ -364,7 +364,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { "update provider_accounts set quota_access_observed_at = $3, quota_access_state = $4, quota_evidence = $5, quota_reset_at = $6, - updated_at = greatest(now(), $3) + updated_at = greatest(now(), updated_at, $3) where id = $1 and credential_revision = $2 and (quota_access_observed_at is null or quota_access_observed_at <= $3)", ) @@ -389,7 +389,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository { require_nonempty(ENTITY, "account_id", account_id)?; let result = sqlx::query( "update provider_accounts - set quota_observed_at = $3, updated_at = greatest(now(), $3) + set quota_observed_at = $3, updated_at = greatest(now(), updated_at, $3) where id = $1 and credential_revision = $2 and provider_quota_json is not null and (quota_observed_at is null or quota_observed_at <= $3)", @@ -685,7 +685,7 @@ impl ProviderAccountAdminRepository for PgProviderAccountRepository { quota_reset_at = null, last_error_reason = null, last_error_message = null, - updated_at = now() + updated_at = greatest(now(), updated_at) where id = $1 returning id", ) @@ -849,7 +849,7 @@ pub(crate) async fn upsert_provider_account_in_transaction( quota_observed_at = null, last_error_reason = null, last_error_message = null, - updated_at = greatest(now(), excluded.credential_observed_at) + updated_at = greatest(now(), provider_accounts.updated_at, excluded.credential_observed_at) returning id", ) .bind(&account.id) diff --git a/backend/crates/gateway-store/tests/postgres/execution.rs b/backend/crates/gateway-store/tests/postgres/execution.rs index 290d4bfba..1dfa7b40d 100644 --- a/backend/crates/gateway-store/tests/postgres/execution.rs +++ b/backend/crates/gateway-store/tests/postgres/execution.rs @@ -627,6 +627,115 @@ fn successful_core_finalization(id: &str) -> CoreModelRequestFinalization { } } +#[tokio::test] +async fn clock_rollback_preserves_terminal_outcomes_and_errors_after_recovery() { + let Some(database) = TestDatabase::create("execution_clock_rollback").await else { + return; + }; + let store = PgExecutionStore::new(database.pool.clone()); + for (index, outcome) in [ + ExecutionOutcome::Succeeded, + ExecutionOutcome::Failed, + ExecutionOutcome::Cancelled, + ExecutionOutcome::Incomplete, + ] + .into_iter() + .enumerate() + { + let id = format!("req_clock_rollback_{index}"); + let request = accepted_request(&id); + store + .create_model_request(request.clone()) + .await + .expect("create request before clock rollback"); + store + .begin_model_request_attempt(ModelRequestAttemptStart { + account_selection_wait_ms: None, + capacity_used_slots: None, + capacity_total_slots: None, + model_request_id: id.clone(), + attempt_count: 1, + provider_kind: "openai".to_owned(), + provider_account_id: None, + provider_account_ref: Some("acct_clock_rollback".to_owned()), + upstream_model_id: Some("coding".to_owned()), + upstream_transport: "websocket".to_owned(), + http_version: None, + }) + .await + .expect("record real attempt before clock rollback"); + let mut finalization = successful_core_finalization(&id); + finalization.outcome = outcome; + finalization.timings.latency_ms = Some(125); + finalization.completed_at = request.started_at - StdDuration::from_millis(400); + if outcome != ExecutionOutcome::Succeeded { + finalization.error = Some(GatewayError::new( + GatewayErrorKind::RateLimited, + "synthetic upstream quota error", + )); + finalization.provider_error_code = Some("usage_limit_reached".to_owned()); + finalization.upstream_status_code = Some(429); + } + ExecutionStore::finalize_model_request(&store, finalization) + .await + .expect("finalize despite wall clock rollback"); + let terminal = stored_row(&database.pool, &id).await; + assert_eq!(terminal["completed_at"], terminal["started_at"]); + assert_eq!(terminal["latency_ms"], 125); + assert_eq!( + terminal["outcome"], + ["succeeded", "failed", "cancelled", "incomplete"][index] + ); + if outcome != ExecutionOutcome::Succeeded { + assert_eq!(terminal["error_kind"], "rate_limited"); + assert_eq!(terminal["provider_error_code"], "usage_limit_reached"); + assert_eq!(terminal["error_message"], "synthetic upstream quota error"); + } + assert_eq!( + store + .recover_expired(request.deadline_at) + .await + .expect("recovery leaves finalized request intact") + .requests, + 0 + ); + assert_eq!(stored_row(&database.pool, &id).await, terminal); + } + database.close().await; +} + +#[tokio::test] +async fn zero_attempt_finalization_preserves_real_error_after_clock_rollback() { + let Some(database) = TestDatabase::create("zero_attempt_clock_rollback").await else { + return; + }; + let store = PgExecutionStore::new(database.pool.clone()); + let request = accepted_request("req_zero_attempt_clock_rollback"); + store + .create_model_request(request.clone()) + .await + .expect("create zero attempt request"); + let mut finalization = early_failure(&request); + finalization.completed_at = request.started_at - StdDuration::from_millis(400); + ExecutionStore::finalize_model_request(&store, finalization) + .await + .expect("finalize zero attempt after clock rollback"); + assert_eq!( + store + .recover_expired(request.deadline_at) + .await + .expect("recovery preserves real early failure") + .requests, + 0 + ); + let terminal = stored_row(&database.pool, request.id.as_str()).await; + assert_eq!(terminal["outcome"], "failed"); + assert_eq!(terminal["error_kind"], "no_available_provider"); + assert_eq!(terminal["attempt_count"], 0); + assert_eq!(terminal["completed_at"], terminal["started_at"]); + database.close().await; +} + #[tokio::test] async fn core_adapter_persists_opaque_response_ids_as_bytes() { let Some(database) = TestDatabase::create("execution_opaque_response_id").await else { diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs index 83a784a65..4f7b90804 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs @@ -5,6 +5,7 @@ use std::{ }; mod quota_forecast; +mod timestamps; use chrono::{TimeDelta, Utc}; use gateway_admin::{ diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/timestamps.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/timestamps.rs new file mode 100644 index 000000000..57c6961ee --- /dev/null +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/timestamps.rs @@ -0,0 +1,261 @@ +use super::*; +use gateway_core::account::AccountConcurrencyLimit; +use gateway_store::postgres::RecoverProviderAccount; + +#[tokio::test] +async fn cross_kind_observations_preserve_time_bounds_and_stale_observation_fences() { + let Some(database) = TestDatabase::create("account_observation_time_bounds").await else { + return; + }; + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let account_id = ProviderAccountId::new("acct_observation_time_bounds").expect("account id"); + let mut seed = account(account_id.as_str(), "user-observation-time-bounds"); + // 合成应用时钟领先数据库的已有事实，避免修改宿主时钟或依赖测试执行速度。 + let anchor = Utc::now() + TimeDelta::hours(1); + seed.credential_observed_at = anchor; + repository + .insert_provider_account(seed) + .await + .expect("seed account"); + let revision = CredentialRevision::new(1).expect("revision"); + let quota_at = SystemTime::from(anchor - TimeDelta::milliseconds(400)); + let observation = QuotaObservation { + plan_type: None, + account_id: account_id.clone(), + expected_revision: revision, + quota: OpaqueProviderData::new(json!({"marker": "current"}).as_object().unwrap().clone()), + observed_at: quota_at, + state: QuotaState::allowed(quota_at), + }; + assert_eq!( + repository + .compare_and_swap_quota(observation.clone()) + .await + .expect("quota before credential observation"), + QuotaWriteOutcome::Updated + ); + assert_eq!( + repository + .touch_quota_observation(QuotaObservationTouch { + account_id: account_id.clone(), + expected_revision: revision, + observed_at: quota_at + Duration::from_millis(100), + }) + .await + .expect("quota touch before credential observation"), + QuotaWriteOutcome::Updated + ); + assert_eq!( + repository + .apply_quota_access(QuotaAccessChange { + account_id: account_id.clone(), + expected_revision: revision, + state: QuotaState::allowed(quota_at + Duration::from_millis(200)), + }) + .await + .expect("access before credential observation"), + QuotaWriteOutcome::Updated + ); + + let access_at = SystemTime::from(anchor + TimeDelta::seconds(1)); + let mut newer = observation.clone(); + newer.observed_at = SystemTime::from(anchor + TimeDelta::milliseconds(500)); + newer.state = QuotaState::exhausted(QuotaEvidence::UsageLimitReached, access_at, None); + assert_eq!( + repository + .compare_and_swap_quota(newer) + .await + .expect("access time may exceed quota document time"), + QuotaWriteOutcome::Updated + ); + repository + .apply_state_change(AccountStateChange { + account_id: account_id.clone(), + expected_revision: revision, + credential_state: CredentialState::Ready, + observed_at: SystemTime::from(anchor + TimeDelta::milliseconds(600)), + error_reason: None, + message: None, + }) + .await + .expect("credential update preserves later quota access time"); + assert_eq!( + repository + .compare_and_swap_quota(observation) + .await + .expect("stale quota remains a conflict"), + QuotaWriteOutcome::Conflict + ); + assert_eq!( + repository + .apply_quota_access(QuotaAccessChange { + account_id: account_id.clone(), + expected_revision: revision, + state: QuotaState::allowed(quota_at), + }) + .await + .expect("stale access remains a conflict"), + QuotaWriteOutcome::Conflict + ); + assert!( + repository + .apply_state_change(AccountStateChange { + account_id: account_id.clone(), + expected_revision: revision, + credential_state: CredentialState::Invalid, + observed_at: SystemTime::from(anchor), + error_reason: Some(AccountErrorReason::CredentialInvalid), + message: None, + }) + .await + .is_err() + ); + let stored = repository + .load_provider_account(account_id.as_str()) + .await + .expect("load bounded account") + .expect("account"); + assert_eq!( + stored.summary.updated_at.timestamp_micros(), + chrono::DateTime::::from(access_at).timestamp_micros() + ); + assert_eq!(stored.summary.credential_state, CredentialState::Ready); + let quota = repository + .get_quotas(std::slice::from_ref(&account_id)) + .await + .expect("load quota") + .pop() + .expect("quota"); + assert_eq!(quota.state.access(), QuotaAccessState::Exhausted); + database.close().await; +} + +#[tokio::test] +async fn credential_and_admin_writes_preserve_existing_update_time_after_clock_rollback() { + let Some(database) = TestDatabase::create("account_update_clock_rollback").await else { + return; + }; + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let account_id = ProviderAccountId::new("acct_update_clock_rollback").expect("account id"); + let mut seed = account(account_id.as_str(), "user-update-clock-rollback"); + let anchor = Utc::now() + TimeDelta::hours(1); + seed.credential_observed_at = anchor; + repository + .insert_provider_account(seed) + .await + .expect("seed account"); + // 账号在回拨前创建，管理恢复和重新导入也必须保留这一时间下界。 + sqlx::query("update provider_accounts set created_at = updated_at where id = $1") + .bind(account_id.as_str()) + .execute(&database.pool) + .await + .expect("seed creation before rollback"); + assert!( + repository + .update_provider_account(profile(account_id.as_str(), "after rollback")) + .await + .expect("update profile") + ); + assert!( + repository + .set_provider_account_enabled(account_id.as_str(), false) + .await + .expect("disable account") + ); + assert!( + repository + .set_provider_account_enabled(account_id.as_str(), true) + .await + .expect("enable account") + ); + assert_eq!( + repository + .compare_and_swap_credentials(credential_update( + account_id.as_str(), + 1, + "synthetic-rotated" + )) + .await + .expect("rotate credentials") + .get(), + 2 + ); + let refreshed = CredentialCasUpdate::new( + account_id.clone(), + CredentialRevision::new(2).expect("revision"), + ProviderAccountUpdate { + account_id: account_id.clone(), + name: "refreshed".to_owned(), + email: None, + plan_type: None, + }, + plaintext_credential("synthetic-refreshed"), + false, + None, + None, + ) + .expect("refresh update") + .with_account_state(CredentialState::Ready, SystemTime::now(), None, None); + assert_eq!( + repository + .compare_and_swap_credential(refreshed.clone()) + .await + .expect("core credential refresh"), + CredentialCasOutcome::Updated(CredentialRevision::new(3).expect("new revision")) + ); + assert_eq!( + repository + .compare_and_swap_credential(refreshed) + .await + .expect("stale refresh conflicts"), + CredentialCasOutcome::Conflict + ); + assert!( + admin_account_store(&database.pool) + .lower_concurrency_limit( + &account_id, + AccountConcurrencyLimit::new(1).expect("limit"), + &MutationContext { + actor: MutationActor::System, + request_id: "rollback-concurrency".to_owned() + }, + ) + .await + .expect("lower concurrency after rollback") + .is_some() + ); + repository + .recover_provider_account_admin(RecoverProviderAccount { + account_id: account_id.as_str().to_owned(), + audit: audit("audit_rollback_recover", "recover", account_id.as_str()), + }) + .await + .expect("recover account after rollback"); + let imported = repository + .import_provider_accounts(ImportProviderAccounts { + settings: None, + outbound_proxy: None, + scope: ProviderAccountAdminScope { + provider_kind: "openai".to_owned(), + }, + accounts: vec![account( + "acct_reimport_clock_rollback", + "user-update-clock-rollback", + )], + audit: audit("audit_rollback_import", "import", account_id.as_str()), + }) + .await + .expect("reimport account after rollback"); + assert_eq!(imported.account_ids, [account_id.as_str()]); + let stored = repository + .load_provider_account(account_id.as_str()) + .await + .expect("load account") + .expect("account"); + assert_eq!( + stored.summary.updated_at.timestamp_micros(), + anchor.timestamp_micros() + ); + assert_eq!(stored.summary.credential_revision.get(), 4); + database.close().await; +} diff --git a/backend/crates/providers/openai/src/credential/catalog.rs b/backend/crates/providers/openai/src/credential/catalog.rs index 64e6f304b..cd4c88d84 100644 --- a/backend/crates/providers/openai/src/credential/catalog.rs +++ b/backend/crates/providers/openai/src/credential/catalog.rs @@ -14,8 +14,8 @@ use gateway_core::provider_ports::{ ProviderCatalogCacheKey, ProviderCatalogCachePort, ProviderCatalogScope, }; use gateway_core::routing::{ - FrozenAccountScope, ProviderCatalogGeneration, ProviderKind, ProviderModelDescriptor, - UpstreamModelId, + FrozenAccountScope, ModelPresentation, ProviderCatalogGeneration, ProviderKind, + ProviderModelContent, ProviderModelDescriptor, UpstreamModelId, }; use secrecy::ExposeSecret; use thiserror::Error; @@ -264,7 +264,8 @@ impl CodexCredentialCatalogService { } } - /// 原生客户端目录使用 Key 范围内排序稳定的首个成功账号，不复用套餐级画像。 + /// OAuth 目录使用 Key 范围内排序稳定的首个成功账号；API Key 聚合按完整对象优先稳定选源。 + /// 客户端原生对象不复用套餐级画像。 /// 同一账号/凭据版本/客户端版本的并发读取合并；原生正文仅保留在有界进程缓存。 pub async fn client_model_catalog( &self, @@ -286,15 +287,20 @@ impl CodexCredentialCatalogService { accounts .retain(|account| scope.allows(account.id()) && eligible_catalog_account(account, now)); accounts.sort_by(|left, right| left.id().cmp(right.id())); - let mut adapted = BTreeMap::new(); + let mut api_catalogs = BTreeMap::new(); let (api_accounts, mut accounts): (Vec<_>, Vec<_>) = accounts.into_iter().partition(|account| { account.authentication_kind() == super::CODEX_AUTHENTICATION_KIND_API_KEY }); let mut catalogs = stream::iter(api_accounts) - .map(|account| async move { - let result = self.cached_or_refresh_account_catalog(&account).await; - (account, result) + .map(|account| { + let profile = request_profile.clone(); + async move { + let result = self + .cached_client_account_catalog(&account, client_version, profile) + .await; + (account, result) + } }) .buffer_unordered(MAX_CONCURRENT_API_CATALOGS); // 所有 API 账号共享等待预算；慢上游不能逐个耗尽超时或阻塞 OAuth 目录。 @@ -302,118 +308,136 @@ impl CodexCredentialCatalogService { while let Ok(Some((account, result))) = tokio::time::timeout_at(deadline, catalogs.next()).await { - if let Ok(catalog) = result { - for id in catalog.models() { - if account.model_access().allows(id) { - let model = gateway_core::routing::UpstreamModelId::new(id.clone()) - .map_err(|_| CodexCredentialCatalogError::InvalidCredentialData)?; - adapted - .entry(id.clone()) - .or_insert(ProviderModelDescriptor { - model, - content: gateway_core::routing::ProviderModelContent::Adapted( - gateway_core::routing::ModelPresentation::new( - Some(id.clone()), - None, - ) - .with_agent_tools(true, false), - ), - }); - } - } + if let Ok(mut models) = result { + models.retain(|model| account.model_access().allows(model.model.as_str())); + api_catalogs.insert(account.id().clone(), models); } } drop(catalogs); + // 完整对象优先于普通 ID；同类目录按账号顺序选来源，并保留上游条目顺序。 + let (native, adapted): (Vec<_>, Vec<_>) = api_catalogs + .into_values() + .flatten() + .partition(|model| matches!(model.content, ProviderModelContent::Native(_))); + let mut seen = BTreeSet::new(); + let api_models = native + .into_iter() + .chain(adapted) + .filter(|model| seen.insert(model.model.clone())) + .collect::>(); accounts.retain(|account| { account.authentication_kind() == super::CODEX_AUTHENTICATION_KIND_OAUTH }); let mut last_error = CodexCredentialCatalogError::NoEligibleCredential; for account in accounts.iter().take(MAX_CATALOG_FETCH_ATTEMPTS) { - let profile = request_profile.clone(); - let key = ClientCatalogKey { - account_id: account.id().clone(), - revision: account.revision(), - plan: account.plan_type().map(str::to_owned), - upstream_account_id: account.upstream_account_id().map(str::to_owned), - client_version: client_version.to_owned(), - profile: profile.clone(), - }; - let value = { - let mut cache = self - .client_catalogs - .lock() - .map_err(|_| CodexCredentialCatalogError::Cache)?; - cache.retain(ClientCatalogEntry::is_fresh); - if let Some(entry) = cache.iter().find(|entry| entry.key == key) { - Arc::clone(&entry.value) - } else { - // 不驱逐正在读取的目录，避免并发请求绕过缓存容量限制制造额外出站。 - if cache.len() >= MAX_CLIENT_CATALOGS { - let Some(index) = cache.iter().position(|entry| entry.value.initialized()) - else { - return Err(CodexCredentialCatalogError::Cache); - }; - cache.remove(index); - } - let value = Arc::new(OnceCell::new()); - cache.push(ClientCatalogEntry { - key, - created_at: Instant::now(), - value: Arc::clone(&value), - }); - value - } - }; - let result = value - .get_or_init(|| async { - // 冻结画像，确保实际请求与缓存身份一致；不传递客户端 Bearer 给上游。 - let client = CodexBackendClient::new( - self.http.clone(), - self.base_url.clone(), - CodexWireProfileState::new(profile), - ); - let result = self - .fetch_account_models(&client, account, Some(client_version)) - .await - .map(|fetched| { - fetched - .models - .into_iter() - .map(|model| ProviderModelDescriptor { - model: model.request_model().clone(), - content: gateway_core::routing::ProviderModelContent::Native( - model.document().clone(), - ), - }) - .collect() - }); - CachedClientCatalog { - completed_at: Instant::now(), - result, - } - }) - .await; - match &result.result { + match self + .cached_client_account_catalog(account, client_version, request_profile.clone()) + .await + { Ok(models) => { - for model in models { - adapted.remove(model.model.as_str()); - } - return Ok(models + let native_ids = models .iter() - .cloned() - .chain(adapted.into_values()) + .map(|model| model.model.clone()) + .collect::>(); + return Ok(models + .into_iter() + .chain( + api_models + .into_iter() + .filter(|model| !native_ids.contains(&model.model)), + ) .collect()); } - Err(error) => last_error = error.clone(), + Err(error) => last_error = error, } } - if adapted.is_empty() { + if api_models.is_empty() { Err(last_error) } else { - Ok(adapted.into_values().collect()) + Ok(api_models) } } + async fn cached_client_account_catalog( + &self, + account: &ProviderAccount, + client_version: &str, + profile: CodexWireProfile, + ) -> ClientCatalogResult { + let key = ClientCatalogKey { + account_id: account.id().clone(), + revision: account.revision(), + plan: account.plan_type().map(str::to_owned), + upstream_account_id: account.upstream_account_id().map(str::to_owned), + client_version: client_version.to_owned(), + profile: profile.clone(), + }; + let value = { + let mut cache = self + .client_catalogs + .lock() + .map_err(|_| CodexCredentialCatalogError::Cache)?; + cache.retain(ClientCatalogEntry::is_fresh); + if let Some(entry) = cache.iter().find(|entry| entry.key == key) { + Arc::clone(&entry.value) + } else { + // 不驱逐正在读取的目录，避免并发请求绕过缓存容量限制制造额外出站。 + if cache.len() >= MAX_CLIENT_CATALOGS { + let Some(index) = cache.iter().position(|entry| entry.value.initialized()) + else { + return Err(CodexCredentialCatalogError::Cache); + }; + cache.remove(index); + } + let value = Arc::new(OnceCell::new()); + cache.push(ClientCatalogEntry { + key, + created_at: Instant::now(), + value: Arc::clone(&value), + }); + value + } + }; + let result = value + .get_or_init(|| async { + // 冻结画像，确保实际请求与缓存身份一致；不传递客户端 Bearer 给上游。 + let client = CodexBackendClient::new( + self.http.clone(), + self.base_url.clone(), + CodexWireProfileState::new(profile), + ); + let result = self + .fetch_account_models(&client, account, Some(client_version)) + .await + .map(|fetched| { + fetched + .models + .into_iter() + .map(|model| ProviderModelDescriptor { + model: model.request_model().clone(), + content: if model.document().protocol() == "codex" { + ProviderModelContent::Native(model.document().clone()) + } else { + ProviderModelContent::Adapted( + ModelPresentation::new( + Some(model.request_model().as_str().to_owned()), + None, + ) + .with_agent_tools(true, false), + ) + }, + }) + .collect() + }); + CachedClientCatalog { + completed_at: Instant::now(), + result, + } + }) + .await; + result.result.clone() + } + #[must_use] pub fn catalog_generation(&self) -> ProviderCatalogGeneration { let cache = self diff --git a/backend/crates/providers/openai/src/transport/catalog.rs b/backend/crates/providers/openai/src/transport/catalog.rs index 843e34c3d..734a17708 100644 --- a/backend/crates/providers/openai/src/transport/catalog.rs +++ b/backend/crates/providers/openai/src/transport/catalog.rs @@ -9,6 +9,8 @@ use gateway_core::routing::{ModelServiceTier, UpstreamModelId}; use reqwest::header::{ETAG, HeaderMap}; use serde::Deserialize; +use super::client::OpenAiUpstreamProtocol; + /// 单次 Codex 模型目录响应允许的最大字节数。 pub const MAX_CODEX_MODEL_CATALOG_BYTES: usize = 1024 * 1024; @@ -340,6 +342,14 @@ enum CodexInputModalityWire { pub fn parse_codex_model_catalog( body: &[u8], etag: Option<&str>, +) -> Result { + parse_native_model_catalog(body, etag, OpenAiUpstreamProtocol::Codex) +} + +fn parse_native_model_catalog( + body: &[u8], + etag: Option<&str>, + protocol: OpenAiUpstreamProtocol, ) -> Result { if body.len() > MAX_CODEX_MODEL_CATALOG_BYTES { return Err(CodexModelCatalogError::ResponseTooLarge); @@ -366,7 +376,7 @@ pub fn parse_codex_model_catalog( ), ) .map_err(|_| CodexModelCatalogError::InvalidWire)?; - let model = normalize_model(wire, document)?; + let model = normalize_model(wire, document, protocol)?; if !seen.insert(model.request_model.as_str().to_owned()) { return Err(CodexModelCatalogError::DuplicateModelSlug); } @@ -390,8 +400,12 @@ pub(super) fn catalog_etag(headers: &HeaderMap) -> Result, CodexM fn normalize_model( wire: CodexModelWire, document: RawJsonPayload, + protocol: OpenAiUpstreamProtocol, ) -> Result { - if !valid_model_slug(&wire.slug) { + // 协商为 Codex 对象不改变 Responses API 的模型 ID 合同，命名空间和微调模型仍有效。 + if wire.slug.trim() != wire.slug + || (protocol == OpenAiUpstreamProtocol::Codex && !valid_model_slug(&wire.slug)) + { return Err(CodexModelCatalogError::InvalidModelSlug); } let request_model = @@ -604,8 +618,14 @@ pub(crate) fn parse_api_model_catalog( if body.len() > MAX_CODEX_MODEL_CATALOG_BYTES { return Err(CodexModelCatalogError::ResponseTooLarge); } - let wire: ApiModels = + let value: serde_json::Value = serde_json::from_slice(body).map_err(|_| CodexModelCatalogError::InvalidWire)?; + // 部分 Responses 上游按客户端版本返回完整 Codex 目录；不能降格为只有 ID 的画像。 + if value.get("models").is_some() { + return parse_native_model_catalog(body, etag, OpenAiUpstreamProtocol::ResponsesApi); + } + let wire: ApiModels = + serde_json::from_value(value).map_err(|_| CodexModelCatalogError::InvalidWire)?; if wire.data.is_empty() { return Err(CodexModelCatalogError::EmptySnapshot); } diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 4750b5450..f353549f6 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -523,11 +523,12 @@ impl CodexBackendClient { .client .get(endpoint_url(&self.base_url, path)) .headers(headers); - if self.protocol == OpenAiUpstreamProtocol::Codex { - request = request.query(&[( - "client_version", - client_version.unwrap_or(profile.codex_version.as_str()), - )]); + let client_version = client_version.or_else(|| { + (self.protocol == OpenAiUpstreamProtocol::Codex) + .then_some(profile.codex_version.as_str()) + }); + if let Some(client_version) = client_version { + request = request.query(&[("client_version", client_version)]); } let response = request.send().await?; let status = response.status(); diff --git a/backend/crates/providers/openai/tests/credential/catalog.rs b/backend/crates/providers/openai/tests/credential/catalog.rs index 1388695ec..41b2e7280 100644 --- a/backend/crates/providers/openai/tests/credential/catalog.rs +++ b/backend/crates/providers/openai/tests/credential/catalog.rs @@ -869,6 +869,269 @@ async fn api_key_catalogs_are_isolated_and_join_oauth_without_claiming_native_me ); } +#[tokio::test] +async fn api_key_client_catalog_negotiates_and_preserves_versioned_native_objects() { + use gateway_core::routing::ProviderModelContent; + use provider_openai::credential::ApiKeyTransport; + use serde_json::json; + + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .await; + let account = store.account("acct_api").expect("API account"); + // 先填满普通 ID 目录，确保客户端不会把已有套餐缓存误当成原生元数据。 + Mock::given(method("GET")) + .and(path("/models")) + .and(|request: &Request| request.url.query().is_none()) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "data": [{"id":"deepseek-flash"}] + }))) + .expect(1) + .mount(&upstream) + .await; + let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache()); + service.synchronize().await.expect("background ID catalog"); + assert_eq!( + service.cached_account_models(&account).unwrap().unwrap(), + ["deepseek-flash"] + ); + let scope = client_scope(&[account]); + for (version, effort) in [("0.155.0", "high"), ("0.156.0", "ultra")] { + let original = json!({ + "slug":"deepseek-flash", "display_name":"DeepSeek Flash", + "default_reasoning_level":effort, + "supported_reasoning_levels":[ + {"effort":"low", "description":"Fast"}, + {"effort":effort, "description":"Upstream description"} + ], + "base_instructions":"Upstream model instructions", + "model_messages":{"instructions_template":"Original template"}, + "context_window":128000, "future_field":{"null":null,"list":[1,true]} + }); + Mock::given(method("GET")) + .and(path("/models")) + .and(query_param("client_version", version)) + .and(header("authorization", "Bearer sk-api-test-only")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!({"models":[original]})) + .set_delay(std::time::Duration::from_millis(20)), + ) + .expect(1) + .mount(&upstream) + .await; + let (first, concurrent) = tokio::join!( + service.client_model_catalog(&scope, version), + service.client_model_catalog(&scope, version) + ); + for models in [first, concurrent] { + let models = models.expect("native client catalog"); + assert_eq!(models.len(), 1); + let ProviderModelContent::Native(payload) = &models[0].content else { + panic!("API Key native metadata must not be replaced by an adapted ID") + }; + assert_eq!(payload.protocol(), "codex"); + assert_eq!( + serde_json::from_slice::(payload.body()).unwrap(), + original + ); + } + } + let cached = service + .client_model_catalog(&scope, "0.155.0") + .await + .expect("separate version cache"); + let ProviderModelContent::Native(payload) = &cached[0].content else { + panic!("native cached model") + }; + let document: serde_json::Value = serde_json::from_slice(payload.body()).unwrap(); + assert_eq!(document["default_reasoning_level"], "high"); + assert!( + service + .client_model_catalog(&client_scope(&[]), "0.155.0") + .await + .is_err() + ); + upstream.verify().await; +} + +#[tokio::test] +async fn api_key_native_catalog_preserves_unknown_reasoning_and_background_entitlements() { + use gateway_core::routing::ProviderModelContent; + use provider_openai::credential::ApiKeyTransport; + use provider_openai::transport::CodexCatalogCapabilityEvidence; + use serde_json::json; + + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .await; + let original = json!({"slug":"z-unknown-reasoning", "display_name":"Unknown reasoning"}); + let second = json!({"slug":"vendor/model", "display_name":"Second upstream model"}); + Mock::given(method("GET")) + .and(path("/models")) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!({"models":[original, second]})), + ) + .expect(2) + .mount(&upstream) + .await; + let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache()); + let snapshot = service + .synchronize() + .await + .expect("native background directory"); + assert_eq!( + snapshot.models()[0].capabilities().reasoning(), + CodexCatalogCapabilityEvidence::Unknown + ); + let account = store.account("acct_api").unwrap(); + assert_eq!( + service.cached_account_models(&account).unwrap().unwrap(), + ["z-unknown-reasoning", "vendor/model"] + ); + let models = service + .client_model_catalog(&client_scope(&[account]), "0.155.0") + .await + .unwrap(); + assert_eq!( + models + .iter() + .map(|model| model.model.as_str()) + .collect::>(), + ["z-unknown-reasoning", "vendor/model"] + ); + let ProviderModelContent::Native(payload) = &models[0].content else { + panic!("native model expected") + }; + assert_eq!( + serde_json::from_slice::(payload.body()).unwrap(), + original + ); + upstream.verify().await; +} + +#[tokio::test] +async fn api_key_client_catalog_keeps_native_sources_stable_and_account_scoped() { + use gateway_core::routing::ProviderModelContent; + use provider_openai::credential::ApiKeyTransport; + use serde_json::json; + + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + let mut accounts = Vec::new(); + for (id, delay, body) in [ + ("acct_a", 0, json!({"data":[{"id":"shared"}]})), + ( + "acct_b", + 30, + json!({"models":[{"slug":"shared", "display_name":"First native", "default_reasoning_level":"high"}]}), + ), + ( + "acct_c", + 0, + json!({"models":[{"slug":"shared", "display_name":"Second native", "default_reasoning_level":"low"}]}), + ), + ] { + store + .seed_api_key( + id, + format!("{}/{id}", upstream.uri()), + ApiKeyTransport::Http, + ) + .await; + accounts.push(store.account(id).unwrap()); + Mock::given(method("GET")) + .and(path(format!("/{id}/models"))) + .and(query_param("client_version", "0.155.0")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(body) + .set_delay(std::time::Duration::from_millis(delay)), + ) + .expect(1) + .mount(&upstream) + .await; + } + let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache()); + for (scope_accounts, expected) in [ + (&accounts[..], "First native"), + (&accounts[2..], "Second native"), + ] { + let models = service + .client_model_catalog(&client_scope(scope_accounts), "0.155.0") + .await + .unwrap(); + assert_eq!(models.len(), 1); + let ProviderModelContent::Native(payload) = &models[0].content else { + panic!("native catalog must take precedence over plain IDs") + }; + let document: serde_json::Value = serde_json::from_slice(payload.body()).unwrap(); + assert_eq!(document["display_name"], expected); + } + upstream.verify().await; +} + +#[tokio::test] +async fn api_key_client_catalog_cache_is_invalidated_by_credential_revision() { + use gateway_core::account::{CredentialCasOutcome, CredentialCasUpdate, ProviderAccountUpdate}; + use provider_openai::credential::ApiKeyTransport; + + let upstream = MockServer::start().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .await; + let account = store.account("acct_api").unwrap(); + let scope = client_scope(std::slice::from_ref(&account)); + let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache()); + for model in ["before-rotation", "after-rotation"] { + upstream.reset().await; + Mock::given(method("GET")) + .and(path("/models")) + .and(query_param("client_version", "0.155.0")) + .respond_with( + ResponseTemplate::new(200).set_body_json(serde_json::json!({"models":[{ + "slug":model, "display_name":model + }]})), + ) + .expect(1) + .mount(&upstream) + .await; + let models = service + .client_model_catalog(&scope, "0.155.0") + .await + .unwrap(); + assert_eq!(models[0].model.as_str(), model); + upstream.verify().await; + if model == "before-rotation" { + let loaded = store.load_current_credential(account.id()).await.unwrap(); + let update = CredentialCasUpdate::new( + account.id().clone(), + account.revision(), + ProviderAccountUpdate { + account_id: account.id().clone(), + name: account.name().to_owned(), + email: None, + plan_type: None, + }, + loaded.credential, + false, + None, + None, + ) + .unwrap(); + assert!(matches!( + store.compare_and_swap_credential(update).await.unwrap(), + CredentialCasOutcome::Updated(_) + )); + } + } +} + #[tokio::test] async fn api_key_catalog_accepts_namespaced_model_ids_without_inventing_capabilities() { use provider_openai::credential::ApiKeyTransport; @@ -920,6 +1183,14 @@ async fn api_key_catalog_accepts_namespaced_model_ids_without_inventing_capabili .collect::>(), ids.into_iter().collect() ); + for model in &models { + let gateway_core::routing::ProviderModelContent::Adapted(presentation) = &model.content + else { + panic!("plain API IDs must remain adapted") + }; + assert!(presentation.supported_reasoning_efforts().is_empty()); + assert_eq!(presentation.default_reasoning_effort(), None); + } } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index f9173b032..203471675 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -8768,15 +8768,19 @@ async fn api_key_default_http_uses_own_prefix_plain_json_and_only_own_authentica provider_openai::credential::ApiKeyTransport::Http, ) .await; - Mock::given(method("GET")) - .and(path(format!("{prefix}/models"))) - .and(header("authorization", "Bearer sk-api-test-only")) - .respond_with( - ResponseTemplate::new(200).set_body_json(json!({"data":[{"id":"gpt-5.4"}]})), - ) - .expect(1) - .mount(&upstream) - .await; + // 后台发现不协商客户端版本；客户端目录独立请求并按实际版本缓存。 + for query in [None, Some("client_version=1.0.0")] { + Mock::given(method("GET")) + .and(path(format!("{prefix}/models"))) + .and(move |request: &wiremock::Request| request.url.query() == query) + .and(header("authorization", "Bearer sk-api-test-only")) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!({"data":[{"id":"gpt-5.4"}]})), + ) + .expect(1) + .mount(&upstream) + .await; + } Mock::given(method("POST")) .and(path(format!("{prefix}/responses"))) .and(header("authorization", "Bearer sk-api-test-only")) @@ -8844,19 +8848,37 @@ async fn api_key_default_http_uses_own_prefix_plain_json_and_only_own_authentica request.headers["user-agent"], wire_profile().snapshot().user_agent() ); - let model_request = requests + let model_requests = requests .iter() - .find(|request| request.method == "GET") - .unwrap(); - assert_eq!( - model_request.headers["user-agent"], - wire_profile().snapshot().user_agent() - ); - assert_eq!(model_request.headers["originator"], "codex_cli_rs"); - assert_eq!(model_request.headers["version"], "0.144.0"); + .filter(|request| request.method == "GET") + .collect::>(); + assert_eq!(model_requests.len(), 2); + for query in [None, Some("client_version=1.0.0")] { + let model_request = model_requests + .iter() + .find(|request| request.url.query() == query) + .expect("separate background and client catalog requests"); + assert_eq!( + model_request.headers["authorization"], + "Bearer sk-api-test-only" + ); + assert_eq!( + model_request.headers["user-agent"], + wire_profile().snapshot().user_agent() + ); + assert_eq!(model_request.headers["originator"], "codex_cli_rs"); + assert_eq!(model_request.headers["version"], "0.144.0"); + for header in ["cookie", "chatgpt-account-id"] { + assert!( + !model_request.headers.contains_key(header), + "unexpected catalog {header}" + ); + } + } let body: Value = serde_json::from_slice(&request.body).expect("ordinary JSON"); assert_eq!(body["stream"], true); assert_eq!(body["model"], "gpt-5.4"); + upstream.verify().await; } } diff --git a/docs/api.md b/docs/api.md index 234124dc1..c42d69a93 100644 --- a/docs/api.md +++ b/docs/api.md @@ -236,13 +236,17 @@ HTML 或截断正文当作 message。 OpenAI Provider 按客户端传入的 `client_version` 请求上游目录，完整保留每个模型 JSON 对象，包括 `base_instructions`、`model_messages`、`service_tiers`、工具与能力字段，以及未知嵌套字段、显式 `null` 和字段缺失的区别。 +API Key 上游返回完整 Codex `models` 目录时沿用该合同；仅返回普通 `data` 模型列表时使用通用画像， +未提供的推理能力保持未知，不补充推理档位。 模型别名仅替换 `slug`，不替换上游展示名、提示词、能力或 `priority`；保持原生模型顺序，新增别名附在后面。 目录按当前路由快照的模型存在性及账号模型政策过滤，避免公布已知无法路由的模型；新模型需待后台目录对账后进入列表。 xAI 没有 Codex 原生目录，继续使用明确的通用画像适配。 -目录账号只能来自本次 Client Key 冻结的账号范围。OpenAI 在其中按账号 ID 排序，使用首个成功读取的 -合格账号，最多尝试三个账号；同名模型不跨账号或套餐混拼字段。因此单账号、无别名时可保持该账号的 -模型对象一致，多账号/多 Provider 聚合不代表“与某个官方账号的整个目录完全一致”，也不会固定后续推理账号。 +目录账号只能来自本次 Client Key 冻结的账号范围。OpenAI 的 OAuth 目录按账号 ID 排序，使用首个成功读取的 +合格账号，最多尝试三个账号；API Key 目录按账号模型权限过滤后聚合。同名模型优先采用 OAuth 原生对象， +其次采用 API Key 的完整 Codex 对象，再使用普通模型 ID；同类 API Key 目录按账号 ID 确定来源，不跨账号 +或套餐混拼字段。因此单账号、无别名时可保持该账号的模型对象一致，多账号/多 Provider 聚合不代表 +“与某个官方账号的整个目录完全一致”，也不会固定后续推理账号。 读取失败返回 `503 model_catalog_unavailable`，不以简化模板或空成功响应覆盖客户端缓存。 原生目录可能返回缓存结果，成功缓存有效期为 5 分钟。每次查询都检查账号资格和 Client Key 范围。 diff --git a/docs/architecture.md b/docs/architecture.md index 704d3a253..964c77384 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -203,9 +203,10 @@ Core 只理解 `Operation`、能力要求、Provider 候选、稳定错误和 ca 类型。Provider 独占 credential schema、OAuth、账号选择、模型目录、额度投影和上游 transport。 OpenAI 的 OAuth 与 API Key 共用现有账号和事务。API Key 的 Base URL、密钥和传输策略属于 Provider 凭据 JSON， -随 credential revision 更新；普通详情只投影非敏感连接设置。API Key 目录按账号和凭据版本隔离，标准 API 模型列表 -通过通用画像输出客户端目录，OAuth 原生对象保留。通用账号层按 Provider 提交的 credential state 调度，不以是否存在 -上游用户 ID 推断可用性；OAuth 未完成身份投影时由 Provider 保持 `unknown`。状态恢复和未补齐身份的凭据轮换保留 `unknown`。 +随 credential revision 更新；普通详情只投影非敏感连接设置。API Key 客户端目录按账号、凭据版本和客户端版本隔离， +标准 `data` 模型列表通过通用画像输出，完整 Codex `models` 目录保留原生对象。通用账号层按 Provider 提交的 +credential state 调度，不以是否存在上游用户 ID 推断可用性；OAuth 未完成身份投影时由 Provider 保持 `unknown`。 +状态恢复和未补齐身份的凭据轮换保留 `unknown`。 API Key 默认 HTTP/SSE，可选 WS 优先；选号先验证传输资格，WS pool 与 continuation 按凭据版本隔离。 OAuth 与 API Key 共用业务请求、响应和能力透传链路，差异限定在上游地址、认证、传输配置及明确的上游请求合同适配。 OpenAI 模型目录用于发现，不因目录缺项拒绝请求；管理员配置的模型权限仍由 Core 与选号链路执行。 diff --git a/frontend/src/views/accounts/components/AccountQuotaPanel/ResetCredits.vue b/frontend/src/views/accounts/components/AccountQuotaPanel/ResetCredits.vue index dd65eba3a..1d56345f5 100644 --- a/frontend/src/views/accounts/components/AccountQuotaPanel/ResetCredits.vue +++ b/frontend/src/views/accounts/components/AccountQuotaPanel/ResetCredits.vue @@ -17,11 +17,12 @@ const props = defineProps<{ }>() const emit = defineEmits<{ - accountUpdated: [account: Account] + consumed: [accountId: string] }>() dayjs.extend(utc) +const panelOpen = shallowRef(false) const { availableCredits, availableCount, @@ -40,10 +41,13 @@ const { confirmConsume, } = useAccountResetCredits({ accountId: () => props.account.id, - onAccountUpdated: account => emit('accountUpdated', account), + onConsumed: (accountId) => { + if (props.account.id === accountId) + panelOpen.value = false + emit('consumed', accountId) + }, }) -const panelOpen = shallowRef(false) const modalTitle = computed(() => { if (!showConfirm.value) return '额度重置' diff --git a/frontend/src/views/accounts/components/AccountQuotaPanel/index.vue b/frontend/src/views/accounts/components/AccountQuotaPanel/index.vue index bd608f02e..2c73489b3 100644 --- a/frontend/src/views/accounts/components/AccountQuotaPanel/index.vue +++ b/frontend/src/views/accounts/components/AccountQuotaPanel/index.vue @@ -18,7 +18,7 @@ const props = defineProps<{ const emit = defineEmits<{ refreshQuota: [accountId: string] - accountUpdated: [account: AccountRow] + quotaReset: [accountId: string] }>() const quotaEntries = computed(() => groupedAccountQuotaWindows( @@ -61,7 +61,7 @@ const profileOpen = shallowRef(false) { try { @@ -257,5 +270,6 @@ export function useAccountMutations(options: { handleRecover, handleRefresh, handleRefreshQuota, + handleQuotaReset, } } diff --git a/frontend/src/views/accounts/composables/useAccountResetCredits.ts b/frontend/src/views/accounts/composables/useAccountResetCredits.ts index 9f5e812dd..980a658c3 100644 --- a/frontend/src/views/accounts/composables/useAccountResetCredits.ts +++ b/frontend/src/views/accounts/composables/useAccountResetCredits.ts @@ -1,9 +1,8 @@ -import type { Account, AccountResetCredit } from '@/api' +import type { AccountResetCredit } from '@/api' import { computed, shallowReactive, shallowRef, watch } from 'vue' import { consumeAccountResetCredit, getAccountResetCredits, - refreshAccountQuota, } from '@/api' import { ApiError } from '@/api/request' @@ -33,7 +32,6 @@ interface ResetCreditsSession { loadError: string loadSequence: number loadController?: AbortController - accountUpdatedListeners: Set<(account: Account) => void> } // 库存仍以主动查询的上游结果为准；未决操作和消费锁必须跨展开行卸载存续。 @@ -50,7 +48,6 @@ function getResetCreditsSession(accountId: string) { loading: false, loadError: '', loadSequence: 0, - accountUpdatedListeners: new Set(), }) sessionsByAccountId.set(accountId, session) } @@ -85,7 +82,7 @@ async function loadSessionCredits(session: ResetCreditsSession, silent = false) export function useAccountResetCredits(options: { accountId: () => string - onAccountUpdated: (account: Account) => void + onConsumed: (accountId: string) => void }) { const session = shallowRef(getResetCreditsSession(options.accountId())) const credits = computed(() => session.value.snapshot?.credits ?? []) @@ -200,21 +197,13 @@ export function useAccountResetCredits(options: { const successMessage = result.code === 'already_redeemed' ? '上次重置已完成' : '额度已重置' + // 消费已确认就结束交互，账号页负责额度回读，避免收起展开行后丢失更新。 + if (session.value === target) + showConfirm.value = false + options.onConsumed(operation.accountId) + toast.success(successMessage) applyConfirmedConsumption(target, operation) await loadSessionCredits(target, true) - try { - const quota = await refreshAccountQuota({ accountId: operation.accountId }, { silent: true }) - // 原组件可能已经卸载或换号，只通知仍订阅该账号的实例。 - for (const listener of target.accountUpdatedListeners) - listener(quota.account) - toast.success(successMessage) - } - catch (error: unknown) { - toast.warning( - `${successMessage}，但最新额度加载失败：${errorMessage(error, '请手动刷新额度')}`, - { duration: 5000 }, - ) - } return true } catch (error: unknown) { @@ -239,14 +228,11 @@ export function useAccountResetCredits(options: { watch( options.accountId, - (accountId, _, onCleanup) => { + (accountId) => { const target = getResetCreditsSession(accountId) session.value = target selectedCreditId.value = '' showConfirm.value = false - const listener = (account: Account) => options.onAccountUpdated(account) - target.accountUpdatedListeners.add(listener) - onCleanup(() => target.accountUpdatedListeners.delete(listener)) }, { immediate: true, flush: 'sync' }, ) diff --git a/frontend/src/views/accounts/index.vue b/frontend/src/views/accounts/index.vue index 7631c464b..edc4695f7 100644 --- a/frontend/src/views/accounts/index.vue +++ b/frontend/src/views/accounts/index.vue @@ -103,6 +103,7 @@ const { handleRecover, handleRefresh, handleRefreshQuota, + handleQuotaReset, } = useAccountMutations({ onImportTaskCreated: importTasks.created, accounts, @@ -337,7 +338,7 @@ const {