fix(openai): respect model capabilities in account selection

Filter known OAuth models by fresh plan catalogs and retain account-scoped model rejections. Preserve credential, transport, adapter, and continuation boundaries, and combine cooldown expiry without dropping protection semantics.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-11-01 22:39:21 +08:00
co-authored by Codex
parent 62516d9769
commit 79df82a4f0
14 changed files with 544 additions and 39 deletions
@@ -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<ProviderAccountId> {
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::<BTreeSet<_>>();
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,
@@ -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<dyn ProviderSessionAffinityPort>,
session_exclusions: Arc<dyn ProviderSessionExclusionPort>,
quota: Arc<CodexCredentialQuotaService>,
catalog: Arc<CodexCredentialCatalogService>,
cooldowns: Arc<dyn ProviderCooldownPort>,
upstream_info: Option<Arc<dyn gateway_core::account::UpstreamInfoSource>>,
cookie_policy: CodexCookiePolicy,
risk_recovery: Mutex<HashMap<String, RiskRecoveryState>>,
@@ -273,6 +279,8 @@ impl CodexCredentialSelector {
session_affinity: Arc<dyn ProviderSessionAffinityPort>,
session_exclusions: Arc<dyn ProviderSessionExclusionPort>,
quota: Arc<CodexCredentialQuotaService>,
catalog: Arc<CodexCredentialCatalogService>,
cooldowns: Arc<dyn ProviderCooldownPort>,
account_feedback: Arc<AccountFeedbackStats>,
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::<Vec<_>>();
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,
@@ -175,6 +175,8 @@ pub async fn initialize(
session_affinity,
session_exclusions,
Arc::clone(&quota),
Arc::clone(&catalog),
ports.cooldowns(),
Arc::clone(&account_feedback),
CodexCookiePolicy::official().map_err(|_| OpenAiInitializeError::CookiePolicy)?,
)
@@ -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())?;
@@ -46,6 +46,7 @@ pub(super) struct MappedProviderFailure {
/// 原始上游错误描述,仅在凭据错误状态下持久化
pub(super) error_message: Option<String>,
pub(super) cyber_policy_failure: bool,
pub(super) model_unsupported: bool,
pub(super) set_cookie_headers: Vec<String>,
pub(super) response_origin: Option<Url>,
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,
@@ -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
@@ -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<MemoryAccountStore>, 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");
}
@@ -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"),
)
@@ -6,6 +6,7 @@ fn cookie_provider(store: &Arc<MemoryAccountStore>, base_url: String) -> Arc<Cod
let profile = wire_profile();
let http = reqwest::Client::builder().no_proxy().build().unwrap();
let leases = Arc::new(TestLeaseCoordinator::default());
let cooldowns = Arc::new(MemoryCooldownPort::new());
let catalog = Arc::new(CodexCredentialCatalogService::new(
store.repository(),
profile.clone(),
@@ -18,7 +19,7 @@ fn cookie_provider(store: &Arc<MemoryAccountStore>, base_url: String) -> Arc<Cod
profile.clone(),
http.clone(),
base_url.clone(),
Arc::new(MemoryCooldownPort::new()),
cooldowns.clone(),
crate::support::runtime_policy(),
));
let feedback = Arc::new(AccountFeedbackStats::default());
@@ -29,6 +30,8 @@ fn cookie_provider(store: &Arc<MemoryAccountStore>, base_url: String) -> Arc<Cod
Arc::new(MemorySessionAffinity::default()),
Arc::new(MemorySessionExclusions::default()),
Arc::clone(&quota),
Arc::clone(&catalog),
cooldowns,
Arc::clone(&feedback),
// 测试仅允许本机模拟上游,生产 Cookie 白名单不放宽
CodexCookiePolicy::new(["cf_clearance"], ["127.0.0.1"]).unwrap(),
@@ -1245,7 +1245,7 @@ fn provider_and_quota_with_runtime_ports(
profile.clone(),
http.clone(),
base_url.clone(),
cooldowns,
cooldowns.clone(),
policy,
));
let account_feedback = Arc::new(AccountFeedbackStats::default());
@@ -1256,6 +1256,8 @@ fn provider_and_quota_with_runtime_ports(
session_affinity,
Arc::new(MemorySessionExclusions::default()),
Arc::clone(&quota),
Arc::clone(&catalog),
cooldowns,
Arc::clone(&account_feedback),
CodexCookiePolicy::official().expect("cookie policy"),
));
@@ -1,4 +1,4 @@
//! Grok 的 ChatGPT OAuth 选号边界与 API Key 未知模型透传合同
//! 验证模型目录、上游拒绝反馈与未知模型的账号选号边界
use super::*;
@@ -139,3 +139,235 @@ async fn unlisted_models_remain_available_without_catalog_discovery() {
);
}
}
#[tokio::test]
async fn known_oauth_models_filter_plan_candidates_before_weight_and_keep_unknown_models() {
let server = MockServer::start().await;
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let mut free = profile("chatgpt-acct_scope_new");
free.plan_type = Some("free".to_owned());
store
.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: "acct_scope_new".to_owned(),
name: "free".to_owned(),
secret: secret("test-free"),
verified_account: free,
next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)),
enabled: true,
})
.await;
store.set_scheduling("acct_scope_new", None, AccountWeight::new(100).unwrap());
for (id, models) in [
(
"acct_scope_new",
json!([{"slug":"gpt-6-luna","display_name":"Luna"}]),
),
(
"acct_provider_contract",
json!([{"slug":"gpt-6-luna","display_name":"Luna"},{"slug":"gpt-6-astra","display_name":"Astra"}]),
),
] {
Mock::given(method("GET"))
.and(path("/codex/models"))
.and(header("chatgpt-account-id", format!("chatgpt-{id}")))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"models":models})))
.expect(1)
.mount(&server)
.await;
}
Mock::given(method("POST"))
.and(path("/codex/responses"))
.respond_with(
ResponseTemplate::new(200).set_body_raw(CAPTURE_COMPLETED_SSE, "text/event-stream"),
)
.expect(3)
.mount(&server)
.await;
let affinity = Arc::new(MemorySessionAffinity::default());
let provider = provider_with_affinity_and_base_url(&store, affinity.clone(), server.uri());
provider.query_model_capabilities().await.unwrap();
for (model, expected) in [
("gpt-6-luna", "acct_scope_new"),
("gpt-6-astra", "acct_provider_contract"),
("custom-unlisted", "acct_scope_new"),
] {
let request_operation = if model == "custom-unlisted" {
operation(model)
} else {
Operation::Generate(GenerateRequest::from_protocol_payload(
ProtocolPayload::json_object(
"openai",
json!({
"model": model, "input": "hello",
"client_metadata": {"x-codex-turn-metadata": json!({
"session_id":"model-session", "thread_id":"model-session"
}).to_string()}
})
.as_object()
.unwrap()
.clone(),
)
.unwrap()
.with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])),
))
};
let mut stream = provider
.clone()
.execute(
planned_request_for_model("openai", request_operation, model),
context("req_plan_model", CancellationToken::new()),
)
.await
.unwrap();
assert_eq!(
stream.metadata().provider_account_id().as_str(),
expected,
"{model}"
);
while let Some(event) = stream.next().await {
event.unwrap();
}
}
assert_eq!(affinity.binding_count(), 1);
let pinned = AttemptContext::new(
RequestAttemptContext::new(
ModelRequestId::new("req_catalog_pin").unwrap(),
ClientApiKeyId::new("key_openai_contract").unwrap(),
),
NonZeroU32::MIN,
SystemTime::now() + Duration::from_secs(30),
account_policy(),
AccountAttemptContext::new(
BTreeSet::new(),
Some(ProviderAccountId::new("acct_scope_new").unwrap()),
None,
)
.with_account_scope(contract_account_scope()),
None,
CancellationToken::new(),
);
let error = provider
.execute(
planned_request_for_model("openai", operation("gpt-6-astra"), "gpt-6-astra"),
pinned,
)
.await
.err()
.expect("a fixed account cannot escape model eligibility");
assert_eq!(error.kind(), ProviderErrorKind::NoEligibleAccount);
}
#[tokio::test]
async fn upstream_model_rejection_cools_only_that_model_across_requests_and_transports() {
use gateway_core::provider_ports::{ProviderCooldownScope, ProviderScopedCooldown};
for websocket in [false, true] {
let server = MockServer::start().await;
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let cooldowns = Arc::new(MemoryCooldownPort::new());
let (provider, pool) = provider_with_cooldowns(&store, server.uri(), cooldowns.clone());
Mock::given(method(if websocket { "GET" } else { "POST" }))
.and(path("/codex/responses"))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({"error":{
"code":"model_not_found", "type":"invalid_request_error",
"message":"The 'gpt-6-astra' model is not supported when using Codex with a ChatGPT account."
}}))).mount(&server).await;
let request_operation = if websocket {
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-6-astra")),
("input".to_owned(), json!("hello")),
]),
)
.unwrap();
Operation::Generate(GenerateRequest::from_protocol_payload(payload))
} else {
operation("gpt-6-astra")
};
let mut stream = provider
.clone()
.execute(
planned_request_for_model("openai", request_operation, "gpt-6-astra"),
context("req_model_rejected", CancellationToken::new()),
)
.await
.unwrap();
let error = loop {
match stream.next().await {
Some(Err(error)) => 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;
}
}
@@ -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<Mutex<BTreeMap<ProviderAccountId, ProviderCooldown>>>,
pub(crate) scoped:
Arc<Mutex<BTreeMap<(ProviderAccountId, ProviderCooldownScope), ProviderScopedCooldown>>>,
before_read: Option<Arc<CooldownReadHook>>,
before_scoped_read: Option<Arc<CooldownReadHook>>,
}
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<CooldownReadHook>) -> 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<bool, ProviderStoreError>> {
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<Option<ProviderScopedCooldown>, 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>(
+7 -2
View File
@@ -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` 账号不会使它进入列表。原生目录保留已有来源选择和完整模型对象
+10 -2
View File
@@ -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。管理端导入和账号编辑共用设置表单,凭据输入独立于设置