diff --git a/backend/crates/providers/openai/src/credential/catalog.rs b/backend/crates/providers/openai/src/credential/catalog.rs index f9ac83dae..537638978 100644 --- a/backend/crates/providers/openai/src/credential/catalog.rs +++ b/backend/crates/providers/openai/src/credential/catalog.rs @@ -29,6 +29,8 @@ use crate::transport::{CodexBackendClient, CodexCatalogModel, CodexRequestContex const MAX_RESPONSE_ETAG_BYTES: usize = 256; const PLAN_CATALOG_CACHE_TTL: Duration = Duration::from_secs(5 * 60); +// 调度证据覆盖默认 15 分钟刷新周期及抖动,长期未刷新时恢复上游验证 +const SCHEDULING_CATALOG_MAX_AGE: Duration = Duration::from_secs(30 * 60); const MAX_CATALOG_FETCH_ATTEMPTS: usize = 3; const MAX_PLAN_CATALOG_MODELS: usize = 2_048; const MAX_CLIENT_CATALOGS: usize = 32; @@ -537,6 +539,47 @@ impl CodexCredentialCatalogService { .map(|models| models.map(<[String]>::to_vec)) } + /// 新鲜 OAuth 套餐目录只约束已发现的模型,未知目录与模型继续交由上游判断 + pub(crate) fn scheduling_exclusions( + &self, + accounts: &[ProviderAccount], + model: &str, + now: SystemTime, + ) -> BTreeSet { + let Ok(cache) = self.cache.read() else { + return BTreeSet::new(); + }; + let Some(snapshot) = cache.snapshot.as_ref().filter(|snapshot| { + now.duration_since(snapshot.observed_at) + .is_ok_and(|age| age < SCHEDULING_CATALOG_MAX_AGE) + }) else { + return BTreeSet::new(); + }; + // API Key 的自定义目录不能为 OAuth 套餐生成排除证据 + let plans = snapshot.scope_catalogs.iter().filter(|(scope, _)| { + scope.as_str().starts_with("plan:") && scope.as_str() != "plan:unknown" + }); + if !plans + .clone() + .any(|(_, catalog)| catalog.models.iter().any(|id| id == model)) + { + return BTreeSet::new(); + } + let excluded_scopes = plans + .filter(|(_, catalog)| !catalog.models.iter().any(|id| id == model)) + .map(|(scope, _)| scope) + .collect::>(); + accounts + .iter() + .filter(|account| { + account.authentication_kind() == super::CODEX_AUTHENTICATION_KIND_OAUTH + && CodexCatalogScope::for_account(account) + .is_ok_and(|scope| excluded_scopes.contains(&scope)) + }) + .map(|account| account.id().clone()) + .collect() + } + /// 优先读取套餐目录 cache;缺失时才用当前账号所属套餐的有限候选集实时填充 pub async fn cached_or_refresh_account_catalog( &self, diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 5dfd5cfa8..97134e17e 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -17,10 +17,11 @@ use gateway_core::account::{ use gateway_core::concurrency::{CapacityWait, ConcurrencyWaitQueue, QueueRejection, WaitPriority}; use gateway_core::engine::{AttemptContext, ContinuationAttempt, policy::AccountPolicyError}; use gateway_core::provider_ports::{ - ProviderConcurrencyPool, ProviderLeaseAcquisition, ProviderLeaseGuard, ProviderLeasePort, - ProviderLeaseRequest, ProviderSchedulingLeaseRequest, ProviderSessionAffinityKey, - ProviderSessionAffinityPort, ProviderSessionBinding, ProviderSessionExclusionPort, - ProviderSessionExclusions, ProviderStoreError, + ProviderConcurrencyPool, ProviderCooldownPort, ProviderCooldownScope, ProviderLeaseAcquisition, + ProviderLeaseGuard, ProviderLeasePort, ProviderLeaseRequest, ProviderSchedulingLeaseRequest, + ProviderScopedCooldown, ProviderSessionAffinityKey, ProviderSessionAffinityPort, + ProviderSessionBinding, ProviderSessionExclusionPort, ProviderSessionExclusions, + ProviderStoreError, }; use gateway_core::routing::ProviderKind; use secrecy::ExposeSecret; @@ -28,6 +29,7 @@ use thiserror::Error; use url::Url; use super::affinity::CodexSessionAffinity; +use super::catalog::CodexCredentialCatalogService; use super::codec::CodexRuntimeAuthentication; use super::cookie::CodexCookiePolicy; use super::eligibility::CodexSelectionModel; @@ -52,6 +54,8 @@ const CYBER_POLICY_TIMEOUT: Duration = Duration::from_millis(100); const CYBER_POLICY_SESSION_TTL: Duration = Duration::from_secs(60 * 60); const MAX_ACCOUNT_SNAPSHOT_RETRIES: u32 = 3; const COOLDOWN_READ_CONCURRENCY: usize = 128; +// 套餐权限可能变化,明确拒绝只作有界的账号模型排除,不冻结整个账号 +const MODEL_UNSUPPORTED_COOLDOWN: Duration = Duration::from_secs(30 * 60); #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CodexAccountFailure { @@ -135,6 +139,8 @@ pub struct CodexCredentialSelector { session_affinity: Arc, session_exclusions: Arc, quota: Arc, + catalog: Arc, + cooldowns: Arc, upstream_info: Option>, cookie_policy: CodexCookiePolicy, risk_recovery: Mutex>, @@ -273,6 +279,8 @@ impl CodexCredentialSelector { session_affinity: Arc, session_exclusions: Arc, quota: Arc, + catalog: Arc, + cooldowns: Arc, account_feedback: Arc, cookie_policy: CodexCookiePolicy, ) -> Self { @@ -283,6 +291,8 @@ impl CodexCredentialSelector { session_affinity, session_exclusions, quota, + catalog, + cooldowns, upstream_info: None, cookie_policy, risk_recovery: Mutex::new(HashMap::new()), @@ -619,6 +629,14 @@ impl CodexCredentialSelector { }) .collect::>(); let mut eligible = Vec::with_capacity(accounts.len()); + let catalog_excluded = match model { + CodexSelectionModel::Requested(model) if !diagnostic => self + .catalog + .scheduling_exclusions(&accounts, model, SystemTime::now()), + _ => Default::default(), + }; + let mut catalog_rejected = 0_usize; + let mut native_accounts = std::collections::BTreeSet::new(); for account in accounts { let adapter = if let Some(model) = adapter_model.as_ref() { match request.attempt.upstream_adapter_for_account( @@ -638,6 +656,14 @@ impl CodexCredentialSelector { }) { continue; } + // 受管适配器拥有其上游模型合同,不套用官方 OAuth 目录与拒绝反馈 + if adapter.is_none() { + if catalog_excluded.contains(account.id()) { + catalog_rejected += 1; + continue; + } + native_accounts.insert(account.id().clone()); + } if request.requires_websocket && adapter.is_none() && pinned_account.as_ref().is_none_or(|id| id == account.id()) @@ -667,6 +693,12 @@ impl CodexCredentialSelector { eligible.push(account); } let accounts = eligible; + if catalog_rejected > 0 && request.attempt.trace().is_enabled() { + request.attempt.trace().record( + "account.model_catalog", + serde_json::json!({"rejectedCount": catalog_rejected}), + ); + } if model_access_rejected > 0 && request.attempt.trace().is_enabled() { request.attempt.trace().record( "account.model_access", @@ -677,13 +709,48 @@ impl CodexCredentialSelector { self.quota.prepare_scheduling(&accounts).await; } let mut rate_limits = HashMap::with_capacity(accounts.len()); + let model_scope = match model { + CodexSelectionModel::Requested(model) => { + gateway_core::routing::UpstreamModelId::new(model) + .ok() + .map(ProviderCooldownScope::upstream_model) + } + _ => None, + }; if !diagnostic { // 同一批候选并发读取实时冷却,避免逐账号累加 Redis 往返 // 分批限制在途读取,取消选号时直接释放本批 future,不启动后续批次 for batch in accounts.chunks(COOLDOWN_READ_CONCURRENCY) { rate_limits.extend( futures::future::join_all(batch.iter().map(|account| async { - let until = self.quota.cooldown(account.id()).await.unwrap_or(None); + let account_cooldown = self.quota.cooldown(account.id()); + let model_cooldown = async { + let scope = model_scope.as_ref()?; + if !native_accounts.contains(account.id()) { + return None; + } + self.cooldowns + .read_scoped(account.id(), scope) + .await + .ok() + .flatten() + .filter(|cooldown| { + cooldown.credential_revision() == account.revision() + && cooldown.until() > SystemTime::now() + }) + }; + let (account_cooldown, model_cooldown) = + futures::join!(account_cooldown, model_cooldown); + // 保留账号保护的探测语义,同时覆盖两种冷却中较晚的到期时间 + let mut until = account_cooldown.unwrap_or(None); + if let Some(model_cooldown) = model_cooldown { + let cooldown = + until.get_or_insert(gateway_core::account::AccountCooldown { + until: model_cooldown.until(), + kind: gateway_core::account::AccountCooldownKind::RateLimit, + }); + cooldown.until = cooldown.until.max(model_cooldown.until()); + } (account.id().clone(), until) })) .await, @@ -1096,6 +1163,23 @@ impl CodexCredentialSelector { } } + pub(crate) async fn record_model_unsupported( + &self, + account: &ProviderAccount, + model: &gateway_core::routing::UpstreamModelId, + ) -> Result<(), ProviderStoreError> { + let now = SystemTime::now(); + self.cooldowns + .put_scoped_if_later(ProviderScopedCooldown::new( + account.id().clone(), + account.revision(), + ProviderCooldownScope::upstream_model(model.clone()), + now.checked_add(MODEL_UNSUPPORTED_COOLDOWN).unwrap_or(now), + )) + .await?; + Ok(()) + } + pub(crate) async fn remember_turn( &self, turn: &ProviderSessionAffinityKey, diff --git a/backend/crates/providers/openai/src/lib.rs b/backend/crates/providers/openai/src/lib.rs index a5bb19f65..c2678a392 100644 --- a/backend/crates/providers/openai/src/lib.rs +++ b/backend/crates/providers/openai/src/lib.rs @@ -175,6 +175,8 @@ pub async fn initialize( session_affinity, session_exclusions, Arc::clone("a), + Arc::clone(&catalog), + ports.cooldowns(), Arc::clone(&account_feedback), CodexCookiePolicy::official().map_err(|_| OpenAiInitializeError::CookiePolicy)?, ) diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index d495fe751..8c1750bcb 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -629,6 +629,7 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea response_origin: &request.response_origin, cyber_policy_scope: None, allows_account_state_mutation, + upstream_model: None, }; let active_account = request.lease.account().clone(); let cookie_header = build_cookie_header(request.lease.cookies())?; @@ -797,6 +798,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { response_origin: &response_origin, cyber_policy_scope: cyber_policy_scope.as_ref(), allows_account_state_mutation, + upstream_model: Some(&upstream_model), }; let mut active_account = lease.account().clone(); let cookie_header = build_cookie_header(lease.cookies())?; diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 7677eeeab..d2500aefc 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -46,6 +46,7 @@ pub(super) struct MappedProviderFailure { /// 原始上游错误描述,仅在凭据错误状态下持久化 pub(super) error_message: Option, pub(super) cyber_policy_failure: bool, + pub(super) model_unsupported: bool, pub(super) set_cookie_headers: Vec, pub(super) response_origin: Option, pub(super) rate_limit_headers: Vec<(String, String)>, @@ -61,6 +62,7 @@ impl MappedProviderFailure { account_failure: None, error_message: None, cyber_policy_failure: false, + model_unsupported: false, set_cookie_headers: Vec::new(), response_origin: None, rate_limit_headers: Vec::new(), @@ -159,6 +161,7 @@ pub(super) struct OpenAiFailureContext<'a> { pub(super) response_origin: &'a Url, pub(super) cyber_policy_scope: Option<&'a CodexCyberPolicyScope>, pub(super) allows_account_state_mutation: bool, + pub(super) upstream_model: Option<&'a UpstreamModelId>, } #[derive(Clone, Copy)] @@ -308,6 +311,16 @@ pub(super) async fn apply_failure( return; } synchronize_passive_quota_headers(context.quota, account, &failure.rate_limit_headers).await; + if failure.model_unsupported + && let Some(model) = context.upstream_model + && let Err(error) = context + .selector + .record_model_unsupported(account, model) + .await + { + tracing::warn!(account_id = %account.id(), error = %error, + "Failed to persist OpenAI model availability cooldown"); + } let needs_authoritative_quota_refresh = matches!( failure.account_failure, Some(CodexAccountFailure::QuotaExhausted | CodexAccountFailure::UsageLimitExhausted { .. }) @@ -473,12 +486,13 @@ pub(super) fn apply_websocket_recovery_policy( return; } // 明确账号拒绝走已有换号路径,业务瞬态失败走请求内退避,不消耗 WS 传输预算 - if failure.error.replay_is_safe() - && (failure.account_failure.is_some() - || matches!( - failure.error.pre_delivery_retry(), - Some(gateway_core::error::PreDeliveryRetry::SameAccountTransientRetry { .. }) - )) + if failure.model_unsupported + || (failure.error.replay_is_safe() + && (failure.account_failure.is_some() + || matches!( + failure.error.pre_delivery_retry(), + Some(gateway_core::error::PreDeliveryRetry::SameAccountTransientRetry { .. }) + ))) { tracing::warn!( request_id = context.request_id, @@ -1490,6 +1504,7 @@ pub(super) fn map_upstream_failure( ), error_message: failure.client_message, cyber_policy_failure, + model_unsupported: category == CodexFailureCategory::ModelUnsupported, // 普通 5xx 与未知流内错误不足以证明容量拒绝,不能用于冻结账号 set_cookie_headers: failure.set_cookie_headers, response_origin: failure.response_origin, diff --git a/backend/crates/providers/openai/src/provider/live.rs b/backend/crates/providers/openai/src/provider/live.rs index 6df1afd13..a36e6ed95 100644 --- a/backend/crates/providers/openai/src/provider/live.rs +++ b/backend/crates/providers/openai/src/provider/live.rs @@ -423,6 +423,7 @@ fn cold_live_call_stream(request: ColdLiveCall) -> EventStream { response_origin: &request.response_origin, cyber_policy_scope: None, allows_account_state_mutation, + upstream_model: None, }; let active_account = request.lease.account().clone(); let authorization = request diff --git a/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs b/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs index 7bdf2318c..592ab314e 100644 --- a/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs +++ b/backend/crates/providers/openai/tests/credential/contract/cooldown_reads.rs @@ -3,8 +3,10 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use gateway_core::provider_ports::{ - ProviderCooldown, ProviderCooldownKind, ProviderStoreError, ProviderStoreErrorKind, + ProviderCooldown, ProviderCooldownKind, ProviderCooldownScope, ProviderScopedCooldown, + ProviderStoreError, ProviderStoreErrorKind, }; +use gateway_core::routing::UpstreamModelId; use tokio::sync::Notify; use super::*; @@ -68,19 +70,26 @@ fn pool_attempt(store: &Arc, count: usize) -> AttemptContext #[tokio::test(start_paused = true)] async fn selector_overlaps_cooldown_reads_with_bounded_in_flight_work() { - for count in [10_usize, 100, 1000] { + for count in [10_usize, 100, 1000, 3000] { let store = Arc::new(MemoryAccountStore::default()); let attempt = pool_attempt(&store, count); let activity = Arc::new(ReadActivity::default()); let observed = activity.clone(); - let cooldowns = Arc::new(MemoryCooldownPort::new().with_read_hook(Arc::new(move |_| { - let activity = observed.clone(); - Box::pin(async move { - let _read = ActiveRead::begin(activity); - tokio::time::sleep(Duration::from_millis(10)).await; - Ok(()) - }) - }))); + let read_hook = Arc::new( + move |_| -> futures::future::BoxFuture<'static, Result<(), ProviderStoreError>> { + let activity = observed.clone(); + Box::pin(async move { + let _read = ActiveRead::begin(activity); + tokio::time::sleep(Duration::from_millis(10)).await; + Ok(()) + }) + }, + ); + let cooldowns = Arc::new( + MemoryCooldownPort::new() + .with_read_hook(read_hook.clone()) + .with_scoped_read_hook(read_hook), + ); let selector = selector_with_runtime( &store, Arc::new(TestLeaseCoordinator::default()), @@ -101,9 +110,9 @@ async fn selector_overlaps_cooldown_reads_with_bounded_in_flight_work() { .expect("available account"); let elapsed = started.elapsed(); assert!(selected.account_id().as_str().starts_with("acct_")); - assert_eq!(activity.started.load(Ordering::SeqCst), count); + assert_eq!(activity.started.load(Ordering::SeqCst), count * 2); assert_eq!(activity.active.load(Ordering::SeqCst), 0); - assert_eq!(activity.peak.load(Ordering::SeqCst), count.min(128)); + assert_eq!(activity.peak.load(Ordering::SeqCst), count.min(128) * 2); // 使用虚拟时间证明等待被合并,不把 CI 的 CPU 调度抖动当成性能回归 assert!(elapsed <= Duration::from_millis(20 * count.div_ceil(128) as u64)); println!( @@ -253,3 +262,63 @@ async fn concurrent_reads_preserve_cooldown_expiry_probe_freeze_and_error_fallba ); } } + +#[tokio::test] +async fn shorter_account_cooldown_cannot_unmask_an_active_model_cooldown_during_selection() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + store.set_scheduling("acct_first", None, AccountWeight::new(100).unwrap()); + let account = store.account("acct_first").unwrap(); + let cooldowns = MemoryCooldownPort::new(); + let account_cooldowns = cooldowns.cooldowns.clone(); + let revision = account.revision(); + let cooldowns = Arc::new(cooldowns.with_read_hook(Arc::new(move |id| { + let account_cooldowns = account_cooldowns.clone(); + Box::pin(async move { + if id.as_str() == "acct_first" { + account_cooldowns.lock().unwrap().insert( + id.clone(), + ProviderCooldown::new( + id, + revision, + SystemTime::now() + Duration::from_millis(100), + ), + ); + } else { + // 后续候选的读取跨过账号冷却期限,模型冷却仍必须阻止高权重账号 + tokio::time::sleep(Duration::from_millis(200)).await; + } + Ok(()) + }) + }))); + cooldowns + .put_scoped_if_later(ProviderScopedCooldown::new( + account.id().clone(), + revision, + ProviderCooldownScope::upstream_model(UpstreamModelId::new("gpt-5.4").unwrap()), + SystemTime::now() + Duration::from_secs(60), + )) + .await + .unwrap(); + let selector = selector_with_runtime( + &store, + Arc::new(TestLeaseCoordinator::default()), + Arc::new(MemorySessionAffinity::default()), + Arc::new(AccountFeedbackStats::default()), + cooldowns.clone(), + ); + let request_url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let attempt = attempt(BTreeSet::new()); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &request_url, + attempt: &attempt, + session_affinity_key: None, + }) + .await + .unwrap(); + assert!(cooldowns.cooldowns.lock().unwrap()[account.id()].until() < SystemTime::now()); + assert_eq!(selected.account_id().as_str(), "acct_second"); +} diff --git a/backend/crates/providers/openai/tests/credential/contract/mod.rs b/backend/crates/providers/openai/tests/credential/contract/mod.rs index e0adae6e8..3684a65e2 100644 --- a/backend/crates/providers/openai/tests/credential/contract/mod.rs +++ b/backend/crates/providers/openai/tests/credential/contract/mod.rs @@ -230,12 +230,21 @@ fn selector_with_runtime( verified_at: chrono::Utc::now(), }); let http = reqwest::Client::builder().build().expect("HTTP client"); + let catalog = Arc::new( + provider_openai::credential::CodexCredentialCatalogService::new( + store.repository(), + profile.clone(), + http.clone(), + OFFICIAL_CODEX_BASE_URL.to_owned(), + crate::support::catalog_cache(), + ), + ); let quota = Arc::new(CodexCredentialQuotaService::new( store.repository(), profile, http, OFFICIAL_CODEX_BASE_URL.to_owned(), - cooldowns, + cooldowns.clone(), crate::support::runtime_policy(), )); CodexCredentialSelector::new( @@ -245,6 +254,8 @@ fn selector_with_runtime( session_affinity, Arc::new(MemorySessionExclusions::default()), quota, + catalog, + cooldowns, account_feedback, CodexCookiePolicy::official().expect("official cookie policy"), ) diff --git a/backend/crates/providers/openai/tests/provider/contract/cookies.rs b/backend/crates/providers/openai/tests/provider/contract/cookies.rs index 6c4929017..33c74f08f 100644 --- a/backend/crates/providers/openai/tests/provider/contract/cookies.rs +++ b/backend/crates/providers/openai/tests/provider/contract/cookies.rs @@ -6,6 +6,7 @@ fn cookie_provider(store: &Arc, base_url: String) -> Arc, base_url: String) -> Arc, base_url: String) -> Arc break error, + Some(Ok(_)) => {} + None => panic!("expected model rejection"), + } + }; + assert_eq!(error.kind(), ProviderErrorKind::Unsupported); + assert!(error.replay_is_safe()); + drop(stream); + let blocked = provider + .clone() + .execute( + planned_request_for_model("openai", operation("gpt-6-astra"), "gpt-6-astra"), + context("req_model_cooled", CancellationToken::new()), + ) + .await + .err() + .expect("model cooldown must survive the first request"); + assert_eq!(blocked.kind(), ProviderErrorKind::NoEligibleAccount); + assert!(cooldowns.cooldowns.lock().unwrap().is_empty()); + server.reset().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200).set_body_raw(CAPTURE_COMPLETED_SSE, "text/event-stream"), + ) + .expect(2) + .mount(&server) + .await; + let mut other = provider + .clone() + .execute( + planned_request_for_model("openai", operation("gpt-6-luna"), "gpt-6-luna"), + context("req_other_model", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = other.next().await { + event.unwrap(); + } + let account = store.account("acct_provider_contract").unwrap(); + let scope = + ProviderCooldownScope::upstream_model(UpstreamModelId::new("gpt-6-astra").unwrap()); + let key = (account.id().clone(), scope.clone()); + { + let mut scoped = cooldowns.scoped.lock().unwrap(); + let until = scoped[&key].until(); + assert!( + (Duration::from_secs(29 * 60)..=Duration::from_secs(30 * 60)) + .contains(&until.duration_since(SystemTime::now()).unwrap()) + ); + scoped.insert( + key, + ProviderScopedCooldown::new( + account.id().clone(), + account.revision(), + scope, + SystemTime::now() - Duration::from_secs(1), + ), + ); + } + let mut recovered = provider + .clone() + .execute( + planned_request_for_model("openai", operation("gpt-6-astra"), "gpt-6-astra"), + context("req_model_recovered", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = recovered.next().await { + event.unwrap(); + } + pool.shutdown().await; + } +} diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 078236cbd..b9fb02062 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -1187,13 +1187,14 @@ pub(crate) fn account_policy() -> gateway_core::account::AccountSelectionPolicy type CooldownReadHook = dyn Fn(ProviderAccountId) -> BoxFuture<'static, Result<(), ProviderStoreError>> + Send + Sync; -/// 内存 `ProviderCooldownPort`:实现 `read`/`put_if_later` 与容量失败计数 -/// 证据(openai selector 429 冷却与容量熔断路径用),scope 变体测试不涉及, -/// 返回占位 +/// 内存账号与模型冷却端口,保留版本和到期时间供选号行为测试 #[derive(Clone, Default)] pub(crate) struct MemoryCooldownPort { pub(crate) cooldowns: Arc>>, + pub(crate) scoped: + Arc>>, before_read: Option>, + before_scoped_read: Option>, } impl MemoryCooldownPort { @@ -1206,6 +1207,11 @@ impl MemoryCooldownPort { self.before_read = Some(hook); self } + + pub(crate) fn with_scoped_read_hook(mut self, hook: Arc) -> Self { + self.before_scoped_read = Some(hook); + self + } } impl ProviderCooldownPort for MemoryCooldownPort { @@ -1254,17 +1260,39 @@ impl ProviderCooldownPort for MemoryCooldownPort { fn put_scoped_if_later( &self, - _cooldown: ProviderScopedCooldown, + cooldown: ProviderScopedCooldown, ) -> BoxFuture<'_, Result> { - Box::pin(async { Ok(false) }) + Box::pin(async move { + let mut scoped = self.scoped.lock().unwrap(); + let key = (cooldown.account_id().clone(), cooldown.scope().clone()); + let should_write = scoped.get(&key).is_none_or(|current| { + current.credential_revision() < cooldown.credential_revision() + || (current.credential_revision() == cooldown.credential_revision() + && current.until() < cooldown.until()) + }); + if should_write { + scoped.insert(key, cooldown); + } + Ok(should_write) + }) } fn read_scoped<'a>( &'a self, - _account_id: &'a ProviderAccountId, - _scope: &'a ProviderCooldownScope, + account_id: &'a ProviderAccountId, + scope: &'a ProviderCooldownScope, ) -> BoxFuture<'a, Result, ProviderStoreError>> { - Box::pin(async { Ok(None) }) + Box::pin(async move { + if let Some(hook) = &self.before_scoped_read { + hook(account_id.clone()).await?; + } + Ok(self + .scoped + .lock() + .unwrap() + .get(&(account_id.clone(), scope.clone())) + .cloned()) + }) } fn clear_scoped<'a>( diff --git a/docs/api.md b/docs/api.md index 2297b0324..7622ff1ff 100644 --- a/docs/api.md +++ b/docs/api.md @@ -551,7 +551,7 @@ xAI 没有 Codex 原生目录,使用官方 CLI proxy 当前目录生成通用 已知模型只路由到明确提供该模型的 Provider,账号不可用时不转发给其他平台的未知模型通道。 未知模型只在当前范围与路由约束下存在唯一 Provider 透传候选时发送;无法确定归属时返回不可用错误, 可通过 Client Key 分组限定目标平台。一个 Provider 内的多个 API Key 账号仍按账号模型政策与调度规则选择, -不要求模型已被上游目录收录;ChatGPT OAuth 账号不接受 Grok 模型 +不要求模型已被上游目录收录;ChatGPT OAuth 账号不接受 Grok 模型,已知模型按[账号模型限制](#账号模型限制)中的套餐资格筛选 目录账号只能来自本次 Client Key 冻结的账号范围。OpenAI 的 OAuth 目录按账号 ID 排序,使用首个成功读取的 合格账号,最多尝试三个账号;API Key 目录按账号模型权限过滤后聚合。同名模型优先采用 OAuth 原生对象, @@ -1172,6 +1172,11 @@ Key 预算及其访问状态随互联默认公开,不要求开启 `shareQuota` 例如 Plus 设置 `allowlist` 并选中 luna 的实际 ID,Pro 设置 `denylist` 并选中同一 ID,即可严格分流; Pro 使用 `all` 时也能参与 luna 调度。套餐名称不自动生成或修改规则 +原生 ChatGPT OAuth Responses 会在评分和亲和前检查最近 30 分钟内同步的套餐目录:模型已在已知套餐中出现时, +不会选中成功读取目录但未包含该模型的套餐账号。未知套餐、目录读取失败或过期、未收录模型仍交由上游验证。 +原生 Responses 上游明确拒绝某个账号使用该模型后,该账号与模型组合冷却 30 分钟;其他模型仍可使用该账号。 +冷却按凭据版本隔离,不修改 `modelAccess`,也不解除原生续写的固定账号约束。受管上游适配器不应用这两项原生限制 + 单账号更新或批量更新省略 `modelAccess` 时保留原值,显式提交 `{ "mode": "all", "models": [] }` 清除限制。 批量接口的调度、分组、模型与代理字段均可省略;省略的字段保持各账号原值。 提供 `groupIds` 时替换完整分组集合,提供 `concurrencyLimit: null` 时恢复继承运行参数 @@ -1181,7 +1186,7 @@ Pro 使用 `all` 时也能参与 luna 调度。套餐名称不自动生成或修 Images 和 Search 按原始请求正文的模型 ID 检查,不依赖模型目录或全局模型映射;可手动添加目录未列出的模型 ID。 模型缺失、重复或无法识别时,仅使用模型政策为 `all` 的账号,不推测上游默认模型。 Search 的选号限制不约束插件在选号后主动改写正文。 -管理员连接测试不受该模型限制;连接测试成功只证明指定账号的上游能力。 +管理员连接测试不受账号模型政策、套餐目录筛选与模型拒绝冷却限制;连接测试成功只证明指定账号的上游能力。 普通 `/v1/models` 和单模型查询将已发现的 OpenAI 模型关联到来源账号,至少一个来源账号在当前 Key 范围内且政策允许时才展示; 同 Provider 中未发现该模型的 `all` 账号不会使它进入列表。原生目录保留已有来源选择和完整模型对象 diff --git a/docs/architecture.md b/docs/architecture.md index 7cd29a4f9..1d2956eb7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -323,7 +323,7 @@ Provider 模型能力、目录代次与 `ProviderCatalogPort` 由 `routing::cata 该只读合同,不反向依赖执行注册表;`ProviderRegistry` 实现目录端口,维护唯一的 Provider 注册集合。已知空目录与查询失败的未知目录保持不同语义,目录替身无需实现请求执行。 Provider 可为模型提供来源账号集合;Core 在公开目录查询时结合冻结账号范围与政策过滤来源, -配置发布复用已缓存的来源事实。来源集合只约束目录展示,不改变发现型目录的推理准入语义 +配置发布复用已缓存的来源事实。来源集合只约束目录展示,不直接作为推理账号白名单;Provider 独占上游资格判断 客户端原生模型目录同样由 Provider 拥有,通过 `ExecutionService` 和现有 Registry 的只读调用传递。 `routing::catalog` 的原生条目只包含模型 ID 与不透明协议正文,Core 负责冻结账号范围和整对象模型映射, @@ -507,7 +507,8 @@ credential state 调度,不以是否存在上游用户 ID 推断可用性;OA 状态恢复和未补齐身份的凭据轮换保留 `unknown`。 API Key 默认 HTTP/SSE,可选 WS 优先;选号先验证传输资格,WS pool 与 continuation 按凭据版本隔离。 OAuth 与 API Key 共用业务请求、响应和能力透传链路,差异限定在上游地址、认证、传输配置及明确的上游请求合同适配。 -OpenAI 模型目录用于发现,不因目录缺项拒绝请求;管理员配置的模型权限仍由 Core 与选号链路执行 +OpenAI 模型目录用于发现;原生 OAuth Responses 还使用新鲜套餐目录筛选已知模型的候选账号。 +未知目录、未知套餐或未收录模型保留上游验证;管理员配置的模型权限仍由 Core 与选号链路执行 - OpenAI 是透明边界。正文兼容、身份归一和显式请求策略之外的业务字段保留原始语义; 业务响应按原始字节转发,客户端错误兼容与原生续写额度恢复按明确例外处理,见 [OpenAI 透传](api.md#openai-透传)。 @@ -733,6 +734,13 @@ OpenAI Provider 在选号时排除 Grok 模型的 ChatGPT OAuth 账号,API Key OpenAI Provider 的账号资格模块统一解释请求类型、认证与套餐限制,Images 和 Live 共用 OAuth 套餐检查。 资格检查复用账号持久事实,在获取租约前与冻结的模型黑白名单共同过滤候选;亲和与重试沿用同一检查 +原生 OAuth Responses 在评分、亲和和租约前按映射后的模型筛选套餐候选:目录快照未满 30 分钟, +且模型已在已知 OAuth 套餐中出现时,排除成功读取目录但未包含该模型的套餐。未知或读取失败的套餐、 +过期快照与未收录模型继续交由上游判断;热路径只读进程缓存,不请求上游目录。 +原生 Responses 明确返回模型不支持时,通过既有模型级冷却端口记录账号、凭据版本与上游模型,30 分钟内不再选用该组合。 +该反馈不冻结整个账号,不改写管理员模型政策,也不占用 WS 传输恢复预算;重试仍遵守交付边界与原生续写固定账号约束。 +管理员诊断旁路这两项限制,受管上游适配器不继承原生目录资格或原生模型拒绝冷却 + 账号编辑在一个事务中更新调度事实,批量更新只应用显式提供的字段。导入与首次 OAuth 可携带统一账号设置,由 Admin 传递给 Store, 与凭据在同一事务中提交;Provider 仍独占凭据解析。未附带设置的导入、重新授权和后台刷新保留已有分组、 权重、并发与模型政策。模型政策变更推进配置 revision,不推进 credential revision。管理端导入和账号编辑共用设置表单,凭据输入独立于设置