From 39aa20a13aae2e7fcc346af22c7032350f4ee86b Mon Sep 17 00:00:00 2001 From: Ke Wang Date: Wed, 4 Nov 2026 13:18:39 +0800 Subject: [PATCH] fix(plugins): prefer account-scoped profiles for adapted models (#588) * fix(plugins): prefer account-scoped model profiles in catalogs Co-authored-by: Codex * fix(models): honor scoped plugin profiles in fallback catalogs Co-authored-by: Codex --------- Co-authored-by: Codex Co-authored-by: Zyy --- .../src/engine/execution/service.rs | 45 +++- .../gateway-core/src/routing/catalog.rs | 13 +- .../gateway-core/src/routing/snapshot.rs | 46 ++-- .../tests/engine/execution/identity.rs | 14 +- .../tests/engine/execution/mod.rs | 228 +++++++++++++----- .../gateway-plugin/sdk/docs/capabilities.md | 3 +- docs/api.md | 3 +- docs/architecture.md | 2 +- 8 files changed, 250 insertions(+), 104 deletions(-) diff --git a/backend/crates/gateway-core/src/engine/execution/service.rs b/backend/crates/gateway-core/src/engine/execution/service.rs index 8123a01a3..31d95d78d 100644 --- a/backend/crates/gateway-core/src/engine/execution/service.rs +++ b/backend/crates/gateway-core/src/engine/execution/service.rs @@ -62,15 +62,16 @@ use crate::{ policy::{ClientApiKeyId, ClientPolicy}, provider_ports::{ProviderSessionAffinityPort, ProviderStoreErrorKind}, routing::{ - FrozenAccountScope, ProviderCatalogUnavailable, PublicModelDescriptor, PublicModelId, - RoutingContext, RuntimeSnapshot, UpstreamModelId, request_settings::RequestSettings, + FrozenAccountScope, ProviderCatalogUnavailable, ProviderModelContent, + PublicModelDescriptor, PublicModelId, RoutingContext, RuntimeSnapshot, UpstreamModelId, + request_settings::RequestSettings, }, runtime::{RuntimeSnapshotHandle, RuntimeSnapshotPublisher}, }; use futures::{FutureExt as _, future::BoxFuture, pin_mut, select_biased}; use futures_timer::Delay; use std::{ - collections::BTreeSet, + collections::{BTreeMap, BTreeSet}, sync::{Arc, Weak}, time::{Instant, SystemTime}, }; @@ -1575,11 +1576,27 @@ impl ExecutionService for DefaultExecutionService { } continue; }; + let contributions: Vec<_> = client + .snapshot + .contributed_models_for_scope(kind, scope) + .collect(); + let mut contributed_by_model = BTreeMap::new(); + if !contributions.is_empty() + && models + .iter() + .any(|entry| matches!(entry.content, ProviderModelContent::Adapted(_))) + { + contributed_by_model.extend( + contributions + .iter() + .map(|model| (&model.id, &model.presentation)), + ); + } // 保留上游顺序；映射只选择完整的目标对象，不能跨账号/模型混拼字段 let by_id = models .iter() .map(|entry| (entry.model.as_str(), entry)) - .collect::>(); + .collect::>(); for entry in &models { let target = client.snapshot.mapped_model(entry.model.as_str()); if !scope.allows_provider_model(kind, &target) { @@ -1598,7 +1615,11 @@ impl ExecutionService for DefaultExecutionService { continue; } if seen.insert(model.clone()) { - result.push(source.content.for_public_model(model)); + result.push( + source + .content + .for_public_model(model, &contributed_by_model), + ); } } for model in client.snapshot.public_models_for_provider(kind) { @@ -1617,13 +1638,17 @@ impl ExecutionService for DefaultExecutionService { } if let Some(source) = by_id.get(target.as_str()) { seen.insert(model.clone()); - result.push(source.content.for_public_model(model)); + result.push( + source + .content + .for_public_model(model, &contributed_by_model), + ); } } - // 原生完整对象优先；只补入当前授权账号实际提供的独立适配画像 - for profile in client.snapshot.contributed_profiles_for_scope(kind, scope) { - if seen.insert(profile.model().clone()) { - result.push(PublicModelDescriptor::Adapted(profile)); + // 没有 Provider 目录条目的适配模型仍由当前授权账号贡献 + for model in contributions { + if seen.insert(model.id.clone()) { + result.push(PublicModelDescriptor::Adapted(model.public_profile())); } } } diff --git a/backend/crates/gateway-core/src/routing/catalog.rs b/backend/crates/gateway-core/src/routing/catalog.rs index e41ed43d2..4f923bad2 100644 --- a/backend/crates/gateway-core/src/routing/catalog.rs +++ b/backend/crates/gateway-core/src/routing/catalog.rs @@ -32,6 +32,12 @@ pub struct ContributedModel { pub presentation: ModelPresentation, } +impl ContributedModel { + pub(crate) fn public_profile(&self) -> PublicModelProfile { + PublicModelProfile::new(self.id.clone(), self.presentation.clone()) + } +} + /// Provider 客户端目录条目；缺少原生协议正文时使用已编译的通用画像 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderModelDescriptor { @@ -56,13 +62,18 @@ pub enum PublicModelDescriptor { } impl ProviderModelContent { - pub(crate) fn for_public_model(&self, model: PublicModelId) -> PublicModelDescriptor { + pub(crate) fn for_public_model( + &self, + model: PublicModelId, + contributed: &BTreeMap<&PublicModelId, &ModelPresentation>, + ) -> PublicModelDescriptor { match self { Self::Native(payload) => PublicModelDescriptor::Native { model, payload: payload.clone(), }, Self::Adapted(presentation) => { + let presentation = contributed.get(&model).copied().unwrap_or(presentation); PublicModelDescriptor::Adapted(PublicModelProfile::new(model, presentation.clone())) } } diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 0703ee48f..754020006 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -701,24 +701,18 @@ impl RuntimeSnapshot { self.extensions.as_ref().map_or(&[], |set| set.models()) } - pub(crate) fn contributed_profiles_for_scope( + pub(crate) fn contributed_models_for_scope( &self, provider: &ProviderKind, scope: &FrozenAccountScope, - ) -> Vec { - self.contributed_models() - .iter() - .filter(|model| { - &model.provider == provider - && model.accounts.iter().any(|account| { - scope.account_provider(account) == Some(provider) - && scope.allows_model(account, &self.mapped_model(model.id.as_str())) - }) - }) - .map(|model| { - super::PublicModelProfile::new(model.id.clone(), model.presentation.clone()) - }) - .collect() + ) -> impl Iterator { + self.contributed_models().iter().filter(move |model| { + &model.provider == provider + && model.accounts.iter().any(|account| { + scope.account_provider(account) == Some(provider) + && scope.allows_model(account, &self.mapped_model(model.id.as_str())) + }) + }) } fn model_alias(&self, model: &str) -> Option<&super::ContributedModelAlias> { @@ -1014,13 +1008,21 @@ impl RuntimeSnapshot { provider: &ProviderKind, scope: &FrozenAccountScope, ) -> Vec { - self.public_model_profiles_for_provider(provider) - .into_iter() - .filter(|profile| { - self.native_catalog_model_allowed_for_scope(provider, profile.model(), scope) - }) - .chain(self.contributed_profiles_for_scope(provider, scope)) - .collect() + let mut profiles = self.public_model_profiles_for_provider(provider); + profiles.retain(|profile| { + self.native_catalog_model_allowed_for_scope(provider, profile.model(), scope) + }); + // Provider 画像按公开 ID 排序，追加的插件画像保留注册顺序，不参与二分查找 + let provider_count = profiles.len(); + for model in self.contributed_models_for_scope(provider, scope) { + match profiles[..provider_count] + .binary_search_by(|profile| profile.model().cmp(&model.id)) + { + Ok(index) => profiles[index].presentation.clone_from(&model.presentation), + Err(_) => profiles.push(model.public_profile()), + } + } + profiles } /// 已知模型只属于明确声明它的 Provider；未知模型仍可由非完整目录透传 diff --git a/backend/crates/gateway-core/tests/engine/execution/identity.rs b/backend/crates/gateway-core/tests/engine/execution/identity.rs index 8f208d8de..16bb88cc0 100644 --- a/backend/crates/gateway-core/tests/engine/execution/identity.rs +++ b/backend/crates/gateway-core/tests/engine/execution/identity.rs @@ -16,11 +16,8 @@ fn identity_queries_keep_frozen_permissions_without_using_execution_resources() let service = DefaultExecutionService::new( snapshots.clone(), store.clone(), - ProviderRegistry::new([Arc::new(NativeCatalogProvider { - fail: false, - empty: false, - }) as Arc]) - .unwrap(), + ProviderRegistry::new([Arc::new(CatalogProvider::Native) as Arc]) + .unwrap(), Arc::new(UnusedAdmissions), Arc::new(UnusedContinuation), usage.clone(), @@ -91,11 +88,8 @@ fn execution_scope_limits_apply_only_when_starting_a_model() { let service = DefaultExecutionService::new( RuntimeSnapshotHandle::new(start_snapshot()), store.clone(), - ProviderRegistry::new([Arc::new(NativeCatalogProvider { - fail: false, - empty: false, - }) as Arc]) - .unwrap(), + ProviderRegistry::new([Arc::new(CatalogProvider::Native) as Arc]) + .unwrap(), Arc::new(UnusedAdmissions), Arc::new(UnusedContinuation), usage.clone(), diff --git a/backend/crates/gateway-core/tests/engine/execution/mod.rs b/backend/crates/gateway-core/tests/engine/execution/mod.rs index 6d2b96eff..9b0c594fc 100644 --- a/backend/crates/gateway-core/tests/engine/execution/mod.rs +++ b/backend/crates/gateway-core/tests/engine/execution/mod.rs @@ -2,7 +2,10 @@ mod identity; mod outbound; -use gateway_core::routing::extensions::{ExtensionSetId, ExtensionSetLease, ExtensionSetReference}; +use gateway_core::routing::{ + ContributedModel, ContributedModelAlias, + extensions::{ExtensionSetId, ExtensionSetLease, ExtensionSetReference}, +}; use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, @@ -3300,15 +3303,22 @@ fn probe_observation_store_failure_preserves_the_provider_error() { struct FailingProvider; -struct NativeCatalogProvider { - fail: bool, - empty: bool, +#[derive(Clone, Copy)] +enum CatalogProvider { + Native, + Empty, + Unavailable, + Generic, + Xai, } #[async_trait] -impl Provider for NativeCatalogProvider { +impl Provider for CatalogProvider { fn name(&self) -> &'static str { - "openai" + match self { + Self::Xai => "xai", + _ => "openai", + } } fn catalog_generation(&self) -> ProviderCatalogGeneration { ProviderCatalogGeneration::default() @@ -3330,29 +3340,41 @@ impl Provider for NativeCatalogProvider { 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); + match self { + Self::Unavailable => return Err(gateway_core::routing::ProviderCatalogUnavailable), + Self::Empty => return Ok(Some(Vec::new())), + Self::Xai => return Ok(None), + Self::Native | Self::Generic => {} } - if self.empty { - return Ok(Some(Vec::new())); - } - Ok(Some( - ["gpt-new", "gpt-start"] - .into_iter() - .map(|id| gateway_core::routing::ProviderModelDescriptor { - model: UpstreamModelId::new(id).expect("id"), - content: gateway_core::routing::ProviderModelContent::Native( - RawJsonPayload::new( - "codex", - serde_json::to_vec(&json!({"slug":id,"future":null})) - .expect("JSON") - .into(), - ) - .expect("payload"), + let mut models = ["gpt-new", "gpt-start"] + .into_iter() + .map(|id| gateway_core::routing::ProviderModelDescriptor { + model: UpstreamModelId::new(id).expect("id"), + content: gateway_core::routing::ProviderModelContent::Native( + RawJsonPayload::new( + "codex", + serde_json::to_vec(&json!({"slug":id,"future":null})) + .expect("JSON") + .into(), + ) + .expect("payload"), + ), + }) + .collect::>(); + if matches!(self, Self::Generic) { + models.extend(["shared-model", "z-last"].map(|id| { + gateway_core::routing::ProviderModelDescriptor { + model: UpstreamModelId::new(id).unwrap(), + content: gateway_core::routing::ProviderModelContent::Adapted( + gateway_core::routing::ModelPresentation::new( + Some(format!("Provider {id}")), + None, + ), ), - }) - .collect(), - )) + } + })); + } + Ok(Some(models)) } async fn execute( self: Arc, @@ -3363,32 +3385,37 @@ impl Provider for NativeCatalogProvider { } } +struct CatalogContributions { + aliases: Vec, + models: Vec, +} + +impl ExtensionSetLease for CatalogContributions { + fn is_ready(&self) -> bool { + true + } + fn model_aliases(&self) -> &[ContributedModelAlias] { + &self.aliases + } + fn models(&self) -> &[ContributedModel] { + &self.models + } +} + #[test] fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models() { - #[derive(Debug)] - struct Aliases( - Vec, - Vec, - ); - impl gateway_core::routing::extensions::ExtensionSetLease for Aliases { - fn is_ready(&self) -> bool { - true - } - fn model_aliases(&self) -> &[gateway_core::routing::ContributedModelAlias] { - &self.0 - } - fn models(&self) -> &[gateway_core::routing::ContributedModel] { - &self.1 - } - } - for (fail, empty) in [(false, false), (false, true), (true, false)] { + for provider in [ + CatalogProvider::Native, + CatalogProvider::Empty, + CatalogProvider::Unavailable, + ] { let contributed = [ ("gpt-start", "acct_start"), ("adapted", "acct_start"), ("private-model", "acct_other"), ] .into_iter() - .map(|(id, account)| gateway_core::routing::ContributedModel { + .map(|(id, account)| ContributedModel { owner: "catalog-plugin".into(), id: PublicModelId::new(id).unwrap(), provider: ProviderKind::new("openai").unwrap(), @@ -3412,25 +3439,21 @@ fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models( .with_settings(&settings) .unwrap() .with_extensions(Some(ExtensionSetReference::new( - gateway_core::routing::extensions::ExtensionSetId::new("catalog-aliases".into()) - .unwrap(), - Arc::new(Aliases( - vec![gateway_core::routing::ContributedModelAlias { + ExtensionSetId::new("catalog-aliases".into()).unwrap(), + Arc::new(CatalogContributions { + aliases: vec![ContributedModelAlias { owner: "catalog-plugin".into(), id: PublicModelId::new("plugin-alias").unwrap(), provider: ProviderKind::new("openai").unwrap(), target: UpstreamModelId::new("gpt-start").unwrap(), }], - contributed, - )), + models: contributed, + }), ))); let service = DefaultExecutionService::new( RuntimeSnapshotHandle::new(snapshot), Arc::new(TrackingExecutionStore::default()), - ProviderRegistry::new([ - Arc::new(NativeCatalogProvider { fail, empty }) as Arc - ]) - .expect("registry"), + ProviderRegistry::new([Arc::new(provider) as Arc]).expect("registry"), Arc::new(UnusedAdmissions), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), @@ -3438,7 +3461,7 @@ fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models( ); let client = service.authenticate("sk_start_test").expect("authenticate"); let result = block_on(service.client_model_catalog(&client, "codex", "0.154.0")); - if fail { + if matches!(provider, CatalogProvider::Unavailable) { assert!( result.is_err(), "native failure cannot use the global synthesized catalog" @@ -3457,13 +3480,13 @@ fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models( .collect::>(); assert_eq!( adapted, - if empty { + if matches!(provider, CatalogProvider::Empty) { vec!["gpt-start", "adapted"] } else { vec!["adapted"] } ); - if empty { + if matches!(provider, CatalogProvider::Empty) { assert_eq!(models.len(), 2); continue; } @@ -3495,6 +3518,95 @@ fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models( } } +#[test] +fn client_catalog_prefers_scoped_profiles_in_provider_order() { + use gateway_core::routing::{ModelPresentation, PublicModelDescriptor}; + + for provider in [CatalogProvider::Generic, CatalogProvider::Xai] { + let kind = ProviderKind::new(provider.name()).unwrap(); + let capabilities = ModelCapabilities::new([OperationKind::Generate].into(), None); + let presentation = + |source: &str, id: &str| ModelPresentation::new(Some(format!("{source} {id}")), None); + let provider_models = ["shared-model", "z-last"].map(|id| { + ProviderModel::new( + kind.clone(), + UpstreamModelId::new(id).unwrap(), + capabilities.clone(), + ) + .with_presentation(presentation("Provider", id)) + }); + let contributions = [ + ("extra", "acct_start"), + ("z-last", "acct_other"), + ("shared-model", "acct_start"), + ] + .map(|(id, account)| ContributedModel { + owner: "catalog-plugin".into(), + id: PublicModelId::new(id).unwrap(), + provider: kind.clone(), + accounts: [ProviderAccountId::new(account).unwrap()].into(), + capabilities: capabilities.clone(), + presentation: presentation("Plugin", id).with_agent_tools(true, true), + }); + let policy = ClientPolicy::new( + ClientApiKeyId::new("key_start_test").unwrap(), + PlaintextClientApiKey::new("sk_start_test").unwrap(), + account_scope(&kind, "acct_start"), + true, + RateLimits::unlimited(), + ); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).unwrap(), + gateway_core::settings::SettingsValues::new(1, "smart", Default::default(), None, None), + vec![kind], + provider_models.into(), + vec![policy], + ) + .unwrap() + .with_extensions(Some(ExtensionSetReference::new( + ExtensionSetId::new("catalog-precedence".into()).unwrap(), + Arc::new(CatalogContributions { + aliases: vec![], + models: contributions.into(), + }), + ))); + let service = DefaultExecutionService::new( + RuntimeSnapshotHandle::new(snapshot), + Arc::new(TrackingExecutionStore::default()), + ProviderRegistry::new([Arc::new(provider) as Arc]).unwrap(), + Arc::new(UnusedAdmissions), + Arc::new(UnusedContinuation), + Arc::new(RecordingClientApiKeyUsage::default()), + Arc::new(crate::RecordingDiagnostics::default()), + ); + let client = service.authenticate("sk_start_test").unwrap(); + let models = block_on(service.client_model_catalog(&client, "codex", "0.154.0")).unwrap(); + let profiles = models + .iter() + .map(|entry| { + let PublicModelDescriptor::Adapted(profile) = entry else { + panic!("expected adapted catalog") + }; + ( + profile.model().as_str(), + profile.presentation().display_name().unwrap(), + profile.presentation().parallel_tool_calls(), + ) + }) + .collect::>(); + assert_eq!( + profiles, + [ + ("shared-model", "Plugin shared-model", true), + ("z-last", "Provider z-last", false), + ("extra", "Plugin extra", true), + ], + "{}", + provider.name() + ); + } +} + #[async_trait] impl Provider for FailingProvider { fn name(&self) -> &'static str { diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index 3fb777a40..e43f7d719 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -578,7 +578,8 @@ let plugin = PluginBuilder::from_json(include_bytes!("../plugin.json"))? 清单须显式声明 `model_catalog.version = 2`，可使用上述 `.model_catalog(...)` 固定目录， 或通过 `PluginBuilder::on(methods::MODEL_CATALOG_REGISTER_V2, handler)` 返回 `TypedReply`。 -原生 Codex 目录成功返回后，宿主保留其中的完整对象，只追加可见且尚未出现的适配模型画像；原生读取错误仍向调用方返回错误。 +Codex 客户端目录保留 Provider 的完整原生对象。对于通用画像，当前账号范围内的同名适配模型使用插件声明的 `presentation`， +包括 xAI 这类由宿主转换的目录。仅由插件提供的可见模型按声明顺序追加到所属 Provider 的条目之后。原生目录读取失败时返回错误。 适配模型的 wire 模型名、协议编解码及摘要由插件负责，宿主只使用公开 ID、能力和账号范围参与现有选号，不生成请求正文或转换历史 ### 重试决策 diff --git a/docs/api.md b/docs/api.md index b31e29882..5df1f6aee 100644 --- a/docs/api.md +++ b/docs/api.md @@ -576,7 +576,8 @@ xAI 没有 Codex 原生目录，使用官方 CLI proxy 当前目录生成通用 “与某个官方账号的整个目录完全一致”，也不会固定后续推理账号。 读取失败返回 `503 model_catalog_unavailable`，不以简化模板或空成功响应覆盖客户端缓存。 成功获取的 API Key 目录被账号模型政策全部过滤时返回空列表，区别于读取失败。 -插件声明的适配模型按受管适配器账号范围与当前 Key 权限交集追加；同名原生完整对象优先保留，不与插件画像拼接字段 +目录保留完整的原生 Codex 模型对象。对于 OpenAI API Key 普通模型列表和 xAI 目录生成的通用画像， +当前 Key 可访问的适配器若声明了同名模型，其画像在原位置替换通用画像。仅由插件提供的可见模型按声明顺序追加到所属 Provider 的条目之后 原生目录可能返回缓存结果，成功缓存有效期为 5 分钟。每次查询都检查账号资格和 Client Key 范围。 成功目录响应带 `Cache-Control: private, no-store`，不借用上游 ETag 标识经过选择/别名/聚合后的正文 diff --git a/docs/architecture.md b/docs/architecture.md index 740e94c8e..ca6a7182a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -134,7 +134,7 @@ flowchart LR - **模型目录与路由共用事实**：`model_catalog` 的目录贡献保存在发布代次引用中，直接别名由插件负责条目正确性，条目冲突或目标不存在不阻止宿主快照发布。 请求按顺序取首个匹配别名并直接使用其目标，不递归解析插件别名；Provider、账号权限与目标能力沿原选路流程处理。 适配模型从同实例受管适配器继承静态账号范围，Core 按冻结权限与请求能力收窄候选账号，不改变调度评分、亲和或租约。 - 公开列表、详情、原生目录和受管模型查询复用相同目录贡献与账号访问范围，不另维护插件专用目录；合同见 SDK [模型目录](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#模型目录) + 公开列表、详情、客户端目录和受管模型查询复用同一组目录贡献与账号访问范围。Core 保留 Provider 的完整原生对象，在投影通用画像时应用当前账号范围的插件声明。合并顺序见 SDK [模型目录](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#模型目录) - **重试由 Core 裁决**：`retry_policy` 只能停止或继续 Core 已允许的恢复路径；绑定按确定顺序委托，故障回退宿主。 发送、交付、续接、外部副作用、预算和取消限制不进入插件控制面，策略返回后继续复核 - **转换声明不替代正文事实**：中间件在 request 阶段声明具体承担的功能和额外上游需求，逐层继承尚未承担的需求，实际正文与有效需求共同参与选路。