From b6e17db0c6e5dfd11d11357ef1ea541cf183b6de Mon Sep 17 00:00:00 2001 From: Zyy Date: Sun, 13 Sep 2026 02:02:17 +0800 Subject: [PATCH] fix(openai): preserve native Codex model catalogs Forward complete upstream model objects and client versions without rebuilding metadata. Keep account-scoped catalogs in a bounded, coalesced cache with explicit invalidation and preserve model aliases without mixing fields. Cover service tiers, native metadata, account isolation, cache behavior, and failure handling with regression tests; document the catalog contract. Co-authored-by: Codex --- .../crates/gateway-api/src/openai/models.rs | 60 +++- .../crates/gateway-api/src/openai/service.rs | 11 +- .../crates/gateway-api/tests/openai/models.rs | 156 +++++++- .../gateway-core/src/engine/execution.rs | 87 ++++- .../gateway-core/src/engine/provider.rs | 14 + .../crates/gateway-core/src/operation/mod.rs | 10 +- .../gateway-core/src/routing/catalog.rs | 24 +- .../crates/gateway-core/src/routing/mod.rs | 2 +- .../gateway-core/tests/engine/execution.rs | 102 ++++++ .../openai/src/credential/catalog.rs | 181 +++++++++- .../providers/openai/src/provider/mod.rs | 19 + .../openai/src/provider/observation.rs | 1 + .../providers/openai/src/transport/catalog.rs | 61 +++- .../openai/src/transport/client_sse.rs | 8 +- .../openai/tests/credential/catalog.rs | 335 +++++++++++++++++- .../openai/tests/provider/contract.rs | 41 ++- .../openai/tests/transport/catalog.rs | 119 ++++++- .../fixtures/official_models_snapshot.json | 7 + .../tests/transport/profile_contract.rs | 2 +- docs/api.md | 28 +- docs/architecture.md | 7 + 21 files changed, 1207 insertions(+), 68 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/models.rs b/backend/crates/gateway-api/src/openai/models.rs index 28a79e599..300f4d0be 100644 --- a/backend/crates/gateway-api/src/openai/models.rs +++ b/backend/crates/gateway-api/src/openai/models.rs @@ -6,7 +6,7 @@ use axum::{ http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, }; -use gateway_core::routing::{PublicModelId, PublicModelProfile}; +use gateway_core::routing::{PublicModelDescriptor, PublicModelId, PublicModelProfile}; use serde::Deserialize; use serde_json::{Value, json}; @@ -14,7 +14,7 @@ use crate::ApiState; use super::{ auth::{authenticate_client, client_access_error_response}, - error::model_not_found_response, + error::{model_not_found_response, openai_error_response}, }; const MODEL_CREATED_TIMESTAMP: i64 = 1_700_000_000; @@ -37,19 +37,53 @@ pub(crate) async fn models( Err(error) => return client_access_error_response(error), }; - if query + if let Some(version) = query .client_version .as_deref() - .is_some_and(|version| !version.trim().is_empty()) + .filter(|version| !version.trim().is_empty()) { - let models = service - .public_model_profiles(&client) + if version.len() > 64 + || !version + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b".-+".contains(&byte)) + { + return openai_error_response( + StatusCode::BAD_REQUEST, + "Invalid client_version", + "invalid_request_error", + "invalid_client_version", + ) + .into_response(); + } + let catalog = match service.client_model_catalog(&client, version).await { + Ok(catalog) => catalog, + Err(_) => return catalog_unavailable_response(), + }; + let models = catalog .iter() .enumerate() - .map(|(index, profile)| codex_model_json(profile, index)) - .collect::>(); + .map(|(index, entry)| match entry { + PublicModelDescriptor::Adapted(profile) => Ok(codex_model_json(profile, index)), + PublicModelDescriptor::Native { model, payload } => { + if payload.protocol() != "codex" { + return Err(()); + } + let mut value: Value = + serde_json::from_slice(payload.body()).map_err(|_| ())?; + let object = value.as_object_mut().ok_or(())?; + // 别名只改变请求标识，能力、提示词、顺序等仍由上游目录决定。 + object.insert("slug".to_owned(), Value::String(model.as_str().to_owned())); + Ok(value) + } + }) + .collect::, ()>>(); + let Ok(models) = models else { + return catalog_unavailable_response(); + }; return ( StatusCode::OK, + // 目录经过账号选择/别名/多 Provider 聚合，不能借用上游 ETag 表示改写后的正文。 + [("cache-control", "private, no-store")], Json(json!({ "models": models, })), @@ -72,6 +106,16 @@ pub(crate) async fn models( .into_response() } +fn catalog_unavailable_response() -> Response { + openai_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "Model catalog is temporarily unavailable", + "server_error", + "model_catalog_unavailable", + ) + .into_response() +} + /// `GET /v1/models/{model_id}`。 pub(crate) async fn model_detail( State(state): State, diff --git a/backend/crates/gateway-api/src/openai/service.rs b/backend/crates/gateway-api/src/openai/service.rs index ee9ee6719..874206cb9 100644 --- a/backend/crates/gateway-api/src/openai/service.rs +++ b/backend/crates/gateway-api/src/openai/service.rs @@ -10,7 +10,7 @@ use gateway_core::engine::execution::{ }; use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::lifecycle::{ConnectionDraining, ConnectionGuard, ConnectionLifecycle}; -use gateway_core::routing::{PublicModelId, PublicModelProfile}; +use gateway_core::routing::{ProviderCatalogUnavailable, PublicModelDescriptor, PublicModelId}; use uuid::Uuid; use super::auth::ClientApiKeyAuthError; @@ -52,11 +52,14 @@ impl OpenAiService { .collect() } - pub(crate) fn public_model_profiles( + pub(crate) async fn client_model_catalog( &self, client: &AuthenticatedClient, - ) -> Vec { - self.execution.public_model_profiles(client) + client_version: &str, + ) -> Result, ProviderCatalogUnavailable> { + self.execution + .client_model_catalog(client, "codex", client_version) + .await } pub(crate) fn contains_public_model( diff --git a/backend/crates/gateway-api/tests/openai/models.rs b/backend/crates/gateway-api/tests/openai/models.rs index 153b764f2..0df1bc7be 100644 --- a/backend/crates/gateway-api/tests/openai/models.rs +++ b/backend/crates/gateway-api/tests/openai/models.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use axum::{ body::{Body, to_bytes}, @@ -11,7 +11,8 @@ use gateway_core::engine::execution::{ }; use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::routing::{ - ModelPresentation, ModelServiceTier, PublicModelId, PublicModelProfile, + ModelPresentation, ModelServiceTier, ProviderCatalogUnavailable, PublicModelDescriptor, + PublicModelId, PublicModelProfile, }; use tower::ServiceExt; @@ -20,6 +21,8 @@ use super::{api_router, authenticated_client, authenticated_client_with_min_vers pub(super) struct ModelsExecution { client: AuthenticatedClient, presentation: Option, + native: Option, ProviderCatalogUnavailable>>, + requested_versions: Mutex>, } impl ModelsExecution { @@ -27,6 +30,8 @@ impl ModelsExecution { Arc::new(Self { client: authenticated_client("sk_models_test"), presentation: None, + native: None, + requested_versions: Mutex::default(), }) } @@ -61,6 +66,8 @@ impl ModelsExecution { Arc::new(Self { client: authenticated_client("sk_models_test"), presentation: Some(presentation), + native: None, + requested_versions: Mutex::default(), }) } @@ -68,6 +75,8 @@ impl ModelsExecution { Arc::new(Self { client: authenticated_client_with_min_versions("sk_models_test", None, Some("0.40.0")), presentation: None, + native: None, + requested_versions: Mutex::default(), }) } @@ -79,11 +88,48 @@ impl ModelsExecution { None, ), presentation: None, + native: None, + requested_versions: Mutex::default(), }) } + + fn profiles(&self) -> Vec { + self.presentation + .as_ref() + .map(|presentation| { + PublicModelProfile::new( + PublicModelId::new("grok-4.5").expect("model"), + presentation.clone(), + ) + }) + .into_iter() + .collect() + } } impl ExecutionService for ModelsExecution { + fn client_model_catalog<'a>( + &'a self, + _client: &'a AuthenticatedClient, + protocol: &'a str, + version: &'a str, + ) -> BoxFuture<'a, Result, ProviderCatalogUnavailable>> { + Box::pin(async move { + assert_eq!(protocol, "codex"); + self.requested_versions + .lock() + .expect("versions") + .push(version.to_owned()); + self.native.clone().unwrap_or_else(|| { + Ok(self + .profiles() + .into_iter() + .map(PublicModelDescriptor::Adapted) + .collect()) + }) + }) + } + fn authenticate( &self, plaintext: &str, @@ -102,16 +148,6 @@ impl ExecutionService for ModelsExecution { .collect() } - fn public_model_profiles(&self, _: &AuthenticatedClient) -> Vec { - let Some(presentation) = &self.presentation else { - return Vec::new(); - }; - vec![PublicModelProfile::new( - PublicModelId::new("grok-4.5").expect("model"), - presentation.clone(), - )] - } - fn contains_public_model(&self, _: &AuthenticatedClient, model: &PublicModelId) -> bool { matches!(model.as_str(), "model-a" | "model-b") } @@ -145,6 +181,102 @@ fn authorized_request(path: &str) -> Request { .expect("build models request") } +#[tokio::test] +async fn native_catalog_preserves_complete_objects_and_only_rewrites_alias_slug() { + let original = serde_json::json!({ + "slug": "gpt-native", "display_name": "Official name", "priority": 47, + "base_instructions": "Official instructions\nDo not replace.", + "model_messages": {"token_budget": "upstream template", "future_section": [null, {"a": true}]}, + "service_tiers": [{"id": "priority", "name": "Fast", "description": "Official speed tier"}], + "tool_mode": "code_mode_only", "use_responses_lite": true, + "future_field": {"nested": [1, null, "value"]}, "explicit_null": null + }); + let entry = |model: &str| PublicModelDescriptor::Native { + model: PublicModelId::new(model).expect("model"), + payload: gateway_core::operation::RawJsonPayload::new( + "codex", + serde_json::to_vec(&original).expect("JSON").into(), + ) + .expect("payload"), + }; + let execution = Arc::new(ModelsExecution { + client: authenticated_client("sk_models_test"), + presentation: None, + native: Some(Ok(vec![entry("gpt-native"), entry("my-alias")])), + requested_versions: Mutex::default(), + }); + let response = api_router(execution.clone()) + .await + .oneshot(authorized_request( + "/v1/models?client_version=0.154.0-alpha.3", + )) + .await + .expect("response"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["cache-control"], "private, no-store"); + assert!(response.headers().get("etag").is_none()); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + let value: serde_json::Value = serde_json::from_slice(&body).expect("JSON"); + assert_eq!(value["models"][0], original); + let mut alias = original; + alias["slug"] = serde_json::json!("my-alias"); + assert_eq!(value["models"][1], alias); + assert_eq!( + *execution.requested_versions.lock().expect("versions"), + ["0.154.0-alpha.3"] + ); +} + +#[tokio::test] +async fn native_catalog_failure_does_not_publish_a_synthetic_or_empty_success() { + let execution = Arc::new(ModelsExecution { + client: authenticated_client("sk_models_test"), + presentation: None, + native: Some(Err(ProviderCatalogUnavailable)), + requested_versions: Mutex::default(), + }); + let response = api_router(execution) + .await + .oneshot(authorized_request("/v1/models?client_version=0.154.0")) + .await + .expect("response"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + let value: serde_json::Value = serde_json::from_slice(&body).expect("JSON"); + assert_eq!(value["error"]["code"], "model_catalog_unavailable"); + assert!(value.get("models").is_none()); +} + +#[tokio::test] +async fn catalog_rejects_unbounded_or_control_character_versions_before_fetching() { + for version in [ + "x".repeat(65), + "0.154.0%0A".to_owned(), + "0.154.0%20".to_owned(), + ] { + let execution = ModelsExecution::new(); + let response = api_router(execution.clone()) + .await + .oneshot(authorized_request(&format!( + "/v1/models?client_version={version}" + ))) + .await + .expect("response"); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert!( + execution + .requested_versions + .lock() + .expect("versions") + .is_empty() + ); + } +} + #[tokio::test] async fn models_should_encode_the_service_visible_catalog() { let response = api_router(ModelsExecution::new()) diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 1d60f6046..987f02550 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -37,7 +37,8 @@ use crate::lifecycle::CancellationToken; use crate::operation::{Operation, ProviderSessionState}; use crate::policy::{ClientApiKeyId, ClientPolicy}; use crate::routing::{ - PublicModelId, PublicModelProfile, RoutingContext, RuntimeSnapshot, UpstreamModelId, + ProviderCatalogUnavailable, PublicModelDescriptor, PublicModelId, RoutingContext, + RuntimeSnapshot, UpstreamModelId, }; use crate::runtime::RuntimeSnapshotHandle; @@ -187,8 +188,13 @@ pub trait ExecutionService: Send + Sync { plaintext: &str, ) -> Result; fn public_models(&self, client: &AuthenticatedClient) -> Vec; - fn public_model_profiles(&self, _client: &AuthenticatedClient) -> Vec { - Vec::new() + fn client_model_catalog<'a>( + &'a self, + _client: &'a AuthenticatedClient, + _protocol: &'a str, + _client_version: &'a str, + ) -> BoxFuture<'a, Result, ProviderCatalogUnavailable>> { + Box::pin(async { Err(ProviderCatalogUnavailable) }) } fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool; fn start( @@ -920,10 +926,77 @@ impl ExecutionService for DefaultExecutionService { .public_models_for_scope(client.policy.account_scope()) } - fn public_model_profiles(&self, client: &AuthenticatedClient) -> Vec { - client - .snapshot - .public_model_profiles_for_scope(client.policy.account_scope()) + fn client_model_catalog<'a>( + &'a self, + client: &'a AuthenticatedClient, + protocol: &'a str, + client_version: &'a str, + ) -> BoxFuture<'a, Result, ProviderCatalogUnavailable>> { + Box::pin(async move { + let scope = client.policy.account_scope(); + let mut result = Vec::new(); + let mut seen = BTreeSet::new(); + for kind in scope.provider_kinds() { + let provider = self.providers.get(kind).ok_or(ProviderCatalogUnavailable)?; + let Some(models) = provider + .query_client_model_catalog(scope, protocol, client_version) + .await? + else { + for profile in client.snapshot.public_model_profiles_for_provider(kind) { + if seen.insert(profile.model().clone()) { + result.push(PublicModelDescriptor::Adapted(profile)); + } + } + continue; + }; + // 保留上游顺序；映射只选择完整的目标对象，不能跨账号/模型混拼字段。 + let by_id = models + .iter() + .map(|entry| (entry.model.as_str(), entry)) + .collect::>(); + for entry in &models { + let target = client.snapshot.mapped_model(entry.model.as_str()); + let Some(source) = by_id.get(target.as_str()) else { + continue; + }; + let model = PublicModelId::new(entry.model.as_str().to_owned()) + .map_err(|_| ProviderCatalogUnavailable)?; + // 原生目录可能比路由快照更新，不能向客户端公布当前已知不可路由的模型。 + if !client + .snapshot + .contains_public_model_for_provider(&model, kind) + { + continue; + } + if seen.insert(model.clone()) { + result.push(PublicModelDescriptor::Native { + model, + payload: source.payload.clone(), + }); + } + } + for model in client.snapshot.public_models_for_provider(kind) { + let target = client.snapshot.mapped_model(model.as_str()); + if target == model.as_str() || seen.contains(&model) { + continue; + } + if !client + .snapshot + .contains_public_model_for_provider(&model, kind) + { + continue; + } + if let Some(source) = by_id.get(target.as_str()) { + seen.insert(model.clone()); + result.push(PublicModelDescriptor::Native { + model, + payload: source.payload.clone(), + }); + } + } + } + Ok(result) + }) } fn contains_public_model(&self, client: &AuthenticatedClient, model: &PublicModelId) -> bool { diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index 82bff0384..e0352bc6c 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -503,6 +503,20 @@ pub trait Provider: Send + Sync { &self, ) -> Result, ProviderError>; + /// 读取当前客户端协议的原生目录，必须限定到认证时冻结的账号范围。 + /// `None` 表示不提供该协议的原生目录；读取失败不能伪装成不支持。 + async fn query_client_model_catalog( + &self, + _scope: &crate::account::scope::FrozenAccountScope, + _protocol: &str, + _client_version: &str, + ) -> Result< + Option>, + crate::routing::ProviderCatalogUnavailable, + > { + Ok(None) + } + /// 选择一个未被排除的资源并返回 cold [`ProviderStream`]。 /// /// 返回成功、返回错误或准备 future 被取消前，均不得发送本次请求的上游握手或业务 diff --git a/backend/crates/gateway-core/src/operation/mod.rs b/backend/crates/gateway-core/src/operation/mod.rs index cee8a8cd0..31d438071 100644 --- a/backend/crates/gateway-core/src/operation/mod.rs +++ b/backend/crates/gateway-core/src/operation/mod.rs @@ -225,10 +225,10 @@ impl fmt::Debug for ProtocolPayload { } } -/// 客户端协议交给 Provider 的未经重编码 JSON 正文。 +/// 协议 adapter 与 Provider 之间不透明传递的 JSON 正文。 /// -/// 协议 adapter 只确定路由端点；Core 保留原始字节与非 wire 上下文， -/// 不验证、解析或改写正文。 +/// Core 保留字节与非 wire 上下文，不验证、解析或改写正文； +/// 请求载荷与原生模型目录均由协议 adapter 和对应 Provider 共同拥有语义。 #[derive(Clone, PartialEq, Eq)] pub struct RawJsonPayload { protocol: String, @@ -237,7 +237,7 @@ pub struct RawJsonPayload { } impl RawJsonPayload { - /// 创建未经重编码的协议 JSON 正文。 + /// 创建不透明的协议 JSON 正文。 /// /// # Errors /// @@ -260,7 +260,7 @@ impl RawJsonPayload { &self.protocol } - /// 返回 adapter 收到的原始 JSON 字节。 + /// 返回协议正文的 JSON 字节。 #[must_use] pub const fn body(&self) -> &Bytes { &self.body diff --git a/backend/crates/gateway-core/src/routing/catalog.rs b/backend/crates/gateway-core/src/routing/catalog.rs index 263f1f36e..bfbeb220c 100644 --- a/backend/crates/gateway-core/src/routing/catalog.rs +++ b/backend/crates/gateway-core/src/routing/catalog.rs @@ -1,12 +1,32 @@ -//! 快照编译需要的 Provider 目录事实，不暴露执行或账号选择能力。 +//! 快照编译与客户端读取需要的 Provider 目录合同，不暴露执行或账号选择能力。 use std::collections::BTreeMap; use futures::future::BoxFuture; use crate::identity::ProviderKind; +use crate::operation::RawJsonPayload; -use super::{ModelCapabilities, ModelPresentation, UpstreamModelId}; +use super::{ + ModelCapabilities, ModelPresentation, PublicModelId, PublicModelProfile, UpstreamModelId, +}; + +/// Provider 原生客户端目录条目；Core 只解释模型标识，不解释协议正文。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProviderModelDescriptor { + pub model: UpstreamModelId, + pub payload: RawJsonPayload, +} + +/// 对外目录保留原生正文；只有没有原生目录的 Provider 才使用通用画像适配。 +#[derive(Debug, Clone)] +pub enum PublicModelDescriptor { + Native { + model: PublicModelId, + payload: RawJsonPayload, + }, + Adapted(PublicModelProfile), +} /// Provider 实时目录编译后的单模型能力。 #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index 7923f4df2..e2c32969c 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -10,7 +10,7 @@ pub use crate::account::scope::{ pub use crate::identity::ProviderKind; pub use catalog::{ ProviderCatalogGeneration, ProviderCatalogPort, ProviderCatalogUnavailable, - ProviderModelCapabilities, + ProviderModelCapabilities, ProviderModelDescriptor, PublicModelDescriptor, }; pub use snapshot::RuntimeSnapshot; diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index 1731eced5..17810a67c 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -996,6 +996,108 @@ fn probe_observation_store_failure_preserves_the_provider_error() { struct FailingProvider; +struct NativeCatalogProvider { + fail: bool, +} + +#[async_trait] +impl Provider for NativeCatalogProvider { + fn name(&self) -> &'static str { + "openai" + } + fn catalog_generation(&self) -> ProviderCatalogGeneration { + ProviderCatalogGeneration::default() + } + async fn query_model_capabilities( + &self, + ) -> Result, ProviderError> { + Ok(Vec::new()) + } + async fn query_client_model_catalog( + &self, + scope: &FrozenAccountScope, + protocol: &str, + version: &str, + ) -> Result< + Option>, + gateway_core::routing::ProviderCatalogUnavailable, + > { + assert!(scope.allows(&ProviderAccountId::new("acct_start").expect("account"))); + assert!(!scope.allows(&ProviderAccountId::new("acct_other").expect("account"))); + assert_eq!((protocol, version), ("codex", "0.154.0")); + if self.fail { + return Err(gateway_core::routing::ProviderCatalogUnavailable); + } + Ok(Some( + ["gpt-new", "gpt-start"] + .into_iter() + .map(|id| gateway_core::routing::ProviderModelDescriptor { + model: UpstreamModelId::new(id).expect("id"), + payload: RawJsonPayload::new( + "codex", + serde_json::to_vec(&json!({"slug":id,"future":null})) + .expect("JSON") + .into(), + ) + .expect("payload"), + }) + .collect(), + )) + } + async fn execute( + &self, + _: ProviderRequest, + _: AttemptContext, + ) -> Result { + panic!("catalog must not use metered execution") + } +} + +#[test] +fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models() { + for fail in [false, true] { + let snapshot = start_snapshot().with_model_mappings(BTreeMap::from([ + ("alias".to_owned(), "gpt-start".to_owned()), + ("missing-alias".to_owned(), "unavailable-model".to_owned()), + ])); + let service = DefaultExecutionService::new( + RuntimeSnapshotHandle::new(snapshot), + Arc::new(TrackingExecutionStore::default()), + ProviderRegistry::new([Arc::new(NativeCatalogProvider { fail }) as Arc]) + .expect("registry"), + Arc::new(UnusedAdmissions), + Arc::new(UnusedCircuits), + Arc::new(UnusedContinuation), + Arc::new(RecordingClientApiKeyUsage::default()), + ); + let client = service.authenticate("sk_start_test").expect("authenticate"); + let result = block_on(service.client_model_catalog(&client, "codex", "0.154.0")); + if fail { + assert!( + result.is_err(), + "native failure cannot use the global synthesized catalog" + ); + continue; + } + let models = result.expect("models"); + let pairs = models + .iter() + .map(|entry| { + let gateway_core::routing::PublicModelDescriptor::Native { model, payload } = entry + else { + panic!("native required") + }; + (model.as_str(), payload.body()) + }) + .collect::>(); + assert_eq!( + pairs.iter().map(|(id, _)| *id).collect::>(), + ["gpt-start", "alias"] + ); + assert_eq!(pairs[0].1, pairs[1].1, "Core preserves alias payload bytes"); + } +} + #[async_trait] impl Provider for FailingProvider { fn name(&self) -> &'static str { diff --git a/backend/crates/providers/openai/src/credential/catalog.rs b/backend/crates/providers/openai/src/credential/catalog.rs index 5781202e6..335fb980c 100644 --- a/backend/crates/providers/openai/src/credential/catalog.rs +++ b/backend/crates/providers/openai/src/credential/catalog.rs @@ -7,25 +7,32 @@ use std::time::{Duration, SystemTime}; use chrono::{DateTime, Utc}; use gateway_core::account::{ - CredentialState, OpaqueProviderData, ProviderAccount, ProviderAccountId, + CredentialRevision, CredentialState, OpaqueProviderData, ProviderAccount, ProviderAccountId, }; use gateway_core::provider_ports::{ ProviderCatalogCacheKey, ProviderCatalogCachePort, ProviderCatalogScope, }; -use gateway_core::routing::{ProviderCatalogGeneration, ProviderKind, UpstreamModelId}; +use gateway_core::routing::{ + FrozenAccountScope, ProviderCatalogGeneration, ProviderKind, ProviderModelDescriptor, + UpstreamModelId, +}; use secrecy::ExposeSecret; use thiserror::Error; -use tokio::sync::Notify; +use tokio::sync::{Notify, OnceCell}; +use tokio::time::Instant; use uuid::Uuid; use super::repository::{CodexCredentialRepository, CredentialRepositoryError}; -use crate::transport::profile::CodexWireProfileState; +use crate::transport::profile::{CodexWireProfile, CodexWireProfileState}; use crate::transport::{CodexBackendClient, CodexCatalogModel, CodexRequestContext}; const MAX_RESPONSE_ETAG_BYTES: usize = 256; const PLAN_CATALOG_CACHE_TTL: Duration = Duration::from_secs(5 * 60); const MAX_CATALOG_FETCH_ATTEMPTS: usize = 3; const MAX_PLAN_CATALOG_MODELS: usize = 2_048; +const MAX_CLIENT_CATALOGS: usize = 32; +const CLIENT_CATALOG_TIMEOUT: Duration = Duration::from_secs(15); +const CLIENT_CATALOG_ERROR_TTL: Duration = Duration::from_secs(5); /// OpenAI 以套餐划分的模型目录作用域。 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] @@ -128,7 +135,7 @@ impl fmt::Debug for CodexCredentialCatalogSnapshot { } } -#[derive(Debug, Error)] +#[derive(Debug, Clone, Error)] pub enum CodexCredentialCatalogError { #[error("Codex model catalog has no eligible account")] NoEligibleCredential, @@ -158,6 +165,44 @@ struct CatalogCacheState { snapshot: Option, } +#[derive(PartialEq, Eq)] +struct ClientCatalogKey { + account_id: ProviderAccountId, + revision: CredentialRevision, + plan: Option, + upstream_account_id: Option, + client_version: String, + profile: CodexWireProfile, +} + +type ClientCatalogResult = Result, CodexCredentialCatalogError>; + +struct CachedClientCatalog { + // 负缓存从请求结束计时，不能让一次慢失败耗尽后续请求的退避窗口。 + completed_at: Instant, + result: ClientCatalogResult, +} + +struct ClientCatalogEntry { + key: ClientCatalogKey, + created_at: Instant, + value: Arc>, +} + +impl ClientCatalogEntry { + fn is_fresh(&self) -> bool { + let Some(value) = self.value.get() else { + return self.created_at.elapsed() < PLAN_CATALOG_CACHE_TTL; + }; + let ttl = if value.result.is_err() { + CLIENT_CATALOG_ERROR_TTL + } else { + PLAN_CATALOG_CACHE_TTL + }; + value.completed_at.elapsed() < ttl + } +} + struct FetchedAccountModels { models: Vec, etag: Option, @@ -182,6 +227,7 @@ pub struct CodexCredentialCatalogService { base_url: String, plan_catalog_cache: Arc, cache: Arc>, + client_catalogs: Arc>>, etags: Arc>, etag_notification: Arc, } @@ -201,11 +247,102 @@ impl CodexCredentialCatalogService { base_url, plan_catalog_cache, cache: Arc::new(RwLock::new(CatalogCacheState::default())), + client_catalogs: Arc::new(Mutex::new(Vec::new())), etags: Arc::new(Mutex::new(CatalogEtagState::default())), etag_notification: Arc::new(Notify::new()), } } + /// 原生客户端目录使用 Key 范围内排序稳定的首个成功账号，不复用套餐级画像。 + /// 同一账号/凭据版本/客户端版本的并发读取合并；原生正文仅保留在有界进程缓存。 + pub async fn client_model_catalog( + &self, + scope: &FrozenAccountScope, + client_version: &str, + ) -> ClientCatalogResult { + let now = SystemTime::now(); + let mut accounts = self.repository.list_for_provider().await?; + accounts + .retain(|account| scope.allows(account.id()) && eligible_catalog_account(account, now)); + accounts.sort_by(|left, right| left.id().cmp(right.id())); + let mut last_error = CodexCredentialCatalogError::NoEligibleCredential; + for account in accounts.iter().take(MAX_CATALOG_FETCH_ATTEMPTS) { + let profile = self.profile.snapshot(); + 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 = tokio::time::timeout( + CLIENT_CATALOG_TIMEOUT, + self.fetch_account_models(&client, account, Some(client_version)), + ) + .await + .map_err(|_| CodexCredentialCatalogError::Upstream { + detail: "client model catalog timed out".to_owned(), + }) + .and_then(|result| result) + .map(|fetched| { + fetched + .models + .into_iter() + .map(|model| ProviderModelDescriptor { + model: model.request_model().clone(), + payload: model.document().clone(), + }) + .collect() + }); + CachedClientCatalog { + completed_at: Instant::now(), + result, + } + }) + .await; + match &result.result { + Ok(models) => return Ok(models.clone()), + Err(error) => last_error = error.clone(), + } + } + Err(last_error) + } + #[must_use] pub fn catalog_generation(&self) -> ProviderCatalogGeneration { let cache = self @@ -377,7 +514,7 @@ impl CodexCredentialCatalogService { ) -> Result { let mut last_error = None; for account in candidates.into_iter().take(MAX_CATALOG_FETCH_ATTEMPTS) { - match self.fetch_account_models(client, &account).await { + match self.fetch_account_models(client, &account, None).await { Ok(fetched) => return Ok(fetched), Err(error) => last_error = Some(error), } @@ -423,6 +560,7 @@ impl CodexCredentialCatalogService { &self, client: &CodexBackendClient, account: &ProviderAccount, + client_version: Option<&str>, ) -> Result { let credential = self .repository @@ -439,12 +577,15 @@ impl CodexCredentialCatalogService { .map_err(|error| CodexCredentialCatalogError::Upstream { detail: error.to_string(), })? - .fetch_models_with_context(CodexRequestContext::auxiliary( - authorization.expose_secret(), - account.upstream_account_id(), - &request_id, - None, - )) + .fetch_models_with_context( + CodexRequestContext::auxiliary( + authorization.expose_secret(), + account.upstream_account_id(), + &request_id, + None, + ), + client_version, + ) .await; let snapshot = result.map_err(|error| CodexCredentialCatalogError::Upstream { detail: error.to_string(), @@ -461,9 +602,19 @@ impl CodexCredentialCatalogService { .write() .map_err(|_| CodexCredentialCatalogError::Cache)?; if cache.snapshot.take().is_some() { - cache.revision = cache.revision.saturating_add(1); cache.generation = cache.generation.saturating_add(1); } + cache.revision = cache.revision.saturating_add(1); + drop(cache); + self.invalidate_client_catalogs()?; + Ok(()) + } + + fn invalidate_client_catalogs(&self) -> Result<(), CodexCredentialCatalogError> { + self.client_catalogs + .lock() + .map_err(|_| CodexCredentialCatalogError::Cache)? + .clear(); Ok(()) } @@ -486,6 +637,7 @@ impl CodexCredentialCatalogService { } }; if changed { + self.invalidate_client_catalogs()?; self.etag_notification.notify_one(); } Ok(changed) @@ -613,6 +765,9 @@ impl CodexCredentialCatalogService { cache.generation = cache.generation.saturating_add(1); } drop(cache); + if changed { + self.invalidate_client_catalogs()?; + } self.record_applied_catalog_etags(fetched.etags)?; Ok(snapshot) } diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 2f27b344a..8e8b0979d 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -291,6 +291,25 @@ impl Provider for CodexProvider { .collect()) } + async fn query_client_model_catalog( + &self, + scope: &gateway_core::account::scope::FrozenAccountScope, + protocol: &str, + client_version: &str, + ) -> Result< + Option>, + gateway_core::routing::ProviderCatalogUnavailable, + > { + if protocol != "codex" { + return Ok(None); + } + self.catalog + .client_model_catalog(scope, client_version) + .await + .map(Some) + .map_err(|_| gateway_core::routing::ProviderCatalogUnavailable) + } + async fn execute( &self, request: ProviderRequest, diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 01e4b7e52..1a399fe34 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -615,6 +615,7 @@ pub(super) fn codex_model_presentation(model: &CodexCatalogModel) -> ModelPresen capabilities.image_detail_original() == CodexCatalogCapabilityEvidence::DeclaredNative, ) .with_verbosity(capabilities.verbosity() == CodexCatalogCapabilityEvidence::DeclaredNative) + .with_service_tiers(model.metadata().service_tiers().to_vec()) .with_hidden(hidden) } diff --git a/backend/crates/providers/openai/src/transport/catalog.rs b/backend/crates/providers/openai/src/transport/catalog.rs index 300f33b45..7b958f1d0 100644 --- a/backend/crates/providers/openai/src/transport/catalog.rs +++ b/backend/crates/providers/openai/src/transport/catalog.rs @@ -3,7 +3,9 @@ use std::collections::BTreeSet; use std::num::NonZeroU64; -use gateway_core::routing::UpstreamModelId; +use bytes::Bytes; +use gateway_core::operation::RawJsonPayload; +use gateway_core::routing::{ModelServiceTier, UpstreamModelId}; use reqwest::header::{ETAG, HeaderMap}; use serde::Deserialize; @@ -13,6 +15,7 @@ pub const MAX_CODEX_MODEL_CATALOG_BYTES: usize = 1024 * 1024; const MAX_CATALOG_MODELS: usize = 2_048; const MAX_DISPLAY_NAME_BYTES: usize = 256; const MAX_DESCRIPTION_BYTES: usize = 4 * 1024; +const MAX_SERVICE_TIER_ID_BYTES: usize = 64; const MAX_ETAG_BYTES: usize = 256; /// 上游目录对一项能力给出的明确证据。 @@ -133,6 +136,7 @@ pub struct CodexCatalogMetadata { description: Option, priority: Option, visibility: Option, + service_tiers: Vec, } impl CodexCatalogMetadata { @@ -153,6 +157,12 @@ impl CodexCatalogMetadata { pub const fn visibility(&self) -> Option { self.visibility } + + /// 返回上游明确声明的服务档位；缺失时为空，不根据模型名补齐。 + #[must_use] + pub fn service_tiers(&self) -> &[ModelServiceTier] { + &self.service_tiers + } } /// Codex 官方 picker 可见性。 @@ -173,6 +183,7 @@ pub enum CodexCatalogVisibility { /// 一个已完整校验的 Codex 真实模型。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct CodexCatalogModel { + document: RawJsonPayload, request_model: UpstreamModelId, display_name: String, capabilities: CodexCatalogCapabilities, @@ -181,6 +192,12 @@ pub struct CodexCatalogModel { } impl CodexCatalogModel { + /// 完整原生对象只用于客户端目录，不写入控制面画像或日志。 + #[must_use] + pub const fn document(&self) -> &RawJsonPayload { + &self.document + } + /// 返回实际写入上游请求的模型 slug。 #[must_use] pub const fn request_model(&self) -> &UpstreamModelId { @@ -270,7 +287,7 @@ pub enum CodexModelCatalogError { #[derive(Debug, Deserialize)] struct CodexModelsWire { - models: Vec, + models: Vec, } #[derive(Debug, Deserialize)] @@ -289,6 +306,15 @@ struct CodexModelWire { context_window: Option, max_context_window: Option, input_modalities: Option>, + #[serde(default)] + service_tiers: Vec, +} + +#[derive(Debug, Deserialize)] +struct CodexServiceTierWire { + id: String, + name: String, + description: String, } #[derive(Debug, Deserialize)] @@ -309,7 +335,7 @@ enum CodexInputModalityWire { /// /// # Errors /// -/// 任一条目、重复 slug 或 ETag 不合法时整轮失败；上游新增的未知字段会忽略。 +/// 任一条目、重复 slug 或 ETag 不合法时整轮失败；未知字段保留在原生正文中。 pub fn parse_codex_model_catalog( body: &[u8], etag: Option<&str>, @@ -330,7 +356,16 @@ pub fn parse_codex_model_catalog( let mut seen = BTreeSet::new(); let mut models = Vec::with_capacity(wire.models.len()); for model in wire.models { - let model = normalize_model(model)?; + let wire = + CodexModelWire::deserialize(&model).map_err(|_| CodexModelCatalogError::InvalidWire)?; + let document = RawJsonPayload::new( + "codex", + Bytes::from( + serde_json::to_vec(&model).map_err(|_| CodexModelCatalogError::InvalidWire)?, + ), + ) + .map_err(|_| CodexModelCatalogError::InvalidWire)?; + let model = normalize_model(wire, document)?; if !seen.insert(model.request_model.as_str().to_owned()) { return Err(CodexModelCatalogError::DuplicateModelSlug); } @@ -351,7 +386,10 @@ pub(super) fn catalog_etag(headers: &HeaderMap) -> Result, CodexM .transpose() } -fn normalize_model(wire: CodexModelWire) -> Result { +fn normalize_model( + wire: CodexModelWire, + document: RawJsonPayload, +) -> Result { if !valid_model_slug(&wire.slug) { return Err(CodexModelCatalogError::InvalidModelSlug); } @@ -361,6 +399,17 @@ fn normalize_model(wire: CodexModelWire) -> Result, CodexModelCatalogError>>()?; let context_window_tokens = optional_positive(wire.context_window)?; let max_context_window_tokens = optional_positive(wire.max_context_window)?; @@ -382,6 +431,7 @@ fn normalize_model(wire: CodexModelWire) -> Result Result, + client_version: Option<&str>, ) -> CodexClientResult { let endpoint = endpoint_url(&self.base_url, "codex/models"); let profile = self.profile.snapshot(); @@ -506,7 +507,10 @@ impl CodexBackendClient { let response = self .client .get(endpoint) - .query(&[("client_version", profile.codex_version.as_str())]) + .query(&[( + "client_version", + client_version.unwrap_or(profile.codex_version.as_str()), + )]) .headers(headers) .send() .await?; diff --git a/backend/crates/providers/openai/tests/credential/catalog.rs b/backend/crates/providers/openai/tests/credential/catalog.rs index f87c80c77..4a40e1928 100644 --- a/backend/crates/providers/openai/tests/credential/catalog.rs +++ b/backend/crates/providers/openai/tests/credential/catalog.rs @@ -5,12 +5,16 @@ use chrono::{TimeZone as _, Utc}; use futures::executor::block_on; use gateway_core::account::{ProviderAccount, ProviderAccountId}; use gateway_core::provider_ports::ProviderCatalogCachePort; +use gateway_core::routing::{ + ClientRoutingScope, FrozenAccountScope, ModelServiceTier, ProviderKind, RuntimeAccount, + RuntimeAccountDirectory, +}; use provider_openai::OFFICIAL_CODEX_BASE_URL; use provider_openai::credential::{ CodexCredentialCatalogError, CodexCredentialCatalogService, ImportCodexOAuthCredential, }; use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState}; -use wiremock::matchers::{method, path}; +use wiremock::matchers::{header, method, path, query_param}; use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate}; use crate::support::{MemoryAccountStore, catalog_cache, profile, secret}; @@ -113,6 +117,289 @@ async fn seed_account_with_plan( store.account(account_id).expect("seeded account") } +fn client_scope(accounts: &[ProviderAccount]) -> FrozenAccountScope { + FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new( + accounts + .iter() + .map(|account| { + ( + account.id().clone(), + RuntimeAccount::new( + ProviderKind::new("openai").expect("provider"), + Default::default(), + ), + ) + }) + .collect(), + )), + ClientRoutingScope::all_accounts(), + ) +} + +#[tokio::test] +async fn client_catalog_is_account_and_version_scoped_and_keeps_original_objects() { + let store = Arc::new(MemoryAccountStore::default()); + let first = seed_account(&store, "acct_client_a").await; + let second = seed_account(&store, "acct_client_b").await; + let server = MockServer::start().await; + for (account, version, text) in [ + ("acct_client_a", "0.154.0", "a-original"), + ("acct_client_b", "0.154.0", "b-original"), + ("acct_client_a", "0.155.0", "a-new-client"), + ] { + Mock::given(method("GET")).and(path("/codex/models")) + .and(header("authorization", format!("Bearer access-{account}"))) + .and(header("chatgpt-account-id", format!("chatgpt-{account}"))) + .and(query_param("client_version", version)) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"models":[{ + "slug":"gpt-native", "display_name":"Native", "base_instructions":text, + "future_field":{"null":null,"list":[1,true]}, "model_messages":{"new_template":text} + }]}))).expect(1).mount(&server).await; + } + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + for (accounts, version, text) in [ + (vec![second.clone(), first.clone()], "0.154.0", "a-original"), + (vec![second.clone()], "0.154.0", "b-original"), + (vec![first.clone()], "0.155.0", "a-new-client"), + (vec![first], "0.154.0", "a-original"), + ] { + let models = service + .client_model_catalog(&client_scope(&accounts), version) + .await + .expect("catalog"); + let document: serde_json::Value = + serde_json::from_slice(models[0].payload.body()).expect("document"); + assert_eq!( + document, + serde_json::json!({ + "slug":"gpt-native", "display_name":"Native", "base_instructions":text, + "future_field":{"null":null,"list":[1,true]}, "model_messages":{"new_template":text} + }) + ); + } + assert!( + service + .client_model_catalog(&client_scope(&[]), "0.154.0") + .await + .is_err() + ); + server.verify().await; +} + +#[tokio::test] +async fn client_catalog_coalesces_reads_and_invalidates_on_etag_expiry_and_explicit_reset() { + let store = Arc::new(MemoryAccountStore::default()); + let account = seed_account(&store, "acct_client_cache").await; + let scope = client_scope(&[account]); + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with( + ResponseTemplate::new(200) + .set_body_raw(OFFICIAL_FIXTURE, "application/json") + .set_delay(std::time::Duration::from_millis(20)), + ) + .expect(4) + .mount(&server) + .await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + let (first, second) = tokio::join!( + service.client_model_catalog(&scope, "0.154.0"), + service.client_model_catalog(&scope, "0.154.0") + ); + assert_eq!(first.expect("first"), second.expect("second")); + service + .invalidate() + .expect("reset without a routing snapshot"); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("after reset"); + assert!(service.observe_response_etag("catalog-new").expect("ETag")); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("after ETag"); + assert!( + !service + .observe_response_etag("catalog-new") + .expect("same ETag") + ); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("still cached"); + tokio::time::pause(); + tokio::time::advance(std::time::Duration::from_secs(301)).await; + tokio::time::resume(); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("after TTL"); + server.verify().await; +} + +#[tokio::test] +async fn client_catalog_cache_does_not_survive_credential_rotation() { + let store = Arc::new(MemoryAccountStore::default()); + let account = seed_account(&store, "acct_client_rotate").await; + let scope = client_scope(std::slice::from_ref(&account)); + let server = MockServer::start().await; + for (token, text) in [ + ("access-acct_client_rotate", "old"), + ("rotated-access", "new"), + ] { + Mock::given(method("GET")) + .and(path("/codex/models")) + .and(header("authorization", format!("Bearer {token}"))) + .respond_with( + ResponseTemplate::new(200).set_body_json(serde_json::json!({"models":[{ + "slug":"gpt-native", "display_name":"Native", "base_instructions":text + }]})), + ) + .expect(1) + .mount(&server) + .await; + } + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + let first = service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("old"); + store + .repository() + .rotate_refreshed_oauth_secret(&account, secret("rotated-access"), None, None) + .await + .expect("rotate"); + let second = service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("new"); + assert_ne!(first, second); + server.verify().await; +} + +#[tokio::test] +async fn failed_client_catalog_never_falls_back_to_an_out_of_scope_account_or_plan_cache() { + let store = Arc::new(MemoryAccountStore::default()); + let allowed = seed_account(&store, "acct_client_allowed").await; + seed_account(&store, "acct_client_outside").await; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .and(header("authorization", "Bearer access-acct_client_allowed")) + .respond_with(ResponseTemplate::new(503)) + .expect(1) + .mount(&server) + .await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + let scope = client_scope(&[allowed]); + for _ in 0..2 { + assert!( + service + .client_model_catalog(&scope, "0.154.0") + .await + .is_err() + ); + } + assert_eq!(server.received_requests().await.expect("requests").len(), 1); + server.verify().await; +} + +#[tokio::test] +async fn raw_instructions_only_change_advances_catalog_generation() { + let store = Arc::new(MemoryAccountStore::default()); + seed_account(&store, "acct_raw_generation").await; + let server = MockServer::start().await; + Mock::given(method("GET")).and(path("/codex/models")) + .respond_with(SequencedCatalogResponder { + calls: Arc::new(AtomicUsize::new(0)), + bodies: [ + br#"{"models":[{"slug":"gpt-native","display_name":"Native","model_messages":{"future":"old"}}]}"#, + br#"{"models":[{"slug":"gpt-native","display_name":"Native","model_messages":{"future":"new"}}]}"#, + ], + }).expect(2).mount(&server).await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + service.synchronize().await.expect("initial"); + let generation = service.catalog_generation(); + service.refresh_catalogs().await.expect("refresh"); + assert!(service.catalog_generation() > generation); + server.verify().await; +} + +#[tokio::test] +async fn client_catalog_evicts_old_entries_after_32_distinct_versions() { + let store = Arc::new(MemoryAccountStore::default()); + let account = seed_account(&store, "acct_client_bounded").await; + let scope = client_scope(&[account]); + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(200).set_body_raw(OFFICIAL_FIXTURE, "application/json")) + .expect(34) + .mount(&server) + .await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + for patch in 0..33 { + service + .client_model_catalog(&scope, &format!("0.154.{patch}")) + .await + .expect("catalog"); + } + service + .client_model_catalog(&scope, "0.154.32") + .await + .expect("newest is cached"); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("oldest was evicted"); + server.verify().await; +} + +#[tokio::test] +async fn client_catalog_retries_after_short_failure_cache_expires() { + let store = Arc::new(MemoryAccountStore::default()); + let account = seed_account(&store, "acct_client_retry").await; + let scope = client_scope(&[account]); + let server = MockServer::start().await; + let failure = Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(503)) + .expect(1) + .mount_as_scoped(&server) + .await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + assert!( + service + .client_model_catalog(&scope, "0.154.0") + .await + .is_err() + ); + drop(failure); + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(200).set_body_raw(OFFICIAL_FIXTURE, "application/json")) + .expect(1) + .mount(&server) + .await; + assert!( + service + .client_model_catalog(&scope, "0.154.0") + .await + .is_err() + ); + tokio::time::pause(); + tokio::time::advance(std::time::Duration::from_secs(6)).await; + tokio::time::resume(); + service + .client_model_catalog(&scope, "0.154.0") + .await + .expect("recovered after failure TTL"); + server.verify().await; +} + #[test] fn catalog_starts_without_a_snapshot_or_generation() { let store = Arc::new(MemoryAccountStore::default()); @@ -192,6 +479,52 @@ async fn plan_catalog_cache_is_shared_and_manual_refresh_replaces_it() { server.verify().await; } +#[tokio::test] +async fn service_tier_changes_should_replace_cached_metadata_and_advance_generation() { + let store = Arc::new(MemoryAccountStore::default()); + seed_account(&store, "acct_service_tiers").await; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(SequencedCatalogResponder { + calls: Arc::new(AtomicUsize::new(0)), + bodies: [ + br#"{"models":[{"slug":"gpt-5.4","display_name":"GPT-5.4","service_tiers":[{"id":"priority","name":"Fast","description":"Priority processing."}]}]}"#, + br#"{"models":[{"slug":"gpt-5.4","display_name":"GPT-5.4","service_tiers":[]}]}"#, + ], + }) + .expect(2) + .mount(&server) + .await; + let service = service_with_catalog_cache(&store, server.uri(), catalog_cache()); + + let initial = service.synchronize().await.expect("initial catalog"); + let generation = service.catalog_generation().get(); + let cached = service.synchronize().await.expect("cached catalog"); + assert_eq!(cached.models(), initial.models()); + assert_eq!( + cached.models()[0].metadata().service_tiers(), + [ModelServiceTier::new( + "priority", + "Fast", + "Priority processing." + )] + ); + assert_eq!(service.catalog_generation().get(), generation); + + let refreshed = service.refresh_catalogs().await.expect("updated catalog"); + assert!(refreshed.models()[0].metadata().service_tiers().is_empty()); + assert_eq!(service.catalog_generation().get(), generation + 1); + assert_eq!( + service + .cached() + .expect("cache read") + .expect("cached snapshot") + .models(), + refreshed.models() + ); +} + #[tokio::test] async fn plan_catalog_refresh_stops_after_three_failed_accounts() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 117429193..f314b8508 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -33,9 +33,9 @@ use gateway_core::operation::{ }; use gateway_core::policy::ClientApiKeyId; use gateway_core::routing::{ - ClientRoutingScope, ConfigRevision, FrozenAccountScope, ModelCapabilities, ProviderKind, - ProviderModel, PublicModelId, RoutingContext, RuntimeAccount, RuntimeAccountDirectory, - RuntimeSnapshot, UpstreamModelId, + ClientRoutingScope, ConfigRevision, FrozenAccountScope, ModelCapabilities, ModelServiceTier, + ProviderKind, ProviderModel, PublicModelId, RoutingContext, RuntimeAccount, + RuntimeAccountDirectory, RuntimeSnapshot, UpstreamModelId, }; use gateway_core::upstream::UpstreamSendState; use provider_openai::config::DEFAULT_STREAM_MAX_RETRIES; @@ -6307,6 +6307,41 @@ async fn provider_compiles_catalog_presentation_for_codex_models() { assert!(presentation.image_detail_original()); assert!(presentation.verbosity()); assert!(!presentation.hidden()); + assert_eq!( + presentation.service_tiers(), + [ModelServiceTier::new( + "priority", + "Fast", + "Priority processing." + )] + ); + + let scope = FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(BTreeMap::from([( + ProviderAccountId::new("acct_presentation").expect("account"), + RuntimeAccount::new( + ProviderKind::new("openai").expect("provider"), + BTreeSet::new(), + ), + )]))), + ClientRoutingScope::all_accounts(), + ); + let native = provider + .query_client_model_catalog(&scope, "codex", "0.154.0") + .await + .expect("native catalog") + .expect("supported"); + let original: Value = serde_json::from_slice(OFFICIAL_FIXTURE).expect("fixture"); + let document: Value = serde_json::from_slice(native[0].payload.body()).expect("document"); + assert_eq!(document, original["models"][0]); + assert!( + server + .received_requests() + .await + .expect("requests") + .iter() + .any(|request| request.url.query() == Some("client_version=0.154.0")) + ); } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/transport/catalog.rs b/backend/crates/providers/openai/tests/transport/catalog.rs index 395c80afe..e8bcb23dd 100644 --- a/backend/crates/providers/openai/tests/transport/catalog.rs +++ b/backend/crates/providers/openai/tests/transport/catalog.rs @@ -1,15 +1,43 @@ use chrono::{TimeZone, Utc}; +use gateway_core::routing::ModelServiceTier; use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState}; use provider_openai::transport::{ CodexBackendClient, CodexCatalogCapabilityEvidence, CodexCatalogVisibility, CodexClientError, CodexModelCatalogError, CodexRequestContext, MAX_CODEX_MODEL_CATALOG_BYTES, build_reqwest_client, parse_codex_model_catalog, }; +use serde_json::{Value, json}; use wiremock::matchers::{header, method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; const OFFICIAL_FIXTURE: &[u8] = include_bytes!("fixtures/official_models_snapshot.json"); +#[test] +fn native_document_keeps_unknown_nested_fields_nulls_and_model_instructions() { + let original = json!({ + "slug": "gpt-native", "display_name": "Native", + "base_instructions": "original model instructions\nwith line breaks", + "model_messages": {"future_instructions": {"template": "do not replace"}}, + "future_field": [null, {"new": true}], "explicit_null": null, + "service_tiers": [{"id":"priority", "name":"Fast", "description":"1.5x speed", "future":true}] + }); + let body = serde_json::to_vec(&json!({"models":[original]})).expect("body"); + let snapshot = parse_codex_model_catalog(&body, None).expect("catalog"); + let model = &snapshot.models()[0]; + let preserved: Value = serde_json::from_slice(model.document().body()).expect("document"); + assert_eq!(preserved, original); + assert!(!format!("{model:?}").contains("original model instructions")); + assert!(!format!("{model:?}").contains("future_instructions")); + let mut changed = original; + changed["model_messages"]["future_instructions"]["template"] = json!("new content"); + let changed = parse_codex_model_catalog( + &serde_json::to_vec(&json!({"models":[changed]})).expect("body"), + None, + ) + .expect("catalog"); + assert_ne!(snapshot.models(), changed.models()); +} + #[test] fn official_fixture_should_produce_safe_full_snapshot() { let snapshot = parse_codex_model_catalog(OFFICIAL_FIXTURE, Some("W/\"codex-v1\"")) @@ -65,7 +93,7 @@ fn official_fixture_should_produce_safe_full_snapshot() { } #[test] -fn non_whitelisted_wire_fields_should_not_survive_normalization() { +fn raw_instructions_and_unknown_fields_should_not_appear_in_snapshot_debug() { let snapshot = parse_codex_model_catalog(OFFICIAL_FIXTURE, None).expect("official fixture should parse"); let debug = format!("{snapshot:?}"); @@ -77,6 +105,91 @@ fn non_whitelisted_wire_fields_should_not_survive_normalization() { ); } +#[test] +fn service_tiers_should_preserve_declared_ids_names_and_descriptions() { + let body = json!({"models": [{ + "slug": "gpt-future", + "display_name": "Future", + "service_tiers": [ + {"id": "priority", "name": "Fast", "description": "Priority processing."}, + {"id": "deferred", "name": "Economy", "description": ""} + ] + }]}); + let snapshot = parse_codex_model_catalog(&serde_json::to_vec(&body).expect("JSON"), None) + .expect("declared service tiers"); + + assert_eq!( + snapshot.models()[0].metadata().service_tiers(), + [ + ModelServiceTier::new("priority", "Fast", "Priority processing."), + ModelServiceTier::new("deferred", "Economy", ""), + ] + ); +} + +#[test] +fn missing_or_empty_service_tiers_should_not_infer_fast_from_legacy_metadata() { + for service_tiers in [None, Some(json!([]))] { + let mut model = json!({ + "slug": "gpt-5.4", + "display_name": "GPT-5.4", + "additional_speed_tiers": ["fast"] + }); + if let Some(service_tiers) = service_tiers { + model["service_tiers"] = service_tiers; + } + let body = json!({"models": [model]}); + let snapshot = parse_codex_model_catalog(&serde_json::to_vec(&body).expect("JSON"), None) + .expect("optional service tiers"); + + assert!(snapshot.models()[0].metadata().service_tiers().is_empty()); + } +} + +#[test] +fn invalid_service_tier_metadata_should_fail_the_entire_snapshot() { + for (field, value) in [ + ("id", String::new()), + ("id", "p".repeat(65)), + ("name", " ".to_owned()), + ("name", "f".repeat(257)), + ("name", "Fast\u{1b}".to_owned()), + ("description", "d".repeat(4097)), + ("description", "unsafe\u{0}".to_owned()), + ] { + let mut tier = json!({"id": "priority", "name": "Fast", "description": "Priority."}); + tier[field] = Value::String(value); + let body = json!({"models": [{ + "slug": "gpt-5.4", "display_name": "GPT-5.4", "service_tiers": [tier] + }]}); + + assert_eq!( + parse_codex_model_catalog(&serde_json::to_vec(&body).expect("JSON"), None), + Err(CodexModelCatalogError::InvalidMetadata), + "invalid {field}" + ); + } +} + +#[test] +fn malformed_service_tiers_should_fail_the_official_wire_contract() { + for service_tiers in [ + Value::Null, + json!("fast"), + json!([{"id": "priority", "name": "Fast"}]), + json!([{"id": 1, "name": "Fast", "description": "Priority."}]), + ] { + let body = json!({"models": [{ + "slug": "gpt-5.4", "display_name": "GPT-5.4", "service_tiers": service_tiers + }]}); + + assert_eq!( + parse_codex_model_catalog(&serde_json::to_vec(&body).expect("JSON"), None), + Err(CodexModelCatalogError::InvalidWire) + ); + } +} + #[test] fn invalid_etag_should_fail_the_entire_snapshot() { let result = parse_codex_model_catalog(OFFICIAL_FIXTURE, Some("raw-unquoted-etag")); @@ -241,7 +354,7 @@ async fn fetch_should_send_official_catalog_headers_and_query() { profile(), ); let snapshot = client - .fetch_models_with_context(context()) + .fetch_models_with_context(context(), None) .await .expect("fetch strict snapshot"); @@ -285,7 +398,7 @@ async fn fetch_should_reject_streamed_body_over_hard_limit() { profile(), ); - let result = client.fetch_models_with_context(context()).await; + let result = client.fetch_models_with_context(context(), None).await; assert!(matches!( result, diff --git a/backend/crates/providers/openai/tests/transport/fixtures/official_models_snapshot.json b/backend/crates/providers/openai/tests/transport/fixtures/official_models_snapshot.json index cbae59899..cf0fa5651 100644 --- a/backend/crates/providers/openai/tests/transport/fixtures/official_models_snapshot.json +++ b/backend/crates/providers/openai/tests/transport/fixtures/official_models_snapshot.json @@ -13,6 +13,13 @@ "supports_image_detail_original": true, "supports_search_tool": true, "support_verbosity": true, + "service_tiers": [ + { + "id": "priority", + "name": "Fast", + "description": "Priority processing." + } + ], "supported_reasoning_levels": [ { "effort": "low", diff --git a/backend/crates/providers/openai/tests/transport/profile_contract.rs b/backend/crates/providers/openai/tests/transport/profile_contract.rs index 4d5d60084..10e30ab43 100644 --- a/backend/crates/providers/openai/tests/transport/profile_contract.rs +++ b/backend/crates/providers/openai/tests/transport/profile_contract.rs @@ -32,7 +32,7 @@ async fn each_core_endpoint_uses_the_current_bundled_release() { }); let context = request_context("audit", Some("audit-account")); client - .fetch_models_with_context(context) + .fetch_models_with_context(context, None) .await .expect_err("fixture rejection"); client diff --git a/docs/api.md b/docs/api.md index fce1ba5f6..e28688c1b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -183,8 +183,34 @@ message/type/code；没有结构化错误时使用稳定安全文案，不把原 `GET /v1/models` 默认返回 OpenAI 兼容列表 `{"object": "list", "data": [...]}`；请求携带非空 `client_version` query 参数（Codex 客户端）时改为返回 Codex 专用目录合同 `{"models": [...]}`。 +OpenAI Provider 按客户端传入的 `client_version` 请求上游目录，完整保留每个模型 JSON 对象，包括 +`base_instructions`、`model_messages`、`service_tiers`、工具与能力字段，以及未知嵌套字段、显式 `null` +和字段缺失的区别。Core 不解释这些协议字段，API 不再根据通用模型画像重建 OpenAI 目录。 +模型别名仅替换 `slug`，不替换上游展示名、提示词、能力或 `priority`；保持原生模型顺序，新增别名附在后面。 +目录仍按当前路由快照的模型存在性过滤，避免公布已知无法路由的模型；新模型需待后台目录对账后进入列表。 +xAI 没有 Codex 原生目录，继续使用明确的通用画像适配。 + +目录账号只能来自本次 Client Key 冻结的账号范围。OpenAI 在其中按账号 ID 排序，使用首个成功读取的 +合格账号，最多尝试三个账号；同名模型不跨账号或套餐混拼字段。因此单账号、无别名时可保持该账号的 +模型对象一致，多账号/多 Provider 聚合不代表“与某个官方账号的整个目录完全一致”，也不会固定后续推理账号。 +读取失败返回 `503 model_catalog_unavailable`，不以简化模板或空成功响应覆盖客户端缓存。 + +原生目录使用 Provider 内的有界缓存：最多 32 份，成功 TTL 为 5 分钟，失败短缓存为 5 秒，单次上游 +读取超时为 15 秒。缓存按账号、凭据 revision、套餐、上游账号身份、客户端版本和请求画像隔离；并发 +读取合并，新的目录 ETag、后台目录内容变化或显式失效会清理缓存。每次读取仍重新检查账号资格和 Key 范围， +不会因缓存命中跨越权限。完整对象不写 PostgreSQL 或 Redis；现有套餐 Redis cache 仍只保存模型 ID。 +成功目录响应带 `Cache-Control: private, no-store`，不借用上游 ETag 标识经过选择/别名/聚合后的正文。 +Codex 自己仍会写 `models_cache.json`；Responses 的 `x-models-etag` 保持原协议，这可能让客户端再次 +读取目录，但通常命中上述服务端缓存，不表示每轮都重新请求上游。 + +`service_tiers` 的 `name` 用于生成 `/fast` 等命令，`id` 是请求使用的 `service_tier` 值，二者不能互换。 +上游未声明或明确返回空数组时不补档位，也不根据模型名或旧 `additional_speed_tiers` 字段推断 Fast。 +该合同对齐 +[官方 Codex 0.154.0 的模型元数据](https://github.com/openai/codex/blob/6b9826e3aa83b1a5947db50f4332cb9c65f1b340/codex-rs/protocol/src/openai_models.rs) +与其动态服务档位命令；升级网关后，已有客户端缓存需要在重新加载模型目录后才能反映修复。 + Codex 专用目录中的 `context_window` 与 `max_context_window` 分别表示默认上下文窗口和客户端本地 -覆盖的上限。OpenAI Provider 分别传递上游目录中的对应字段，缺失时保留 `null`；网关不通过部署配置 +覆盖的上限。OpenAI Provider 原样保留上游对应字段，缺失与 `null` 不互相转换；网关不通过部署配置 覆盖这些值。Codex 客户端配置 `model_context_window` 后，按该值与非空 `max_context_window` 的较小值 使用窗口；上限为空时保留客户端本地值。xAI 目录只声明一个窗口，其 Provider 继续以该值作为客户端覆盖上限。 diff --git a/docs/architecture.md b/docs/architecture.md index 3ab39f71b..e2d223336 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -121,6 +121,13 @@ Provider 模型能力、目录代次与 `ProviderCatalogPort` 由 `routing::cata 该只读合同，不反向依赖执行注册表；现有 `ProviderRegistry` 直接实现目录端口，仍只维护一份 Provider 注册集合。已知空目录与查询失败的未知目录保持不同语义，目录替身无需实现请求执行。 +客户端原生模型目录同样由 Provider 拥有，通过 `ExecutionService` 和现有 Registry 的只读调用传递。 +`routing::catalog` 的原生条目只包含模型 ID 与不透明协议正文，Core 负责冻结账号范围和整对象模型映射， +不解析上游字段；API 负责协议输出，不直接依赖具体 Provider。原生目录失败不会退回通用画像，只有不提供 +该协议原生目录的 Provider（如 xAI 的 Codex 适配）才走画像转换。目录读取不创建计量请求或推理 attempt。 +OpenAI 的账号/凭据 revision/客户端版本隔离、并发合并、TTL、容量与失效均归属已有 credential catalog +service；客户端原生对象只进有界进程缓存，不扩展套餐 Redis ID cache 或持久化模型字段副本。 + `engine::observation` 统一维护单次响应的用量、费用、时间和响应 ID，并负责重试前清理；协调器继续 独占发送、提交、重试和终结顺序。Provider 上报费用优先于本地估算，丢弃的 attempt 不得污染最终计量。 Client Key 费用账本独立累计各次 attempt 的实际费用，不能因请求重试而清空已产生的费用或未知计费状态。