diff --git a/backend/apps/gateway/src/bootstrap.rs b/backend/apps/gateway/src/bootstrap.rs index 3d74ef536..585955725 100644 --- a/backend/apps/gateway/src/bootstrap.rs +++ b/backend/apps/gateway/src/bootstrap.rs @@ -187,6 +187,14 @@ async fn launch( core.snapshot_control(), ); or_shutdown!(plugin_runtime, plugin_runtime.bind_client_key_ports(&keys)); + let plugin_resources = gateway_admin::initialize_plugin_resources( + store.admin_ports().plugin_resources(), + core.snapshot_control(), + ); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_resource_ports(&plugin_resources) + ); let core = or_shutdown!(plugin_runtime, core.activate().await); let nested_models = core.nested_model_execution_port(); let affinity_lookup = core.affinity_lookup_port(); @@ -231,6 +239,14 @@ async fn launch( plugin_runtime, plugin_runtime.bind_client_key_ports(&plugin_keys) ); + let plugin_resources = gateway_admin::initialize_plugin_resources( + store.admin_ports().plugin_resources(), + core.snapshot_control(), + ); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_resource_ports(&plugin_resources) + ); let mut core = or_shutdown!(plugin_runtime, core.activate().await); let nested_models = core.nested_model_execution_port(); let affinity_lookup = core.affinity_lookup_port(); @@ -322,6 +338,7 @@ async fn launch( plan.extend(openai.take_worker_contributions()); plan.extend(xai.take_worker_contributions()); plan.extend(admin.take_worker_contributions()); + plan.push(plugin_runtime.maintenance_worker(core.snapshots())?); or_shutdown!( plugin_runtime, host.start_workers(plan, store.worker_leader_lease()) diff --git a/backend/crates/gateway-admin/src/lib.rs b/backend/crates/gateway-admin/src/lib.rs index ac6ed3830..2979f9756 100644 --- a/backend/crates/gateway-admin/src/lib.rs +++ b/backend/crates/gateway-admin/src/lib.rs @@ -539,6 +539,15 @@ pub fn initialize_plugin_client_keys( Arc::new(use_case::plugin_client_keys::DefaultPluginClientKeyAccess::new(service)) } +/// 为 Runtime 组合实例自有资源写入；权限和归属在同一存储事务复核。 +#[must_use] +pub fn initialize_plugin_resources( + store: Arc, + snapshot: Arc, +) -> Arc { + Arc::new(use_case::plugin_resources::DefaultPluginResourceAccess { store, snapshot }) +} + /// Backup Worker 注册：单个可取消 Daemon，owner 固定为 `backup`。 fn backup_worker_contribution( task: backup::task::BackupTask, diff --git a/backend/crates/gateway-admin/src/model/mod.rs b/backend/crates/gateway-admin/src/model/mod.rs index 101460417..bbefdba2c 100644 --- a/backend/crates/gateway-admin/src/model/mod.rs +++ b/backend/crates/gateway-admin/src/model/mod.rs @@ -12,6 +12,7 @@ pub mod import_tasks; pub mod key_usage; pub mod observability; pub mod plugin_client_keys; +pub mod plugin_resources; pub mod plugins; pub mod pricing; pub mod provider_credentials; diff --git a/backend/crates/gateway-admin/src/model/plugin_resources.rs b/backend/crates/gateway-admin/src/model/plugin_resources.rs new file mode 100644 index 000000000..717bb3640 --- /dev/null +++ b/backend/crates/gateway-admin/src/model/plugin_resources.rs @@ -0,0 +1,42 @@ +//! 插件资源归属由宿主调用上下文给出，不能由插件参数选择实例或代次。 + +use super::Revision; + +#[derive(Debug, Clone)] +pub struct PluginResourceOwner { + pub instance_id: String, + pub artifact_sha256: String, + pub revision: Revision, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedResource { + pub id: String, + pub name: String, + pub enabled: bool, +} + +pub struct ResourceMutation { + /// 没有事实变化时不递增配置 revision，也不触发发布。 + pub revision: Option, + pub value: T, +} + +pub struct GroupMembersChange { + pub resource_key: String, + pub add: Vec, + pub remove: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GroupMembersChanged { + pub added: u64, + pub removed: u64, +} + +/// 插件 Key 的首次创建设置；分组始终从本实例资源键解析。 +pub struct ManagedKeyConfig { + pub name: String, + pub limits: gateway_core::policy::RateLimits, + pub budget: gateway_core::engine::budget::ClientBudgetLimits, +} diff --git a/backend/crates/gateway-admin/src/ports/mod.rs b/backend/crates/gateway-admin/src/ports/mod.rs index e261fd4c1..f085998c0 100644 --- a/backend/crates/gateway-admin/src/ports/mod.rs +++ b/backend/crates/gateway-admin/src/ports/mod.rs @@ -6,6 +6,7 @@ pub mod plugin_accounts; pub mod plugin_client_keys; pub mod plugin_management; pub mod plugin_release; +pub mod plugin_resources; pub mod plugins; pub mod pricing; pub mod provider; diff --git a/backend/crates/gateway-admin/src/ports/plugin_resources.rs b/backend/crates/gateway-admin/src/ports/plugin_resources.rs new file mode 100644 index 000000000..f39583eff --- /dev/null +++ b/backend/crates/gateway-admin/src/ports/plugin_resources.rs @@ -0,0 +1,63 @@ +//! 插件自有资源的管理入口与原子持久化端口。 + +use super::store::AdminStoreResult; +use crate::model::{ + AdminError, MutationContext, + account_groups::{CreateAccountGroup, NewAccountGroup}, + client_keys::NewClientKey, + plugin_resources::{ + GroupMembersChange, GroupMembersChanged, ManagedKeyConfig, ManagedResource, + PluginResourceOwner, ResourceMutation, + }, +}; +use async_trait::async_trait; + +#[async_trait] +pub trait PluginResourceAccess: Send + Sync { + async fn ensure_group( + &self, + owner: &PluginResourceOwner, + resource_key: String, + command: CreateAccountGroup, + context: &MutationContext, + ) -> Result; + async fn ensure_key( + &self, + owner: &PluginResourceOwner, + resource_key: String, + groups: Vec, + command: ManagedKeyConfig, + context: &MutationContext, + ) -> Result; + async fn change_members( + &self, + owner: &PluginResourceOwner, + command: GroupMembersChange, + context: &MutationContext, + ) -> Result; +} + +#[async_trait] +pub trait PluginResourceStore: Send + Sync { + async fn ensure_group( + &self, + owner: &PluginResourceOwner, + resource_key: String, + command: NewAccountGroup, + context: &MutationContext, + ) -> AdminStoreResult>; + async fn ensure_key( + &self, + owner: &PluginResourceOwner, + resource_key: String, + groups: Vec, + command: NewClientKey, + context: &MutationContext, + ) -> AdminStoreResult>; + async fn change_members( + &self, + owner: &PluginResourceOwner, + command: GroupMembersChange, + context: &MutationContext, + ) -> AdminStoreResult>; +} diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index be9cbf5bf..c316023fb 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -515,9 +515,15 @@ pub struct AdminStorePorts { backup: BackupStorePorts, plugins: Arc, plugin_state: Arc, + plugin_resources: Arc, } impl AdminStorePorts { + #[must_use] + pub fn plugin_resources(&self) -> Arc { + self.plugin_resources.clone() + } + #[must_use] pub fn plugins(&self) -> Arc { Arc::clone(&self.plugins) @@ -542,6 +548,7 @@ impl AdminStorePorts { backup: BackupStorePorts, plugins: Arc, plugin_state: Arc, + plugin_resources: Arc, ) -> Self { Self { accounts, @@ -552,6 +559,7 @@ impl AdminStorePorts { backup, plugins, plugin_state, + plugin_resources, } } diff --git a/backend/crates/gateway-admin/src/use_case/client_keys.rs b/backend/crates/gateway-admin/src/use_case/client_keys.rs index 342838ffb..cffb37b1e 100644 --- a/backend/crates/gateway-admin/src/use_case/client_keys.rs +++ b/backend/crates/gateway-admin/src/use_case/client_keys.rs @@ -124,9 +124,7 @@ impl ClientKeyService for DefaultClientKeyService { let plaintext = if let Some(key) = command.custom_key { key.expose_for_auth().to_owned() } else { - let mut bytes = [0_u8; 32]; - OsRng.fill_bytes(&mut bytes); - format!("sk_{}", URL_SAFE_NO_PAD.encode(bytes)) + generate_key() }; let (config_revision, record) = self .store @@ -257,3 +255,10 @@ fn validate_cursor(query: &ClientKeyListQuery) -> Result<(), AdminError> { Err(AdminError::invalid("Client API Key 游标不合法")) } } + +// 原生与插件创建共用相同的密钥生成规则。 +pub(super) fn generate_key() -> String { + let mut bytes = [0_u8; 32]; + OsRng.fill_bytes(&mut bytes); + format!("sk_{}", URL_SAFE_NO_PAD.encode(bytes)) +} diff --git a/backend/crates/gateway-admin/src/use_case/mod.rs b/backend/crates/gateway-admin/src/use_case/mod.rs index 0b8cae96f..ebca86abc 100644 --- a/backend/crates/gateway-admin/src/use_case/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/mod.rs @@ -12,6 +12,7 @@ pub mod key_usage; pub mod observability; pub mod plugin_accounts; pub mod plugin_client_keys; +pub mod plugin_resources; pub(crate) mod plugin_update; pub mod plugins; pub mod proxies; diff --git a/backend/crates/gateway-admin/src/use_case/plugin_resources.rs b/backend/crates/gateway-admin/src/use_case/plugin_resources.rs new file mode 100644 index 000000000..c680295a4 --- /dev/null +++ b/backend/crates/gateway-admin/src/use_case/plugin_resources.rs @@ -0,0 +1,108 @@ +use crate::{ + model::{ + AdminError, MutationContext, + account_groups::{CreateAccountGroup, NewAccountGroup}, + client_keys::NewClientKey, + plugin_resources::{ + GroupMembersChange, GroupMembersChanged, ManagedKeyConfig, ManagedResource, + PluginResourceOwner, ResourceMutation, + }, + }, + ports::plugin_resources::{PluginResourceAccess, PluginResourceStore}, +}; +use async_trait::async_trait; +use gateway_core::{policy::ClientApiKeyId, routing::AccountGroupId, runtime::SnapshotControl}; +use std::sync::Arc; +use uuid::Uuid; + +pub(crate) struct DefaultPluginResourceAccess { + pub store: Arc, + pub snapshot: Arc, +} + +impl DefaultPluginResourceAccess { + async fn publish( + &self, + result: crate::ports::store::AdminStoreResult>, + ) -> Result { + let result = result.map_err(|error| super::map_store_error(error, "plugin resource"))?; + if let Some(revision) = result.revision { + super::publish_committed(self.snapshot.as_ref(), revision).await?; + } + Ok(result.value) + } +} + +#[async_trait] +impl PluginResourceAccess for DefaultPluginResourceAccess { + async fn ensure_group( + &self, + owner: &PluginResourceOwner, + resource_key: String, + command: CreateAccountGroup, + context: &MutationContext, + ) -> Result { + let id = AccountGroupId::new(format!("grp_{}", Uuid::now_v7().simple())) + .map_err(|_| AdminError::internal("创建账号组 ID 失败"))?; + self.publish( + self.store + .ensure_group( + owner, + resource_key, + NewAccountGroup { + id, + name: command.name, + description: command.description, + color: command.color, + disable_fast: command.disable_fast, + }, + context, + ) + .await, + ) + .await + } + + async fn ensure_key( + &self, + owner: &PluginResourceOwner, + resource_key: String, + groups: Vec, + command: ManagedKeyConfig, + context: &MutationContext, + ) -> Result { + let id = ClientApiKeyId::new(format!("key_{}", Uuid::now_v7().simple())) + .map_err(|_| AdminError::internal("创建 Key ID 失败"))?; + self.publish( + self.store + .ensure_key( + owner, + resource_key, + groups, + NewClientKey { + id, + name: command.name, + label: None, + group_ids: Vec::new(), + limits: command.limits, + budget: command.budget, + request_profile_overrides: Default::default(), + plaintext: super::client_keys::generate_key(), + }, + context, + ) + .await, + ) + .await + } + + async fn change_members( + &self, + owner: &PluginResourceOwner, + command: GroupMembersChange, + context: &MutationContext, + ) -> Result { + self.publish(self.store.change_members(owner, command, context).await) + .await + } +} diff --git a/backend/crates/gateway-admin/src/use_case/plugins/artifacts.rs b/backend/crates/gateway-admin/src/use_case/plugins/artifacts.rs index 64624ea8c..c85cdf0dd 100644 --- a/backend/crates/gateway-admin/src/use_case/plugins/artifacts.rs +++ b/backend/crates/gateway-admin/src/use_case/plugins/artifacts.rs @@ -377,7 +377,11 @@ pub(super) fn default_bindings(metadata: &PluginArtifactMetadata) -> Vec gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::ManagedResource, + >, + > { + unreachable!("resource port is not used by this fixture") + } + async fn ensure_key( + &self, + _: &gateway_admin::model::plugin_resources::PluginResourceOwner, + _: String, + _: Vec, + _: gateway_admin::model::client_keys::NewClientKey, + _: &gateway_admin::model::MutationContext, + ) -> gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::ManagedResource, + >, + > { + unreachable!("resource port is not used by this fixture") + } + async fn change_members( + &self, + _: &gateway_admin::model::plugin_resources::PluginResourceOwner, + _: gateway_admin::model::plugin_resources::GroupMembersChange, + _: &gateway_admin::model::MutationContext, + ) -> gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::GroupMembersChanged, + >, + > { + unreachable!("resource port is not used by this fixture") + } +} diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index f5b8704a9..0616a518a 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -169,6 +169,7 @@ impl AdminTestFixture { gateway_admin::ports::backup::BackupStorePorts::disabled(), plugin_ports.clone(), plugin_ports.clone(), + plugin_ports.clone(), ); let providers: Vec> = vec![ Arc::new(UnusedProvider::new("openai", Arc::clone(&provider_error))), diff --git a/backend/crates/gateway-api/tests/admin/plugins/mod.rs b/backend/crates/gateway-api/tests/admin/plugins/mod.rs index 4b6e668c3..78b84775b 100644 --- a/backend/crates/gateway-api/tests/admin/plugins/mod.rs +++ b/backend/crates/gateway-api/tests/admin/plugins/mod.rs @@ -1008,3 +1008,46 @@ async fn version_plan_requires_admin_and_an_explicit_target() { assert_eq!(response.headers()["cache-control"], "no-store"); } } + +#[async_trait::async_trait] +impl gateway_admin::ports::plugin_resources::PluginResourceStore for TestPluginPorts { + async fn ensure_group( + &self, + _: &gateway_admin::model::plugin_resources::PluginResourceOwner, + _: String, + _: gateway_admin::model::account_groups::NewAccountGroup, + _: &gateway_admin::model::MutationContext, + ) -> gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::ManagedResource, + >, + > { + unreachable!("resource port is not used by this fixture") + } + async fn ensure_key( + &self, + _: &gateway_admin::model::plugin_resources::PluginResourceOwner, + _: String, + _: Vec, + _: gateway_admin::model::client_keys::NewClientKey, + _: &gateway_admin::model::MutationContext, + ) -> gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::ManagedResource, + >, + > { + unreachable!("resource port is not used by this fixture") + } + async fn change_members( + &self, + _: &gateway_admin::model::plugin_resources::PluginResourceOwner, + _: gateway_admin::model::plugin_resources::GroupMembersChange, + _: &gateway_admin::model::MutationContext, + ) -> gateway_admin::ports::store::AdminStoreResult< + gateway_admin::model::plugin_resources::ResourceMutation< + gateway_admin::model::plugin_resources::GroupMembersChanged, + >, + > { + unreachable!("resource port is not used by this fixture") + } +} diff --git a/backend/crates/gateway-core/src/runtime/mod.rs b/backend/crates/gateway-core/src/runtime/mod.rs index 4483999fb..2635c7bd9 100644 --- a/backend/crates/gateway-core/src/runtime/mod.rs +++ b/backend/crates/gateway-core/src/runtime/mod.rs @@ -69,6 +69,7 @@ pub struct RuntimeSnapshotUnavailable; #[derive(Clone, Default)] pub struct RuntimeSnapshotHandle { current: Arc>>>, + publications: Arc>>>, } impl RuntimeSnapshotHandle { @@ -76,15 +77,38 @@ impl RuntimeSnapshotHandle { pub fn new(initial: RuntimeSnapshot) -> Self { Self { current: Arc::new(RwLock::new(Some(Arc::new(initial)))), + publications: Default::default(), } } pub fn publish(&self, snapshot: RuntimeSnapshot) { *write_unpoisoned(&self.current) = Some(Arc::new(snapshot)); + self.notify_publication(); } pub fn suspend(&self) { *write_unpoisoned(&self.current) = None; + self.notify_publication(); + } + + /// 合并发布与暂停通知；订阅者收到后读取当前快照，不依赖逐条投递。 + pub fn subscribe_publications(&self) -> futures::channel::mpsc::Receiver<()> { + let (sender, receiver) = futures::channel::mpsc::channel(0); + self.publications + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(sender); + receiver + } + + fn notify_publication(&self) { + self.publications + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .retain_mut(|sender| match sender.try_send(()) { + Ok(()) => true, + Err(error) => error.is_full(), + }); } #[must_use] diff --git a/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json b/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json index 30ce53997..40e4578eb 100644 --- a/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json +++ b/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json @@ -13,7 +13,8 @@ { "capability": "web_socket_observer", "versions": [1] }, { "capability": "usage", "versions": [1] }, { "capability": "command_line", "versions": [1] }, - { "capability": "management", "versions": [1] } + { "capability": "management", "versions": [1] }, + { "capability": "maintenance", "versions": [1] } ], "permissions": [ "network", @@ -21,6 +22,8 @@ "accounts", "data", "requests", - "public_endpoints" + "public_endpoints", + "groups", + "keys" ] } diff --git a/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs b/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs index c21c9fb93..9eab3f6d7 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs @@ -304,7 +304,7 @@ fn credential_facts( } } -fn mutation_context(context: &CallContext) -> MutationContext { +pub(super) fn mutation_context(context: &CallContext) -> MutationContext { MutationContext { actor: MutationActor::System, // RPC 回调使用真实 call_id，RPC 结果的后置提交仍使用原始上下文；scope 由宿主签发且全局唯一。 @@ -347,7 +347,7 @@ fn credential_timestamp(value: Option) -> Result>, Plu .transpose() } -fn encode(value: &impl serde::Serialize) -> Result { +pub(super) fn encode(value: &impl serde::Serialize) -> Result { Ok(RpcReply { result: serde_json::json!({}), payload: serde_json::to_vec(value).map_err(|_| invalid())?, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/data.rs b/backend/crates/gateway-plugin/runtime/src/callback/data.rs index aaae4ee47..fdc6b998d 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/data.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/data.rs @@ -37,7 +37,12 @@ impl PluginData { payload: &[u8], ) -> Result { // 管理范围的只读授权不继承到客户端请求链，避免借数据查询扩大当前 Key 的范围。 - if !self.authorized || !matches!(context.stage, Stage::Management | Stage::CommandLine) { + if !self.authorized + || !matches!( + context.stage, + Stage::Management | Stage::CommandLine | Stage::Maintenance + ) + { return Err(denied()); } if params != serde_json::json!({}) { diff --git a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs index fa3795f58..ab71d167d 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs @@ -7,6 +7,7 @@ mod log; mod middleware; mod model; pub(crate) mod private_state; +mod resources; mod scope; use std::{ @@ -16,7 +17,7 @@ use std::{ }; use futures::future::BoxFuture; -use gateway_admin::model::{AdminError, plugins::instances::PluginPermissionGrant}; +use gateway_admin::model::AdminError; use gateway_core::{account::OutboundProxy, lifecycle::CancellationToken}; use gateway_host::outbound::{HttpBody, HttpClient, NetworkPolicy}; use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault}; @@ -30,6 +31,7 @@ pub(crate) use middleware::{ MiddlewareBinding, MiddlewareBodyAuthority, MiddlewareCompletionBody, MiddlewareInvocation, }; pub(crate) use model::PluginModelPortSlot; +pub(crate) use resources::PluginResourcePorts; pub(crate) use scope::NetworkScope; pub(crate) struct PluginCallbackPorts { @@ -39,6 +41,7 @@ pub(crate) struct PluginCallbackPorts { keys: Arc, models: Arc, affinity: Arc, + resources: Arc, } impl PluginCallbackPorts { @@ -49,6 +52,7 @@ impl PluginCallbackPorts { keys: Arc, models: Arc, affinity: Arc, + resources: Arc, ) -> Self { Self { http, @@ -57,12 +61,14 @@ impl PluginCallbackPorts { keys, models, affinity, + resources, } } } pub(crate) struct PluginCallbacks { accounts: Arc, + resources: Arc, data: Arc, keys: Arc, models_authorized: bool, @@ -117,14 +123,16 @@ impl HttpStream { impl PluginCallbacks { pub(crate) fn new( - grants: &[PluginPermissionGrant], + instance: &gateway_admin::model::plugins::instances::PluginInstance, maximum_payload: usize, manifest: &gateway_plugin_sdk::Manifest, log_slots: Arc, private_state: Arc, ports: PluginCallbackPorts, ) -> Result { + let grants = &instance.grants; Ok(Self { + resources: Arc::new(resources::PluginResources::new(instance, ports.resources)), data: Arc::new(data::PluginData::new(ports.accounts.clone(), grants)), accounts: Arc::new(accounts::PluginAccounts::new(ports.accounts, grants)), keys: ports.keys, @@ -374,6 +382,7 @@ impl CallbackHandler for PluginCallbacks { let log = self.log.clone(); let private_state = self.private_state.clone(); let accounts = self.accounts.clone(); + let resources = self.resources.clone(); let data = self.data.clone(); let keys = self.keys.clone(); let models_authorized = self.models_authorized; @@ -417,6 +426,14 @@ impl CallbackHandler for PluginCallbacks { if is_private_state { return private_state.call(&method, params, &payload).await; } + if matches!( + method.as_str(), + gateway_plugin_sdk::call::resources::GROUP_ENSURE + | gateway_plugin_sdk::call::resources::GROUP_MEMBERS + | gateway_plugin_sdk::call::resources::KEY_ENSURE + ) { + return resources.call(&context, &method, params, &payload).await; + } if method == "host.keys.list" { if !models_authorized { return Err(denied()); diff --git a/backend/crates/gateway-plugin/runtime/src/callback/resources.rs b/backend/crates/gateway-plugin/runtime/src/callback/resources.rs new file mode 100644 index 000000000..9a77f6626 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/callback/resources.rs @@ -0,0 +1,147 @@ +use super::{ + accounts::{encode, map_admin_error, mutation_context}, + denied, invalid, +}; +use crate::RpcReply; +use gateway_admin::{ + model::{ + AdminError, + account_groups::{AccountGroupColor, CreateAccountGroup}, + plugin_resources::{GroupMembersChange, ManagedKeyConfig, PluginResourceOwner}, + plugins::instances::PluginInstance, + }, + ports::plugin_resources::PluginResourceAccess, +}; +use gateway_core::{engine::budget::ClientBudgetLimits, policy::RateLimits}; +use gateway_plugin_sdk::{CallContext, PluginFault, call::resources}; +use std::sync::{Arc, OnceLock, Weak}; + +pub(crate) struct PluginResourcePorts(OnceLock>); +impl PluginResourcePorts { + pub(crate) const fn new() -> Self { + Self(OnceLock::new()) + } + pub(crate) fn bind(&self, port: &Arc) -> Result<(), AdminError> { + self.0 + .set(Arc::downgrade(port)) + .map_err(|_| AdminError::conflict("插件资源端口已经绑定")) + } +} + +pub(super) struct PluginResources { + owner: PluginResourceOwner, + ports: Arc, +} + +impl PluginResources { + pub(super) fn new(instance: &PluginInstance, ports: Arc) -> Self { + Self { + owner: PluginResourceOwner { + instance_id: instance.id.clone(), + artifact_sha256: instance.artifact_sha256.clone(), + revision: instance.revision, + }, + ports, + } + } + + pub(super) async fn call( + &self, + context: &CallContext, + method: &str, + params: serde_json::Value, + payload: &[u8], + ) -> Result { + if params != serde_json::json!({}) { + return Err(invalid()); + } + let access = self + .ports + .0 + .get() + .and_then(Weak::upgrade) + .ok_or_else(denied)?; + let mutation = mutation_context(context); + match method { + resources::GROUP_ENSURE => { + let request: resources::GroupEnsureRequest = + serde_json::from_slice(payload).map_err(|_| invalid())?; + let value = access + .ensure_group( + &self.owner, + request.resource_key, + CreateAccountGroup { + disable_fast: false, + name: request.name, + description: request.description, + color: AccountGroupColor::parse(&request.color).ok_or_else(invalid)?, + }, + &mutation, + ) + .await + .map_err(map_admin_error)?; + encode(&resources::ManagedResource { + id: value.id, + name: value.name, + enabled: value.enabled, + }) + } + resources::KEY_ENSURE => { + let request: resources::KeyEnsureRequest = + serde_json::from_slice(payload).map_err(|_| invalid())?; + let value = access + .ensure_key( + &self.owner, + request.resource_key, + request.group_resource_keys, + ManagedKeyConfig { + name: request.name, + limits: RateLimits { + max_concurrency: request.max_concurrency, + requests_per_minute: request.requests_per_minute, + }, + budget: ClientBudgetLimits { + daily_usd: request + .daily_limit_usd + .parse() + .map_err(|_| invalid())?, + weekly_usd: request + .weekly_limit_usd + .parse() + .map_err(|_| invalid())?, + }, + }, + &mutation, + ) + .await + .map_err(map_admin_error)?; + encode(&resources::ManagedResource { + id: value.id, + name: value.name, + enabled: value.enabled, + }) + } + resources::GROUP_MEMBERS => { + let request: resources::GroupMembersChange = + serde_json::from_slice(payload).map_err(|_| invalid())?; + let value = access + .change_members( + &self.owner, + GroupMembersChange { + resource_key: request.resource_key, + add: request.add, + remove: request.remove, + }, + &mutation, + ) + .await + .map_err(map_admin_error)?; + encode(&resources::GroupMembersChanged { + added: value.added, + removed: value.removed, + }) + } + _ => Err(invalid()), + } + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/generation/configuration.rs b/backend/crates/gateway-plugin/runtime/src/generation/configuration.rs index a8da3d8c9..02cfba4a2 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/configuration.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/configuration.rs @@ -38,8 +38,11 @@ pub(super) fn validate( for binding in &instance.bindings { let resolved = crate::contribution::resolve(manifest, binding)?; let capability = resolved.capability; - if matches!(capability, Capability::Management | Capability::CommandLine) { - return Err(AdminError::invalid("管理页面与命令行无需功能绑定")); + if matches!( + capability, + Capability::Management | Capability::CommandLine | Capability::Maintenance + ) { + return Err(AdminError::invalid("管理页面、命令行与维护无需功能绑定")); } let stage: Stage = serde_json::from_value(serde_json::Value::String(binding.stage.clone())) .map_err(|_| AdminError::invalid("插件调用阶段不合法"))?; diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs index fa7c20d98..259ba7083 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs @@ -34,7 +34,6 @@ impl PluginRuntime { let limits = self.config.package_limits; let directory = self.config.cache_directory.clone(); let host_version = self.config.host_version.clone(); - let grants = instance.grants.clone(); let validation_slot = self .validators .clone() @@ -145,7 +144,7 @@ impl PluginRuntime { contributes: manifest.contributes.clone(), }; let callbacks = Arc::new(PluginCallbacks::new( - &grants, + &instance, self.config.rpc_limits.maximum_buffered_body_bytes, &manifest, self.log_slots.clone(), @@ -157,6 +156,7 @@ impl PluginRuntime { self.client_key_ports.clone(), self.model_ports.clone(), self.affinity_ports.clone(), + self.resource_ports.clone(), ), )?); let session = Arc::new( @@ -241,6 +241,9 @@ impl PluginRuntime { revision: instance.revision, session, private_state, + maintenance: manifest + .contributes + .contains_key(&gateway_plugin_sdk::Capability::Maintenance), }); Ok(PreparedContributions { sessions, diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs new file mode 100644 index 000000000..036146533 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs @@ -0,0 +1,141 @@ +//! 对账只从已发布集合开始；通知合并，实例串行，失败和停用不阻塞其他实例。 + +use super::{PluginRuntime, PreparedSet, RpcSession}; +use futures::{StreamExt as _, future::BoxFuture}; +use gateway_core::{ + lifecycle::CancellationToken, + runtime::RuntimeSnapshotHandle, + task::{ + DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerDefinitionError, WorkerId, + WorkerKind, WorkerRegistration, WorkerRunnable, WorkerTaskError, + }, +}; +use gateway_plugin_sdk::Stage; +use std::{sync::Arc, time::Duration}; +use tokio::{sync::Notify, task::JoinSet}; + +const RECONCILE_INTERVAL: Duration = Duration::from_secs(30); +const RETRY_DELAY: Duration = Duration::from_secs(5); +const CALL_TIMEOUT: Duration = Duration::from_secs(30); + +struct MaintenanceTask { + runtime: Arc, + snapshots: RuntimeSnapshotHandle, +} + +impl PluginRuntime { + /// 由 Host 监督维护任务；CLI 和只读准备不会启动后台业务。 + pub fn maintenance_worker( + self: &Arc, + snapshots: RuntimeSnapshotHandle, + ) -> Result { + Ok(WorkerContribution::Registration( + WorkerRegistration::try_new( + WorkerId::try_new( + WorkerKind::RuntimeSnapshotReconciliation, + "plugin-maintenance", + )?, + WorkerRunnable::Daemon { + restart: DaemonRestartPolicy::try_new( + Duration::from_secs(1), + Duration::from_secs(30), + )?, + task: Box::new(MaintenanceTask { + runtime: self.clone(), + snapshots, + }), + }, + )?, + )) + } +} + +impl DaemonTask for MaintenanceTask { + fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { + let mut publications = self.snapshots.subscribe_publications(); + let mut active: Option> = None; + let mut revision = None; + let mut stop = CancellationToken::new(); + let mut tasks = JoinSet::new(); + let mut notifications = Vec::new(); + loop { + let snapshot = self.snapshots.acquire().ok(); + let reference = snapshot.as_ref().and_then(|s| s.extensions()); + let same_set = active.as_ref().map(|s| &s.id) == reference.map(|s| s.id()); + if !same_set { + stop.cancel(); + while tasks.join_next().await.is_some() {} + notifications.clear(); + stop = CancellationToken::new(); + active = match reference { + Some(reference) => { + Some(self.runtime.prepared_set(reference).await.map_err(|_| { + WorkerTaskError::safe("published plugin set is unavailable") + })?) + } + None => None, + }; + if let Some(set) = &active { + for instance in set.sessions.iter().filter(|i| i.maintenance) { + let dirty = Arc::new(Notify::new()); + dirty.notify_one(); + notifications.push(dirty.clone()); + tasks.spawn(reconcile(instance.session.clone(), dirty, stop.clone())); + } + } + } else if revision != snapshot.as_ref().map(|s| s.revision()) { + for dirty in ¬ifications { + dirty.notify_one(); + } + } + revision = snapshot.as_ref().map(|s| s.revision()); + // 发布变更优先于下一轮维护；不在快照发布锁中等待插件或数据库。 + tokio::select! { + biased; + () = cancellation.cancelled() => break, + changed = publications.next() => if changed.is_none() { break; }, + result = tasks.join_next(), if !tasks.is_empty() => { + if result.is_some() { + stop.cancel(); + while tasks.join_next().await.is_some() {} + return Err(WorkerTaskError::safe("plugin maintenance task stopped unexpectedly")); + } + } + } + } + stop.cancel(); + while tasks.join_next().await.is_some() {} + Ok(()) + }) + } +} + +async fn reconcile(session: Arc, dirty: Arc, cancellation: CancellationToken) { + loop { + tokio::select! { + biased; + () = cancellation.cancelled() => return, + () = dirty.notified() => {}, + () = tokio::time::sleep(RECONCILE_INTERVAL) => {}, + } + let context = session.context(Stage::Maintenance, CALL_TIMEOUT); + let instance_id = context.instance_id.clone(); + let result = tokio::select! { + biased; + () = cancellation.cancelled() => return, + result = session.call("plugin.reconcile", context, serde_json::json!({}), Vec::new()) => result, + }; + if !result + .is_ok_and(|reply| reply.result == serde_json::json!({}) && reply.payload.is_empty()) + { + tracing::warn!(%instance_id, "plugin maintenance failed; retrying after bounded delay"); + tokio::select! { + biased; + () = cancellation.cancelled() => return, + () = tokio::time::sleep(RETRY_DELAY) => {}, + } + dirty.notify_one(); + } + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/mod.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/mod.rs index 3bf43bec6..704679123 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/mod.rs @@ -1,4 +1,5 @@ mod instance; +mod maintenance; use std::{ collections::{BTreeMap, BTreeSet}, @@ -95,6 +96,7 @@ pub struct PluginRuntime { log_slots: Arc, pub(super) account_ports: Arc, client_key_ports: Arc, + resource_ports: Arc, model_ports: Arc, affinity_ports: Arc, observers: RequestObserverExtensionIndex, @@ -159,6 +161,7 @@ struct PreparedInstance { revision: Revision, session: Arc, private_state: Arc, + maintenance: bool, } impl PreparedSet { @@ -228,6 +231,7 @@ impl PluginRuntime { validators: Arc::new(Semaphore::new(2)), account_ports: Arc::new(PluginAccountPortSlot::new()), client_key_ports: Arc::new(PluginClientKeyPortSlot::new()), + resource_ports: Arc::new(crate::callback::PluginResourcePorts::new()), model_ports: Arc::new(PluginModelPortSlot::new()), affinity_ports: Arc::new(PluginAffinityPortSlot::new()), observers: RequestObserverExtensionIndex::default(), @@ -1219,3 +1223,12 @@ fn instance_fingerprint( let bytes = serde_json::to_vec(&value).map_err(|_| AdminError::invalid("插件配置无法编码"))?; Ok(hex::encode(Sha256::digest(bytes))) } + +impl PluginRuntime { + pub fn bind_resource_ports( + &self, + access: &Arc, + ) -> Result<(), AdminError> { + self.resource_ports.bind(access) + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs index 4953fd853..ece340046 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs @@ -220,6 +220,18 @@ fn callback_allowed(method: &str, stage: Stage, permissions: &[Permission]) -> b ) { return matches!(stage, Stage::Request | Stage::Attempt); } + if stage == Stage::Maintenance + && !matches!( + method, + "host.data.accounts.list" + | "host.data.quota.get" + | "host.groups.ensure" + | "host.groups.change_members" + | "host.keys.ensure" + ) + { + return false; + } let permission = match method { "host.http.do" | "host.http.do_stream" @@ -236,10 +248,29 @@ fn callback_allowed(method: &str, stage: Stage, permissions: &[Permission]) -> b } "host.affinity.lookup" => Permission::Requests, "host.data.accounts.list" | "host.data.quota.get" - if matches!(stage, Stage::Management | Stage::CommandLine) => + if matches!( + stage, + Stage::Management | Stage::CommandLine | Stage::Maintenance + ) => { Permission::Data } + "host.groups.ensure" | "host.groups.change_members" + if matches!( + stage, + Stage::Management | Stage::CommandLine | Stage::Maintenance + ) => + { + Permission::Groups + } + "host.keys.ensure" + if matches!( + stage, + Stage::Management | Stage::CommandLine | Stage::Maintenance + ) => + { + Permission::Keys + } _ => return false, }; permissions.contains(&permission) diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs b/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs new file mode 100644 index 000000000..a2aa674c8 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs @@ -0,0 +1,432 @@ +use crate::support::environment::{Environment, account_grant, mutation}; +use gateway_admin::{ + model::{ + account_groups::{AccountGroupColor, CreateAccountGroup}, + plugin_resources::{GroupMembersChange, ManagedKeyConfig, PluginResourceOwner}, + }, + ports::plugin_resources::PluginResourceAccess, +}; +use gateway_core::{ + lifecycle::CancellationToken, + task::{WorkerContribution, WorkerRunnable}, +}; +use gateway_plugin_runtime::PluginRuntime; +use serde_json::{Value, json}; +use std::{sync::Arc, time::Duration}; + +fn group(name: &str) -> CreateAccountGroup { + CreateAccountGroup { + name: name.into(), + description: None, + disable_fast: false, + color: AccountGroupColor::parse("#2563EBFF").unwrap(), + } +} + +fn key() -> ManagedKeyConfig { + ManagedKeyConfig { + name: " managed key ".into(), + limits: gateway_core::policy::RateLimits::unlimited(), + budget: Default::default(), + } +} + +async fn owner(environment: &Environment) -> PluginResourceOwner { + let instance = environment + .store + .admin_ports() + .plugins() + .load_instances() + .await + .unwrap() + .instances + .remove(0); + PluginResourceOwner { + instance_id: instance.id, + artifact_sha256: instance.artifact_sha256, + revision: instance.revision, + } +} + +async fn import_account( + environment: &Environment, + core: &gateway_core::CoreBundle, +) -> gateway_core::account::ProviderAccountId { + use gateway_admin::model::provider_credentials::{ + PreparedCredentialCreate, PreparedPluginAccountSave, ProviderDocument, + }; + let access = gateway_admin::initialize_plugin_accounts( + crate::support::native::admin_registry(), + environment.store.admin_ports().accounts(), + core.snapshot_control(), + ); + let result = access + .save( + PreparedPluginAccountSave::Create(PreparedCredentialCreate { + account_id: gateway_core::account::ProviderAccountId::new(format!( + "acct_{}", + uuid::Uuid::new_v4().simple() + )) + .unwrap(), + provider_kind: gateway_core::routing::ProviderKind::new("openai").unwrap(), + name: "imported maintenance fixture".into(), + email: None, + upstream_user_id: None, + upstream_account_id: None, + plan_type: None, + authentication_kind: "api_key".into(), + provider_material: ProviderDocument::new( + gateway_core::account::OpaqueProviderData::new( + json!({"key":"test-only"}).as_object().unwrap().clone(), + ), + ), + model_access: None, + outbound_proxy: None, + has_refresh_token: false, + access_token_expires_at: None, + next_refresh_at: None, + enabled: true, + credential_state: gateway_core::account::CredentialState::Ready, + credential_observed_at: chrono::Utc::now(), + }), + &mutation(), + ) + .await + .unwrap(); + result.account_id +} + +async fn revision(environment: &Environment) -> gateway_admin::model::Revision { + environment + .store + .admin_ports() + .plugins() + .load_instances() + .await + .unwrap() + .config_revision +} + +async fn start( + environment: &Environment, +) -> ( + Arc, + gateway_core::CoreBundle, + Arc, + CancellationToken, + tokio::task::JoinHandle<()>, +) { + let (runtime, core) = environment.runtime().await; + let resources = gateway_admin::initialize_plugin_resources( + environment.store.admin_ports().plugin_resources(), + core.snapshot_control(), + ); + runtime.bind_resource_ports(&resources).unwrap(); + let WorkerContribution::Registration(registration) = + runtime.maintenance_worker(core.snapshots()).unwrap() + else { + panic!("maintenance registration") + }; + let WorkerRunnable::Daemon { task, .. } = registration.runnable else { + panic!("maintenance daemon") + }; + let stop = CancellationToken::new(); + let cancellation = stop.clone(); + let task = tokio::spawn(async move { + task.run(cancellation).await.unwrap(); + }); + (runtime, core, resources, stop, task) +} + +async fn wait_done(path: &std::path::Path, after: usize) -> Value { + tokio::time::timeout(Duration::from_secs(15), async { + loop { + let entries: Vec = std::fs::read_to_string(path) + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str(line).ok()) + .collect(); + let done: Vec<_> = entries + .iter() + .filter(|value| value["phase"] == "done") + .collect(); + if done.len() > after { + return (*done.last().unwrap()).clone(); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("plugin reconciliation finished") +} + +#[tokio::test] +async fn published_reconciliation_provisions_retries_imports_and_restores_without_overwriting_groups() + { + let Some(environment) = Environment::create().await else { + return; + }; + let account = environment.account(None).await; + let original = environment.account_group_with_account(&account).await; + let marker = environment.directory.path().join("maintenance.jsonl"); + let failed = environment.directory.path().join("failed-once"); + environment.install_plugin(json!({"maintenance_fixture":true, "maintenance_marker":marker, "maintenance_fail_once":failed}), vec![account_grant("groups"), account_grant("keys"), account_grant("data")]).await; + let (runtime, core, resources, stop, task) = start(&environment).await; + let first = wait_done(&marker, 0).await; + assert!(failed.exists()); + let group_id = + gateway_core::routing::AccountGroupId::new(first["group"].as_str().unwrap()).unwrap(); + let members = environment + .store + .admin_ports() + .account_groups() + .load_account_group_members(&[original.clone(), group_id.clone()]) + .await + .unwrap(); + assert_eq!( + members + .iter() + .filter(|m| m.account_id == account.as_str()) + .count(), + 2 + ); + let owner = owner(&environment).await; + let before = revision(&environment).await; + let same = resources + .ensure_group( + &owner, + "pool".into(), + group("ignored replacement name"), + &mutation(), + ) + .await + .unwrap(); + assert_eq!(same.id, first["group"]); + assert_eq!(same.name, "plugin fixture group"); + assert_eq!(before, revision(&environment).await); + let imported = import_account(&environment, &core).await; + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let members = environment + .store + .admin_ports() + .account_groups() + .load_account_group_members(std::slice::from_ref(&group_id)) + .await + .unwrap(); + if members + .iter() + .any(|member| member.account_id == imported.as_str()) + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + stop.cancel(); + task.await.unwrap(); + runtime.shutdown().await; + environment.release_plugin_accounts(&runtime); + drop(resources); + drop(core); + drop(runtime); + // 离线新增账号后重新启动真实插件子进程；稳定资源键必须复用原分组与 Key。 + let offline = environment.account(None).await; + std::fs::write(&marker, "").unwrap(); + let (runtime, core, resources, stop, task) = start(&environment).await; + let restored = wait_done(&marker, 0).await; + assert_eq!(restored["group"], first["group"]); + assert_eq!(restored["key"], first["key"]); + let members = environment + .store + .admin_ports() + .account_groups() + .load_account_group_members(std::slice::from_ref(&group_id)) + .await + .unwrap(); + assert!(members.iter().any(|m| m.account_id == offline.as_str())); + let store = environment.store.admin_ports().plugins(); + let snapshot = store.load_instances().await.unwrap(); + let mut instance = snapshot.instances[0].clone(); + instance.enabled = false; + let disabled = store + .save_instance(instance, snapshot.config_revision, &mutation()) + .await + .unwrap(); + core.snapshot_control() + .publish_committed( + gateway_core::routing::ConfigRevision::new(disabled.config_revision.get()).unwrap(), + ) + .await; + assert!( + resources + .change_members( + &owner, + GroupMembersChange { + resource_key: "pool".into(), + add: vec![], + remove: vec![account.as_str().to_owned()] + }, + &mutation() + ) + .await + .is_err() + ); + stop.cancel(); + task.await.unwrap(); + runtime.shutdown().await; + environment.release_plugin_accounts(&runtime); + drop(resources); + drop(core); + drop(runtime); + environment.close().await; +} + +#[tokio::test] +async fn resource_transactions_enforce_ownership_current_revision_and_idempotent_membership() { + let Some(environment) = Environment::create().await else { + return; + }; + environment + .install_plugin( + json!({}), + vec![account_grant("groups"), account_grant("keys")], + ) + .await; + let (runtime, core) = environment.runtime().await; + let resources = gateway_admin::initialize_plugin_resources( + environment.store.admin_ports().plugin_resources(), + core.snapshot_control(), + ); + let owner = owner(&environment).await; + let account = environment.account(None).await; + let concurrent_context = mutation(); + let (left, right) = tokio::join!( + resources.ensure_group( + &owner, + "one".into(), + group("plugin one"), + &concurrent_context + ), + resources.ensure_group( + &owner, + "one".into(), + group("concurrent name"), + &concurrent_context + ) + ); + let one = left.unwrap(); + assert_eq!(one, right.unwrap()); + let other = environment.account_group_with_account(&account).await; + assert!( + resources + .ensure_group( + &owner, + "collision".into(), + group(&format!("fixture {}", other.as_str())), + &mutation() + ) + .await + .is_err() + ); + assert!( + resources + .ensure_key(&owner, "key".into(), vec![], key(), &mutation()) + .await + .is_err() + ); + assert!( + resources + .ensure_key( + &owner, + "key".into(), + vec![other.as_str().to_owned()], + key(), + &mutation() + ) + .await + .is_err() + ); + let created = resources + .ensure_key(&owner, "key".into(), vec!["one".into()], key(), &mutation()) + .await + .unwrap(); + assert_eq!(created.name, "managed key"); + assert_eq!( + created, + resources + .ensure_key(&owner, "key".into(), vec!["one".into()], key(), &mutation()) + .await + .unwrap() + ); + let change = || GroupMembersChange { + resource_key: "one".into(), + add: vec![account.as_str().to_owned()], + remove: vec![], + }; + assert_eq!( + resources + .change_members(&owner, change(), &mutation()) + .await + .unwrap() + .added, + 1 + ); + let before = revision(&environment).await; + assert_eq!( + resources + .change_members(&owner, change(), &mutation()) + .await + .unwrap() + .added, + 0 + ); + assert_eq!(revision(&environment).await, before); + let result = resources + .change_members( + &owner, + GroupMembersChange { + resource_key: "one".into(), + add: vec![], + remove: vec![account.as_str().to_owned()], + }, + &mutation(), + ) + .await + .unwrap(); + assert_eq!(result.removed, 1); + let members = environment + .store + .admin_ports() + .account_groups() + .load_account_group_members(&[ + other.clone(), + gateway_core::routing::AccountGroupId::new(one.id).unwrap(), + ]) + .await + .unwrap(); + assert_eq!(members.len(), 1); + assert_eq!(members[0].group_id, other); + let store = environment.store.admin_ports().plugins(); + let snapshot = store.load_instances().await.unwrap(); + let mut instance = snapshot.instances[0].clone(); + instance.name = "new configuration revision".into(); + store + .save_instance(instance, snapshot.config_revision, &mutation()) + .await + .unwrap(); + assert!( + resources + .ensure_group(&owner, "stale".into(), group("stale"), &mutation()) + .await + .is_err() + ); + runtime.shutdown().await; + environment.release_plugin_accounts(&runtime); + drop(resources); + drop(core); + drop(runtime); + environment.close().await; +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/mod.rs index b55a81894..9d6f2dcd4 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/mod.rs @@ -1,6 +1,7 @@ mod authentication; mod configuration; mod data; +mod maintenance; mod observer; mod policy; mod prepare; diff --git a/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs b/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs index 758b77c9e..59c9490cb 100644 --- a/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs +++ b/backend/crates/gateway-plugin/runtime/tests/package/compatibility.rs @@ -82,7 +82,8 @@ fn assert_exhaustive_capability(capability: Capability) { | Capability::WebSocketObserver | Capability::Usage | Capability::CommandLine - | Capability::Management => {} + | Capability::Management + | Capability::Maintenance => {} } } @@ -93,11 +94,13 @@ fn assert_exhaustive_permission(permission: Permission) { | Permission::Models | Permission::Accounts | Permission::Data - | Permission::PublicEndpoints => {} + | Permission::PublicEndpoints + | Permission::Groups + | Permission::Keys => {} } } -fn all_capabilities() -> [Capability; 11] { +fn all_capabilities() -> [Capability; 12] { [ Capability::FrontendAuthentication, Capability::Scheduler, @@ -110,5 +113,6 @@ fn all_capabilities() -> [Capability; 11] { Capability::Usage, Capability::CommandLine, Capability::Management, + Capability::Maintenance, ] } diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/dispatch.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/dispatch.rs index 1e82d4efc..bd5fe00fa 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/dispatch.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/dispatch.rs @@ -289,3 +289,72 @@ async fn retry_stage_allows_only_logs_and_private_state() { assert_eq!(callbacks.called.load(Ordering::Relaxed), 2); session.shutdown(Duration::from_secs(1)).await; } + +#[tokio::test] +async fn managed_resources_require_their_domains_and_control_plane_stages() { + for (method, permission) in [ + ("host.groups.ensure", Permission::Groups), + ("host.groups.change_members", Permission::Groups), + ("host.keys.ensure", Permission::Keys), + ] { + for granted in [false, true] { + let callbacks = Arc::new(Callbacks::default()); + let (_cache, session) = session_with_permissions( + callbacks.clone(), + if granted { vec![permission] } else { vec![] }, + ) + .await; + for stage in [ + Stage::Registration, + Stage::Configuration, + Stage::PublicManagement, + Stage::Request, + Stage::Observation, + Stage::Management, + Stage::CommandLine, + Stage::Maintenance, + ] { + let reply = invoke_callback(&session, stage, method).await; + if granted + && matches!( + stage, + Stage::Management | Stage::CommandLine | Stage::Maintenance + ) + { + assert!(reply.is_ok()); + } else { + assert_permission_denied(reply); + } + } + session.shutdown(Duration::from_secs(1)).await; + } + } +} + +#[tokio::test] +async fn maintenance_has_data_and_state_access_but_cannot_execute_models_or_read_credentials() { + let callbacks = Arc::new(Callbacks::default()); + let (_cache, session) = session(callbacks).await; + for method in [ + "host.data.accounts.list", + "host.data.quota.get", + "host.state.put", + "host.log", + ] { + assert!( + invoke_callback(&session, Stage::Maintenance, method) + .await + .is_ok() + ); + } + for method in [ + "host.auth.get", + "host.auth.save", + "host.http.do", + "host.model.execute", + "host.keys.list", + ] { + assert_permission_denied(invoke_callback(&session, Stage::Maintenance, method).await); + } + session.shutdown(Duration::from_secs(1)).await; +} diff --git a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs index 77a0d583a..036e9f17a 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs @@ -167,6 +167,16 @@ impl Environment { .map(|grant| serde_json::from_value(json!(grant.permission)).unwrap()) .collect(); let mut contributes = Contributions::new(); + if configuration.get("maintenance_fixture").is_some() { + contributes.extend([super::contribution_for_id( + &plugin_id, + Capability::Maintenance, + vec![Stage::Maintenance], + vec![], + vec![], + )]); + } + if configuration.get("command_registration").is_some() { contributes.extend([super::contribution_for_id( &plugin_id, diff --git a/backend/crates/gateway-plugin/runtime/tests/support/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/mod.rs index e14d8c214..05fcf7ee0 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/mod.rs @@ -50,6 +50,7 @@ pub fn contribution_for_id( Capability::Usage => "usage", Capability::CommandLine => "commandLine", Capability::Management => "management", + Capability::Maintenance => "maintenance", }; ( capability, diff --git a/backend/crates/gateway-plugin/runtime/tests/support/worker.rs b/backend/crates/gateway-plugin/runtime/tests/support/worker.rs index 169abac8b..15730d021 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/worker.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/worker.rs @@ -81,6 +81,68 @@ struct Peer { } impl Peer { + async fn resource_fixture( + &self, + id: u64, + method: &str, + value: Value, + ) -> Result { + let (result, payload) = self + .callback_payload(id, method, json!({}), serde_json::to_vec(&value).unwrap()) + .await?; + assert_eq!(result, json!({})); + Ok(serde_json::from_slice(&payload).unwrap()) + } + + async fn reconcile_fixture(&self, id: u64) -> Result<(), PluginFault> { + self.append_observation_marker("maintenance_marker", &json!({"phase":"start"})); + let group = self + .resource_fixture( + id, + "host.groups.ensure", + json!({"resource_key":"pool", "name":"plugin fixture group", "color":"#2563EBFF"}), + ) + .await?; + if let Some(path) = self.configuration["maintenance_fail_once"].as_str() + && !std::path::Path::new(path).exists() + { + std::fs::write(path, "failed").unwrap(); + return Err(PluginFault::new(ErrorCode::Fault, "fixture retry")); + } + let mut cursor = Value::Null; + loop { + let page = self + .resource_fixture( + id, + "host.data.accounts.list", + json!({"cursor":cursor, "limit":100}), + ) + .await?; + let accounts: Vec<_> = page["accounts"] + .as_array() + .unwrap() + .iter() + .map(|account| account["account_id"].clone()) + .collect(); + self.resource_fixture( + id, + "host.groups.change_members", + json!({"resource_key":"pool", "add":accounts}), + ) + .await?; + cursor = page["next_cursor"].clone(); + if cursor.is_null() { + break; + } + } + let key = self.resource_fixture(id, "host.keys.ensure", json!({"resource_key":"key", "name":"plugin fixture key", "group_resource_keys":["pool"], "daily_limit_usd":"1", "weekly_limit_usd":"5"})).await?; + self.append_observation_marker( + "maintenance_marker", + &json!({"phase":"done", "group":group["id"], "key":key["id"]}), + ); + Ok(()) + } + async fn log_fixture(&self, id: u64) { let Some(entries) = self .configuration @@ -297,6 +359,24 @@ impl Peer { return; } match method.as_str() { + "plugin.reconcile" => { + let result = self.reconcile_fixture(id).await; + match result { + Ok(()) => { + self.send( + Message::Result { + id, + result: json!({}), + }, + vec![], + ) + .await + } + Err(error) => self.send(Message::Error { id, error }, vec![]).await, + } + return; + } + "policy.retry_decision" => { self.append_observation_marker("retry_marker", ¶ms); if let Some(delay) = self.configuration["retry_delay_ms"].as_u64() { diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index 5a4e53353..9003f1c27 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -54,6 +54,7 @@ Provider 固定为宿主内置的 OpenAI 与 xAI。插件提供以下扩展能 | `frontend_authentication` | `methods::FRONTEND_IDENTIFIER`、`methods::FRONTEND_AUTHENTICATE` | | `management` | `PluginBuilder::management`；公开回调另用 `methods::MANAGEMENT_CALLBACK` | | `command_line` | `PluginBuilder::command_line` | +| `maintenance` | `methods::RECONCILE`；无需功能绑定 | ## 访问域 @@ -65,9 +66,11 @@ Provider 固定为宿主内置的 OpenAI 与 xAI。插件提供以下扩展能 | `network` | 通过 `host.http.*` 使用宿主受管出站网络；仍受统一代理、超时、大小和流控规则约束 | | `models` | 列出非秘密 Key、按所选 Key 查询模型，以及通过 `host.model.*` 调用模型；调用可能产生消耗 | | `accounts` | 查询账号、读取原始凭据及创建或替换账号；写入仍经过 revision CAS、审计和发布事务 | -| `data` | 在管理／命令阶段只读全部账号的最小基础信息及已有额度观测，不包含凭据、写入或预测 | +| `data` | 在管理／命令／维护阶段只读全部账号的最小基础信息及已有额度观测，不包含凭据、写入或预测 | | `requests` | 参与请求／响应处理、路由、调度、观察及亲和查询 | | `public_endpoints` | 提供无需登录即可访问的已声明静态资源或一次性票据回调 | +| `groups` | 创建本实例分组，允许将所有现有及未来新增账号加入或移出这些分组，保留其他分组关系 | +| `keys` | 创建仅绑定本实例分组的 Key，返回非秘密身份；不授予其他 Key 的修改或明文读取权限 | `host.log` 和本插件声明的 `host.state.*` 是基础设施，不需要额外 permission。公开调用阶段不开放任何 宿主回调；权限也不能把一个父调用的句柄、流或上下文转移到另一个调用。 @@ -79,7 +82,7 @@ Provider 固定为宿主内置的 OpenAI 与 xAI。插件提供以下扩展能 ### 基础事实 纯展示页面声明 `management` 和 `data` 即可，不需要请求处理 binding 或 `accounts` 权限。 -`data` 是独立的管理员授权域，只允许 `management`、`command_line` 阶段使用；请求链、注册及公开回调均拒绝。 +`data` 是独立的管理员授权域，只允许 `management`、`command_line`、`maintenance` 阶段使用；请求链、注册及公开回调均拒绝。 它可读取全部账号的下列最小事实，不继承或授予客户端 Key 的模型执行权。 | SDK 方法 | 回调 | 查询与结果 | @@ -118,6 +121,34 @@ for account in page.accounts { 插件可以自行选择其业务需要的账号和 Provider，不需要安装器预先配置允许列表。宿主仍校验账号归属、资格、 凭据版本和并发事务；原始凭据不得进入日志、审计正文或控制元数据。 +### 自有资源与维护 + +声明 `maintenance` 并注册 `methods::RECONCILE`，宿主会在实例启用发布、进程恢复及配置变更后调用 +`plugin.reconcile`，并每 30 秒补偿一次。处理器必须幂等；调用可能重复，配置通知会合并，不表示逐条账号事件。 +每个实例串行执行，不同实例相互独立；调用限时 30 秒，失败后等待 5 秒重试。维护失败不回滚已经提交的资源， +下一次对账继续补齐。只读校验、准备候选与 CLI 帮助不会启动维护；停用、替换和宿主关闭时取消旧任务。 + +维护阶段允许日志、私有状态，以及已授权的 `data`、`groups`、`keys` 回调；不开放网络、凭据或模型执行。 +其他管理／命令入口也可使用下列资源方法： + +| SDK 方法 | 参数与行为 | +| --- | --- | +| `call.host.ensure_group(GroupEnsureRequest)` | `resource_key`、`name`、`color`、可选 `description`；首次创建，之后返回已有 `{id,name,enabled}` | +| `call.host.change_group_members(GroupMembersChange)` | `resource_key`、`add`、`remove`；两列表合计最多 200 个账号 ID，不得重复或交叉，返回实际 `added`／`removed` 数量 | +| `call.host.ensure_key(KeyEnsureRequest)` | `resource_key`、`name`、非空 `group_resource_keys`（最多 64 个）、`max_concurrency`、`requests_per_minute`、`daily_limit_usd`、`weekly_limit_usd`；返回 `{id,name,enabled}`，预算零值沿用原生无限制语义 | + +资源键使用 1～64 位小写字母、数字、`_`、`.`、`-`，首位为字母或数字，在实例和资源类型内唯一。 +归属由宿主签发，插件不能指定其他实例。`ensure` 只在首次创建时使用属性，不覆盖管理员对名称、启用状态或限制的修改； +同名的管理员资源不会被接管，名称冲突会失败。账号已被删除时增量加入会忽略该 ID。 +无实际变化时不会写审计或递增配置版本。创建与归属登记、成员变更、授权复验均在同一事务完成。 + +插件升级沿用实例资源；停用保留分组和 Key，其原生启用状态保持不变。删除实例解除归属，资源仍由管理员管理； +重新安装为新实例不会接管旧资源。管理员删除自有资源后，仍启用的维护逻辑可在下次对账重新创建。 +Key 明文仍通过宿主管理面查看，插件模型调用使用返回的 Key ID。 + +典型处理器先确保分组存在，再通过 `data` 分页查询账号并增量补齐成员，最后确保 Key 存在。 +安装 `groups` 域即授权纳入全部当前及未来账号；插件可按自己的配置筛选账号，但该筛选不构成宿主的权限边界。 + ### Key、模型与模型调用 `models` 域开放： diff --git a/backend/crates/gateway-plugin/sdk/docs/manifest.md b/backend/crates/gateway-plugin/sdk/docs/manifest.md index acd0662bf..cdcb1212f 100644 --- a/backend/crates/gateway-plugin/sdk/docs/manifest.md +++ b/backend/crates/gateway-plugin/sdk/docs/manifest.md @@ -49,6 +49,7 @@ capability 固定并由工具生成： | `registration` / `retry` | `model_catalog` / `retry_policy` | | `observation` | `request_lifecycle`、`web_socket_observer`、`usage` | | `management` / `command_line` | `management` / `command_line` | +| `maintenance` | `maintenance` | `middleware` 必须显式选择 `request`、`attempt` 或两者；协议格式等真实业务选择也不能省略。 安装清单若携带不同的固定阶段会被拒绝，而不是在加载时静默改写。 @@ -62,8 +63,10 @@ capability 固定并由工具生成： | `network` | 使用宿主受管网络 | | `models` | 查询非秘密 Key 与模型并调用模型，可能产生消耗 | | `accounts` | 查询、读取原始凭据和修改账号 | -| `data` | 仅在管理与命令阶段只读全部账号的基础信息和已有额度观测 | +| `data` | 仅在管理、命令与维护阶段只读全部账号的基础信息和已有额度观测 | | `requests` | 查看和处理请求、响应、路由、调度与观察事实 | +| `groups` | 创建本实例分组并管理所有当前及未来账号在这些分组中的成员关系 | +| `keys` | 创建绑定本实例分组的 Key，不读取密钥明文 | | `public_endpoints` | 提供无需登录即可访问的资源或回调 | 安装时统一接受清单声明的域，不再填写逐方法、用途、Key、账号或 Provider 白名单。日志与清单声明的 diff --git a/backend/crates/gateway-plugin/sdk/src/call/mod.rs b/backend/crates/gateway-plugin/sdk/src/call/mod.rs index 4ba086e39..31bda6250 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/mod.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/mod.rs @@ -10,3 +10,4 @@ pub mod model; pub mod observation; pub mod policy; pub mod registration; +pub mod resources; diff --git a/backend/crates/gateway-plugin/sdk/src/call/resources.rs b/backend/crates/gateway-plugin/sdk/src/call/resources.rs new file mode 100644 index 000000000..77515c902 --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/src/call/resources.rs @@ -0,0 +1,58 @@ +//! 实例自有资源：稳定 resource_key 用于重试，名称只在首次创建时使用。 + +use serde::{Deserialize, Serialize}; + +pub const GROUP_ENSURE: &str = "host.groups.ensure"; +pub const GROUP_MEMBERS: &str = "host.groups.change_members"; +pub const KEY_ENSURE: &str = "host.keys.ensure"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct GroupEnsureRequest { + pub resource_key: String, + pub name: String, + pub color: String, + #[serde(default)] + pub description: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct GroupMembersChange { + pub resource_key: String, + #[serde(default)] + pub add: Vec, + #[serde(default)] + pub remove: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct KeyEnsureRequest { + pub resource_key: String, + pub name: String, + /// 必须包含至少一个本实例分组，不能创建不受分组约束的 Key。 + pub group_resource_keys: Vec, + #[serde(default)] + pub max_concurrency: u64, + #[serde(default)] + pub requests_per_minute: u64, + pub daily_limit_usd: String, + pub weekly_limit_usd: String, +} + +/// 返回非秘密身份；调用模型时使用 id，密钥明文仍由管理员管理。 +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManagedResource { + pub id: String, + pub name: String, + pub enabled: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct GroupMembersChanged { + pub added: u64, + pub removed: u64, +} diff --git a/backend/crates/gateway-plugin/sdk/src/capability.rs b/backend/crates/gateway-plugin/sdk/src/capability.rs index 9024aaaec..64a9f0c3c 100644 --- a/backend/crates/gateway-plugin/sdk/src/capability.rs +++ b/backend/crates/gateway-plugin/sdk/src/capability.rs @@ -17,6 +17,7 @@ pub enum Capability { Usage, CommandLine, Management, + Maintenance, } impl Capability { @@ -35,6 +36,7 @@ impl Capability { Self::Usage => "usage", Self::CommandLine => "command_line", Self::Management => "management", + Self::Maintenance => "maintenance", } } @@ -51,6 +53,7 @@ impl Capability { Self::RequestLifecycle | Self::WebSocketObserver | Self::Usage => &[Stage::Observation], Self::CommandLine => &[Stage::CommandLine], Self::Management => &[Stage::Management], + Self::Maintenance => &[Stage::Maintenance], } } } @@ -72,6 +75,8 @@ pub enum Stage { CommandLine, /// 公共登录回调不继承插件实例的任何宿主回调权限。 PublicManagement, + /// 宿主针对已发布实例签发的幂等维护调用。 + Maintenance, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -144,17 +149,21 @@ pub enum Permission { Data, Requests, PublicEndpoints, + Groups, + Keys, } impl Permission { /// 当前公开访问域，安装摘要与授权校验复用同一集合。 - pub const ALL: [Self; 6] = [ + pub const ALL: [Self; 8] = [ Self::Network, Self::Models, Self::Accounts, Self::Data, Self::Requests, Self::PublicEndpoints, + Self::Groups, + Self::Keys, ]; #[must_use] @@ -166,6 +175,8 @@ impl Permission { Self::Data => "data", Self::Requests => "requests", Self::PublicEndpoints => "public_endpoints", + Self::Groups => "groups", + Self::Keys => "keys", } } @@ -179,6 +190,8 @@ impl Permission { Self::Data => "基础数据", Self::Requests => "请求处理", Self::PublicEndpoints => "公开入口", + Self::Groups => "专用账号分组", + Self::Keys => "专用 API Key", } } @@ -190,10 +203,14 @@ impl Permission { Self::Models => "查询模型与 API Key 信息并调用模型，可能产生消耗", Self::Accounts => "读取和修改账号，包括访问原始凭据", Self::Data => { - "在管理或命令入口只读查询所有账号的基础信息与已有额度观测，不包含凭据或修改权限" + "在管理、命令或维护入口只读查询所有账号的基础信息与已有额度观测，不包含凭据或修改权限" } Self::Requests => "查看和处理请求、响应、路由及账号选择", Self::PublicEndpoints => "提供无需登录即可访问的资源与回调入口", + Self::Groups => { + "创建本插件的分组，可将所有现有及未来新增账号加入或移出这些分组，不修改其他分组" + } + Self::Keys => "创建绑定本插件分组的 API Key，不读取密钥明文或修改管理员创建的 Key", } } } diff --git a/backend/crates/gateway-plugin/sdk/src/client/data.rs b/backend/crates/gateway-plugin/sdk/src/client/data.rs index fd453f6c4..2f642660e 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/data.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/data.rs @@ -1,11 +1,9 @@ -use serde::{Serialize, de::DeserializeOwned}; +use crate::{PluginFault, call::data}; -use crate::{ErrorCode, PluginFault, call::data}; - -use super::{HostClient, SessionError}; +use super::{HostClient, payload_call}; impl HostClient { - /// 在已授权的 management／command_line 调用中分页读取账号基础事实。 + /// 在已授权的 management／command_line／maintenance 调用中分页读取账号基础事实。 /// /// # Errors /// @@ -14,7 +12,7 @@ impl HostClient { &self, query: data::AccountFactsQuery, ) -> Result { - data_call(self, data::ACCOUNTS_LIST, query).await + payload_call(self, data::ACCOUNTS_LIST, query).await } /// 读取已有额度观测，不触发上游刷新，也不返回宿主预测结果。 @@ -26,26 +24,6 @@ impl HostClient { &self, query: data::QuotaFactsQuery, ) -> Result { - data_call(self, data::QUOTA_GET, query).await + payload_call(self, data::QUOTA_GET, query).await } } - -async fn data_call( - host: &HostClient, - method: &str, - query: T, -) -> Result { - let invalid = || PluginFault::new(ErrorCode::InvalidInput, "invalid data callback payload"); - let reply = host - .call( - method, - serde_json::json!({}), - serde_json::to_vec(&query).map_err(|_| invalid())?, - ) - .await - .map_err(SessionError::into_plugin_fault)?; - if reply.result != serde_json::json!({}) { - return Err(invalid()); - } - serde_json::from_slice(&reply.payload).map_err(|_| invalid()) -} diff --git a/backend/crates/gateway-plugin/sdk/src/client/mod.rs b/backend/crates/gateway-plugin/sdk/src/client/mod.rs index a36906f81..a7bb2cd43 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/mod.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/mod.rs @@ -4,8 +4,12 @@ mod data; mod frame; mod middleware; mod plugin; +mod resources; mod session; +use crate::{ErrorCode, PluginFault}; +use serde::{Serialize, de::DeserializeOwned}; + pub use frame::{read_frame, validate_frame, write_frame}; pub use middleware::{ MiddlewareBody, MiddlewareBodySender, MiddlewareCall, MiddlewareNext, MiddlewarePlugin, @@ -18,3 +22,23 @@ pub use session::{ CallCancellation, CallFuture, CallReply, HostClient, HostReply, PluginCall, PluginHandler, PluginSession, ResponseStream, SessionConfig, SessionError, StreamSender, }; + +async fn payload_call( + host: &HostClient, + method: &str, + query: T, +) -> Result { + let invalid = || PluginFault::new(ErrorCode::InvalidInput, "invalid host callback payload"); + let reply = host + .call( + method, + serde_json::json!({}), + serde_json::to_vec(&query).map_err(|_| invalid())?, + ) + .await + .map_err(SessionError::into_plugin_fault)?; + if reply.result != serde_json::json!({}) { + return Err(invalid()); + } + serde_json::from_slice(&reply.payload).map_err(|_| invalid()) +} diff --git a/backend/crates/gateway-plugin/sdk/src/client/plugin/methods.rs b/backend/crates/gateway-plugin/sdk/src/client/plugin/methods.rs index 94bcd9af6..ced914673 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/plugin/methods.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/plugin/methods.rs @@ -126,3 +126,12 @@ pub const STATE_MIGRATE: Method = Method::new( + "plugin.reconcile", + &[C::Maintenance], + &[S::Maintenance], + decode_metadata_without_payload, + encode_metadata, +); diff --git a/backend/crates/gateway-plugin/sdk/src/client/plugin/validation.rs b/backend/crates/gateway-plugin/sdk/src/client/plugin/validation.rs index 09e1aeb43..e63d8bcb5 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/plugin/validation.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/plugin/validation.rs @@ -16,6 +16,7 @@ pub(super) fn validate( }; for capability in manifest.contributes.keys() { match capability { + C::Maintenance => required(methods::RECONCILE.name)?, C::Middleware => required(crate::call::middleware::HANDLE_METHOD)?, C::ModelRouter => required(methods::ROUTE_MODEL.name)?, C::ModelCatalog => required(methods::MODEL_CATALOG_REGISTER.name)?, diff --git a/backend/crates/gateway-plugin/sdk/src/client/resources.rs b/backend/crates/gateway-plugin/sdk/src/client/resources.rs new file mode 100644 index 000000000..06a9f6409 --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/src/client/resources.rs @@ -0,0 +1,38 @@ +use crate::{PluginFault, call::resources}; + +use super::{HostClient, payload_call}; + +impl HostClient { + /// 幂等创建或读取本实例分组，不覆盖管理员修改的现有属性。 + /// + /// # Errors + /// 权限、实例版本或输入不合法，或者宿主写入失败时返回错误。 + pub async fn ensure_group( + &self, + request: resources::GroupEnsureRequest, + ) -> Result { + payload_call(self, resources::GROUP_ENSURE, request).await + } + + /// 增量修改本实例分组的成员；保留账号在其他分组中的成员关系。 + /// + /// # Errors + /// 权限、实例版本、资源归属或输入不合法，或者宿主写入失败时返回错误。 + pub async fn change_group_members( + &self, + request: resources::GroupMembersChange, + ) -> Result { + payload_call(self, resources::GROUP_MEMBERS, request).await + } + + /// 幂等创建或读取仅绑定本实例分组的 Key；不返回密钥明文。 + /// + /// # Errors + /// 权限、实例版本、资源归属或输入不合法，或者宿主写入失败时返回错误。 + pub async fn ensure_key( + &self, + request: resources::KeyEnsureRequest, + ) -> Result { + payload_call(self, resources::KEY_ENSURE, request).await + } +} diff --git a/backend/crates/gateway-store/src/bundle.rs b/backend/crates/gateway-store/src/bundle.rs index 842ae8ac0..d5915837e 100644 --- a/backend/crates/gateway-store/src/bundle.rs +++ b/backend/crates/gateway-store/src/bundle.rs @@ -179,6 +179,7 @@ async fn connect( }), backup_ports(pool.clone(), &config)?, plugins.clone(), + plugins.clone(), plugins, ); diff --git a/backend/crates/gateway-store/src/postgres/account_groups.rs b/backend/crates/gateway-store/src/postgres/account_groups.rs index da8ed6b17..659c67b6a 100644 --- a/backend/crates/gateway-store/src/postgres/account_groups.rs +++ b/backend/crates/gateway-store/src/postgres/account_groups.rs @@ -228,7 +228,6 @@ impl AccountGroupStore for PgAccountGroupRepository { command: NewAccountGroup, context: &MutationContext, ) -> AdminStoreResult { - validate_group_fields(&command.name, command.description.as_deref())?; let id = command.id.clone(); let audit = mutation_audit( context, @@ -244,19 +243,7 @@ impl AccountGroupStore for PgAccountGroupRepository { let revision = self .mutate(audit, |transaction| { Box::pin(async move { - sqlx::query( - "insert into account_groups - (id, name, description, color, disable_fast, enabled, created_at, updated_at) - values ($1, $2, $3, $4, $5, true, now(), now())", - ) - .bind(command.id.as_str()) - .bind(command.name) - .bind(command.description) - .bind(command.color.as_str()) - .bind(command.disable_fast) - .execute(&mut **transaction) - .await - .map_err(|error| map_group_write_error(error, command.id.as_str()))?; + insert_account_group_in_transaction(transaction, &command).await?; Ok(()) }) }) @@ -669,3 +656,26 @@ fn invalid(message: &str) -> StoreError { fn unavailable(message: &'static str) -> StoreError { postgres_unavailable(message) } + +/// 原生管理与插件自有分组共用字段校验和写入规则。 +pub(crate) async fn insert_account_group_in_transaction( + transaction: &mut Transaction<'_, Postgres>, + command: &NewAccountGroup, +) -> StoreResult<()> { + validate_group_fields(&command.name, command.description.as_deref()) + .map_err(|_| invalid("invalid account group fields"))?; + sqlx::query( + "insert into account_groups + (id, name, description, color, disable_fast, enabled, created_at, updated_at) + values ($1, $2, $3, $4, $5, true, now(), now())", + ) + .bind(command.id.as_str()) + .bind(&command.name) + .bind(&command.description) + .bind(command.color.as_str()) + .bind(command.disable_fast) + .execute(&mut **transaction) + .await + .map_err(|error| map_group_write_error(error, command.id.as_str()))?; + Ok(()) +} diff --git a/backend/crates/gateway-store/src/postgres/plugins/mod.rs b/backend/crates/gateway-store/src/postgres/plugins/mod.rs index f186c747e..245454817 100644 --- a/backend/crates/gateway-store/src/postgres/plugins/mod.rs +++ b/backend/crates/gateway-store/src/postgres/plugins/mod.rs @@ -1,6 +1,7 @@ mod artifacts; mod credentials; mod instances; +mod resources; mod sources; mod state; diff --git a/backend/crates/gateway-store/src/postgres/plugins/resources.rs b/backend/crates/gateway-store/src/postgres/plugins/resources.rs new file mode 100644 index 000000000..e250f0f4f --- /dev/null +++ b/backend/crates/gateway-store/src/postgres/plugins/resources.rs @@ -0,0 +1,320 @@ +//! 插件资源写入在同一事务中验证当前实例、访问域与资源归属。 + +use super::super::{ + account_groups::insert_account_group_in_transaction, + append_admin_audit_event_in_transaction, bump_config_revision_in_transaction, + client_keys::{NewClientApiKey, insert_client_api_key_in_transaction}, +}; +use super::PgPluginStore; +use crate::{admin_revision, admin_store_error, mutation_audit}; +use async_trait::async_trait; +use gateway_admin::{ + model::{ + MutationContext, + account_groups::NewAccountGroup, + client_keys::NewClientKey, + plugin_resources::{ + GroupMembersChange, GroupMembersChanged, ManagedResource, PluginResourceOwner, + ResourceMutation, + }, + }, + ports::{ + plugin_resources::PluginResourceStore, + store::{AdminStoreError, AdminStoreErrorKind, AdminStoreResult}, + }, +}; +use sqlx::{Postgres, Row as _, Transaction}; +use std::collections::BTreeSet; + +fn unavailable() -> AdminStoreError { + AdminStoreError::new( + AdminStoreErrorKind::Unavailable, + "plugin resource", + "plugin resource store is unavailable", + ) +} + +fn denied() -> AdminStoreError { + AdminStoreError::new( + AdminStoreErrorKind::Conflict, + "plugin resource", + "plugin instance or resource authorization changed", + ) +} +fn invalid() -> AdminStoreError { + AdminStoreError::new( + AdminStoreErrorKind::Invalid, + "plugin resource", + "invalid plugin resource input", + ) +} + +fn validate_resource_key(key: &str) -> AdminStoreResult<()> { + if key.is_empty() + || key.len() > 64 + || !key.as_bytes()[0].is_ascii_alphanumeric() + || !key + .bytes() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || b"_.-".contains(&c)) + { + return Err(invalid()); + } + Ok(()) +} + +impl PgPluginStore { + async fn resource_transaction( + &self, + owner: &PluginResourceOwner, + permission: &str, + ) -> AdminStoreResult> { + let mut tx = self.pool.begin().await.map_err(|_| unavailable())?; + // 与全部管理写入保持相同锁顺序；无变化的对账只锁定，不递增 revision。 + sqlx::query("select config_revision from runtime_settings where id=1 for update") + .execute(&mut *tx) + .await + .map_err(|_| unavailable())?; + let id = uuid::Uuid::parse_str(&owner.instance_id).map_err(|_| denied())?; + let revision = i64::try_from(owner.revision.get()).map_err(|_| denied())?; + let allowed: bool = sqlx::query_scalar( + "select exists( + select 1 from plugin_instances i + join plugin_artifacts a on a.sha256=i.artifact_sha256 + where i.id=$1 and i.enabled and i.revision=$2 and i.artifact_sha256=$3 + and a.accepted_at is not null and a.metadata_json->'requestedPermissions' ? $4 + )", + ) + .bind(id) + .bind(revision) + .bind(&owner.artifact_sha256) + .bind(permission) + .fetch_one(&mut *tx) + .await + .map_err(|_| unavailable())?; + if !allowed { + return Err(denied()); + } + Ok(tx) + } +} + +async fn finish( + mut tx: Transaction<'_, Postgres>, + context: &MutationContext, + kind: &str, + id: &str, + changed: bool, + value: T, +) -> AdminStoreResult> { + let revision = if changed { + let revision = bump_config_revision_in_transaction(&mut tx) + .await + .map_err(|e| admin_store_error("plugin resource", e))?; + let audit = mutation_audit( + context, + "plugin_reconcile", + kind, + id, + vec!["plugin_owned_resource".to_owned()], + ); + append_admin_audit_event_in_transaction(&mut tx, audit, revision) + .await + .map_err(|e| admin_store_error("plugin resource", e))?; + Some(admin_revision(revision)?) + } else { + None + }; + tx.commit().await.map_err(|_| unavailable())?; + Ok(ResourceMutation { revision, value }) +} + +fn resource(row: &sqlx::postgres::PgRow) -> AdminStoreResult { + Ok(ManagedResource { + id: row.try_get("id").map_err(|_| unavailable())?, + name: row.try_get("name").map_err(|_| unavailable())?, + enabled: row.try_get("enabled").map_err(|_| unavailable())?, + }) +} + +#[async_trait] +impl PluginResourceStore for PgPluginStore { + async fn ensure_group( + &self, + owner: &PluginResourceOwner, + resource_key: String, + command: NewAccountGroup, + context: &MutationContext, + ) -> AdminStoreResult> { + validate_resource_key(&resource_key)?; + let mut tx = self.resource_transaction(owner, "groups").await?; + let instance = uuid::Uuid::parse_str(&owner.instance_id).map_err(|_| denied())?; + if let Some(row) = sqlx::query( + "select g.id,g.name,g.enabled from plugin_group_resources r + join account_groups g on g.id=r.group_id + where r.instance_id=$1 and r.resource_key=$2", + ) + .bind(instance) + .bind(&resource_key) + .fetch_optional(&mut *tx) + .await + .map_err(|_| unavailable())? + { + return finish(tx, context, "account_group", "", false, resource(&row)?).await; + } + insert_account_group_in_transaction(&mut tx, &command) + .await + .map_err(|e| admin_store_error("account group", e))?; + sqlx::query("insert into plugin_group_resources(instance_id,resource_key,group_id) values($1,$2,$3)") + .bind(instance) + .bind(resource_key) + .bind(command.id.as_str()) + .execute(&mut *tx) + .await.map_err(|_| unavailable())?; + let value = ManagedResource { + id: command.id.as_str().to_owned(), + name: command.name, + enabled: true, + }; + finish( + tx, + context, + "account_group", + command.id.as_str(), + true, + value, + ) + .await + } + + async fn ensure_key( + &self, + owner: &PluginResourceOwner, + resource_key: String, + groups: Vec, + command: NewClientKey, + context: &MutationContext, + ) -> AdminStoreResult> { + validate_resource_key(&resource_key)?; + if groups.is_empty() + || groups.len() > 64 + || groups.iter().collect::>().len() != groups.len() + { + return Err(invalid()); + } + for group in &groups { + validate_resource_key(group)?; + } + let mut tx = self.resource_transaction(owner, "keys").await?; + let instance = uuid::Uuid::parse_str(&owner.instance_id).map_err(|_| denied())?; + if let Some(row) = sqlx::query( + "select k.id,k.name,k.enabled from plugin_key_resources r + join client_api_keys k on k.id=r.key_id + where r.instance_id=$1 and r.resource_key=$2", + ) + .bind(instance) + .bind(&resource_key) + .fetch_optional(&mut *tx) + .await + .map_err(|_| unavailable())? + { + return finish(tx, context, "client_api_key", "", false, resource(&row)?).await; + } + let group_ids: Vec = sqlx::query_scalar("select group_id from plugin_group_resources where instance_id=$1 and resource_key=any($2::text[])") + .bind(instance) + .bind(&groups) + .fetch_all(&mut *tx) + .await.map_err(|_| unavailable())?; + if group_ids.len() != groups.len() { + return Err(denied()); + } + let key = NewClientApiKey { + request_profile_overrides: command.request_profile_overrides, + id: command.id.as_str().to_owned(), + name: command.name, + label: command.label, + key: command.plaintext, + group_ids, + budget: command.budget, + max_concurrency: command.limits.max_concurrency, + requests_per_minute: command.limits.requests_per_minute, + }; + insert_client_api_key_in_transaction(&mut tx, &key) + .await + .map_err(|e| admin_store_error("client API key", e))?; + sqlx::query( + "insert into plugin_key_resources(instance_id,resource_key,key_id) values($1,$2,$3)", + ) + .bind(instance) + .bind(resource_key) + .bind(&key.id) + .execute(&mut *tx) + .await + .map_err(|_| unavailable())?; + let value = ManagedResource { + id: key.id.clone(), + name: key.name.trim().to_owned(), + enabled: true, + }; + finish(tx, context, "client_api_key", &key.id, true, value).await + } + + async fn change_members( + &self, + owner: &PluginResourceOwner, + command: GroupMembersChange, + context: &MutationContext, + ) -> AdminStoreResult> { + validate_resource_key(&command.resource_key)?; + let add: BTreeSet<_> = command.add.iter().collect(); + let remove: BTreeSet<_> = command.remove.iter().collect(); + if command.add.len() + command.remove.len() > 200 + || !add.is_disjoint(&remove) + || add.len() != command.add.len() + || remove.len() != command.remove.len() + { + return Err(invalid()); + } + let mut tx = self.resource_transaction(owner, "groups").await?; + let instance = uuid::Uuid::parse_str(&owner.instance_id).map_err(|_| denied())?; + let group: String = sqlx::query_scalar( + "select group_id from plugin_group_resources where instance_id=$1 and resource_key=$2", + ) + .bind(instance) + .bind(command.resource_key) + .fetch_optional(&mut *tx) + .await + .map_err(|_| unavailable())? + .ok_or_else(denied)?; + // 已删除账号自然不再需要加入；外键负责与并发账号删除保持一致。 + let added = sqlx::query( + "insert into account_group_accounts(account_group_id,provider_account_id,created_at) + select $1,id,now() from provider_accounts where id=any($2::text[]) + on conflict do nothing", + ) + .bind(&group) + .bind(command.add) + .execute(&mut *tx) + .await + .map_err(|_| unavailable())? + .rows_affected(); + let removed = sqlx::query( + "delete from account_group_accounts + where account_group_id=$1 and provider_account_id=any($2::text[])", + ) + .bind(&group) + .bind(command.remove) + .execute(&mut *tx) + .await + .map_err(|_| unavailable())? + .rows_affected(); + finish( + tx, + context, + "account_group", + &group, + added + removed > 0, + GroupMembersChanged { added, removed }, + ) + .await + } +} diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index ac9fcd2ff..c43aca1d8 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -247,8 +247,10 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly "plugin_artifact_credentials", "plugin_artifact_platforms", "plugin_artifacts", + "plugin_group_resources", "plugin_instance_secrets", "plugin_instances", + "plugin_key_resources", "plugin_source_credentials", "plugin_state_generations", "plugin_state_records", diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 7e4329af4..e2ffe16dc 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -16,3 +16,4 @@ b20e425c0b062b72d35fdebe294e023eb17ceb44ac93083ff3d48400715bf430 0012_upstream_ a1e23b8535ed41de27861833cd2d5ee7498a9bb0e523156ba3f011f9c6b3966b 0016_client_profiles_and_model_pricing.sql fcfae912ddefd54551ebf8e7e733442519d2862d2977134d548ebc1e75c1dc34 0017_unlimited_default_account_concurrency.sql 229c70296764b94fe69c3f19889e566fa290c6c44c64970efe96d7fd2a0008d5 0018_plugins.sql +4daf61846ce59c449865523eb2f7439627dbb65076e56f20de8a74d50905bbf0 0019_plugin_resources.sql diff --git a/backend/migrations/0019_plugin_resources.sql b/backend/migrations/0019_plugin_resources.sql new file mode 100644 index 000000000..12f9b92d8 --- /dev/null +++ b/backend/migrations/0019_plugin_resources.sql @@ -0,0 +1,14 @@ +-- 归属与稳定资源键属于宿主；升级沿用实例，删除实例只解除归属，不删除业务资源。 +create table plugin_group_resources ( + instance_id uuid not null references plugin_instances(id) on delete cascade, + resource_key text not null check (resource_key ~ '^[a-z0-9][a-z0-9_.-]{0,63}$'), + group_id text not null unique references account_groups(id) on delete cascade, + primary key (instance_id, resource_key) +); + +create table plugin_key_resources ( + instance_id uuid not null references plugin_instances(id) on delete cascade, + resource_key text not null check (resource_key ~ '^[a-z0-9][a-z0-9_.-]{0,63}$'), + key_id text not null unique references client_api_keys(id) on delete cascade, + primary key (instance_id, resource_key) +); diff --git a/docs/architecture.md b/docs/architecture.md index 2fe4cb56a..341c247e6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -139,11 +139,16 @@ flowchart LR 正文按需读取，透传无需额外读取权限；转换和改写不能绕过 Core 的交付、取消与结算边界。 数据面插件通过受管 HTTP 已发送或无法证明未发送时，Runtime 把该事实并入 Core 的请求副作用水位，后续不能按 Provider 的 `not_sent` 结果透明重放。 -- **访问域不跨调用。** `network`、`models`、`accounts`、`data`、`requests`、`public_endpoints` 只开放对应资源域； +- **访问域不跨调用。** `network`、`models`、`accounts`、`data`、`requests`、`public_endpoints`、`groups`、`keys` 只开放对应资源域； 账号、HTTP、模型、私有状态和日志回调仍绑定有效父调用与阶段。管理页和 CLI 不预绑定 Client Key，模型调用按次 选择当前 Key；页面 Responses 桥还复核精确实例目标和 `models` 域，并在等待与交付期间持续撤销检查。 `frontend_authentication` 必须显式配置 principal 到 Key 的映射；公开登录回调使用一次性票据且不继承宿主回调权限。 - `data` 仅向管理和命令阶段提供账号基础投影与已有额度观测，不提供凭据、预测或 SQL，不进入客户端请求链。 + `data` 仅向管理、命令和维护阶段提供账号基础投影与已有额度观测，不提供凭据、预测或 SQL，不进入客户端请求链。 +- **维护只作用于已发布实例。** `maintenance` 由 Host 监督的 Runtime worker 调用，启用、恢复、配置发布和周期补偿共用幂等对账入口。 + 发布通知只用于唤醒，事实始终来自当前快照与数据库；同实例串行，通知合并，停用与版本替换取消旧任务。 + 自有分组和 Key 按实例、资源类型与稳定资源键登记；Admin 组织用例，Store 在同一事务复核当前启用 revision、 + 制品授权与归属，并复用原生资源写入、审计和配置发布。无变化不递增 revision；插件不能接管管理员资源， + 成员变更不覆盖账号的其他分组。停用保留资源；删除实例只解除归属。 - **版本切换保持数据一致。** 同一插件最多启用一个实例；每个实例与制品摘要保存最近启用时提交的配置、密钥和绑定， 快照与实例修改共享事务，停用草稿不覆盖恢复点。版本切换优先恢复目标快照，否则合并缺失默认值并保留显式设置。 私有状态按实例、schema 与配置版本隔离，写入使用精确记录版本。 diff --git a/docs/plugins.md b/docs/plugins.md index d0d305388..7262a778e 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -34,7 +34,7 @@ flowchart LR 确认安装后，若该插件还没有任何配置，宿主会： 1. 从 schema 填入普通配置默认值，不把敏感字段的默认值写入普通配置 -2. 为需要实例范围的贡献项生成默认绑定；管理页面和命令行按声明注册，入口认证不自动绑定 +2. 为需要实例范围的贡献项生成默认绑定；管理页面、命令行和自动维护按声明注册，入口认证不自动绑定 3. 配置完整时直接启用并发布；仅缺少必填项时保存为停用并打开配置页 类型错误、非法 schema 或把敏感值混入普通配置会直接拒绝，不会伪装成“待配置”。更新已有插件时不会创建另一套配置， @@ -49,9 +49,11 @@ flowchart LR | `network` | 访问网络 | | `models` | 查询模型和 Client Key 基本信息并调用模型，可能产生消耗 | | `accounts` | 读取和修改账号，包括访问原始凭据 | -| `data` | 仅在管理或命令入口只读全部账号的基础信息和已有额度观测，不含凭据或预测 | +| `data` | 仅在管理、命令或维护入口只读全部账号的基础信息和已有额度观测，不含凭据或预测 | | `requests` | 查看和处理请求、响应、路由及账号选择 | | `public_endpoints` | 提供无需登录即可访问的资源与回调入口 | +| `groups` | 创建插件自有分组，可将所有现有及未来账号加入或移出这些分组 | +| `keys` | 创建绑定插件自有分组的 Client Key，不读取密钥明文 | 访问域不是操作列表或资源白名单。实例配置不能扩大、缩小或伪造制品声明；切换版本后使用目标制品已经接受的精确声明。 插件进程与网关使用相同的系统身份，接受访问域不是操作系统沙箱，因此只安装可信来源的包。 @@ -73,6 +75,9 @@ flowchart LR 也不会自动删除历史配置、密钥或私有数据。升级前已经启用多套的插件会提示整理，宿主不会擅自选择停用对象，后续启用或更新 必须收敛到一套配置。 +声明自动维护的插件可自行准备已授权的分组和 Key，无需提供管理员密码。停用会停止维护，但保留这些资源及其原生启用状态； +删除插件配置也不自动删除资源。需要撤销 Key 访问时，在 Key 管理中停用或删除相应 Key。 + “已启用”表示配置已经发布，不是外部服务的健康证明。“等待生效”表示候选仍在准备；“需要处理”会显示发布或运行失败原因。 并发编辑冲突时应重新加载当前 revision 后确认，不能覆盖另一窗口的修改。 @@ -81,6 +86,7 @@ flowchart LR | 功能 | 使用方式 | | --- | --- | | 管理页面 | 从插件详情或左侧插件导航打开；页面在隔离 iframe 中运行 | +| 自动维护 | 启用、重启及账号等配置变化后自动对账；失败重试，周期补漏 | | 页面模型请求 | 每次选择 Client Key，经固定实例版本的 Responses 桥执行，返回原生 JSON 或 SSE | | 请求中间件、模型路由、账号调度、重试策略、请求观察 | 匹配功能范围的正常网关请求自动使用；重试策略故障时交给后续处理 | | 模型目录 | 启用后原子发布别名，列表与路由继承目标模型的能力和现有访问范围 |