mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
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 <noreply@openai.com>
This commit is contained in:
@@ -717,6 +717,7 @@ impl ModelRequestRepository for PgExecutionStore {
|
||||
finalization: ModelRequestFinalization,
|
||||
) -> StoreResult<bool> {
|
||||
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,
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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<bool> {
|
||||
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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -5,6 +5,7 @@ use std::{
|
||||
};
|
||||
|
||||
mod quota_forecast;
|
||||
mod timestamps;
|
||||
|
||||
use chrono::{TimeDelta, Utc};
|
||||
use gateway_admin::{
|
||||
|
||||
@@ -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::<Utc>::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;
|
||||
}
|
||||
@@ -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::<Vec<_>>();
|
||||
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::<BTreeSet<_>>();
|
||||
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
|
||||
|
||||
@@ -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<CodexModelCatalogSnapshot, CodexModelCatalogError> {
|
||||
parse_native_model_catalog(body, etag, OpenAiUpstreamProtocol::Codex)
|
||||
}
|
||||
|
||||
fn parse_native_model_catalog(
|
||||
body: &[u8],
|
||||
etag: Option<&str>,
|
||||
protocol: OpenAiUpstreamProtocol,
|
||||
) -> Result<CodexModelCatalogSnapshot, CodexModelCatalogError> {
|
||||
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<Option<String>, CodexM
|
||||
fn normalize_model(
|
||||
wire: CodexModelWire,
|
||||
document: RawJsonPayload,
|
||||
protocol: OpenAiUpstreamProtocol,
|
||||
) -> Result<CodexCatalogModel, CodexModelCatalogError> {
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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::<serde_json::Value>(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::<Vec<_>>(),
|
||||
["z-unknown-reasoning", "vendor/model"]
|
||||
);
|
||||
let ProviderModelContent::Native(payload) = &models[0].content else {
|
||||
panic!("native model expected")
|
||||
};
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<serde_json::Value>(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::<std::collections::BTreeSet<_>>(),
|
||||
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]
|
||||
|
||||
@@ -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::<Vec<_>>();
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+7
-3
@@ -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 范围。
|
||||
|
||||
@@ -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 与选号链路执行。
|
||||
|
||||
@@ -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 '额度重置'
|
||||
|
||||
@@ -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)
|
||||
<AccountResetCredits
|
||||
v-if="account.provider === 'openai' && account.authenticationKind === 'oauth'"
|
||||
:account="account"
|
||||
@account-updated="emit('accountUpdated', $event)"
|
||||
@consumed="emit('quotaReset', $event)"
|
||||
/>
|
||||
<BaseIconButton
|
||||
variant="ghost"
|
||||
|
||||
@@ -186,6 +186,19 @@ export function useAccountMutations(options: {
|
||||
})
|
||||
}
|
||||
|
||||
async function handleQuotaReset(accountId: string) {
|
||||
try {
|
||||
const result = await refreshAccountQuota({ accountId }, { silent: true })
|
||||
await options.replaceAccount(result.account)
|
||||
}
|
||||
catch (error: unknown) {
|
||||
toast.warning(
|
||||
`额度已重置,但最新额度加载失败:${errorMessage(error, '请手动刷新额度')}`,
|
||||
{ duration: 5000 },
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
async function handleRecover(accountId: string) {
|
||||
await recoveringAccounts.run(accountId, async () => {
|
||||
try {
|
||||
@@ -257,5 +270,6 @@ export function useAccountMutations(options: {
|
||||
handleRecover,
|
||||
handleRefresh,
|
||||
handleRefreshQuota,
|
||||
handleQuotaReset,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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' },
|
||||
)
|
||||
|
||||
@@ -103,6 +103,7 @@ const {
|
||||
handleRecover,
|
||||
handleRefresh,
|
||||
handleRefreshQuota,
|
||||
handleQuotaReset,
|
||||
} = useAccountMutations({
|
||||
onImportTaskCreated: importTasks.created,
|
||||
accounts,
|
||||
@@ -337,7 +338,7 @@ const {
|
||||
<AccountQuotaPanel
|
||||
:account="row"
|
||||
:refreshing="refreshingQuotaAccountIds.has(row.id)"
|
||||
@account-updated="void replaceAccount($event)"
|
||||
@quota-reset="handleQuotaReset"
|
||||
@refresh-quota="handleRefreshQuota"
|
||||
/>
|
||||
<AccountUsagePanel
|
||||
|
||||
Reference in New Issue
Block a user