refactor(plugins): separate client identity from model execution ports (#527)

Co-authored-by: Zyy <zyycn@users.noreply.github.com> Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-10-26 13:21:31 +08:00
parent 614c81c375
commit e38cf31323
16 changed files with 337 additions and 194 deletions
@@ -103,6 +103,10 @@ async fn execute(
    let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control());
    or_shutdown!(plugin_runtime, management.bind(&plugin_runtime));
    let core = or_shutdown!(plugin_runtime, core.activate().await);
+    or_shutdown!(
+        plugin_runtime,
+        plugin_runtime.bind_client_identity_port(&core.client_identity_port())
+    );
    let nested_models = core.nested_model_execution_port();
    let affinity_lookup = core.affinity_lookup_port();
    or_shutdown!(
@@ -38,6 +38,10 @@ pub async fn run() -> Result<(), BootstrapError> {
    let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control());
    or_shutdown!(plugin_runtime, management.bind(&plugin_runtime));
    let mut core = or_shutdown!(plugin_runtime, core.activate().await);
+    or_shutdown!(
+        plugin_runtime,
+        plugin_runtime.bind_client_identity_port(&core.client_identity_port())
+    );
    let nested_models = core.nested_model_execution_port();
    let affinity_lookup = core.affinity_lookup_port();
    or_shutdown!(
@@ -9,10 +9,32 @@ use std::{
 use futures::future::BoxFuture;

 use crate::{
+    error::GatewayError,
    policy::{ClientApiKeyId, PlaintextClientApiKey},
-    routing::extensions::{ExtensionSetId, ExtensionSetReference},
+    routing::{
+        extensions::{ExtensionSetId, ExtensionSetReference},
+        request_settings::RequestSettings,
+    },
 };

+use super::execution::AuthenticatedClient;
+
+/// 身份解析与认证只返回冻结权限，不提供模型执行入口
+pub trait ClientIdentityPort: Send + Sync {
+    /// 只解析身份与冻结权限，不建立模型执行图或记录使用事实
+    fn resolve_client(
+        &self,
+        client_key_id: ClientApiKeyId,
+        settings: Option<RequestSettings>,
+    ) -> BoxFuture<'_, Result<AuthenticatedClient, GatewayError>>;
+
+    /// 复用入口认证计划及 principal 映射，不进行模型准入
+    fn verify_client(
+        &self,
+        request: ClientAuthenticationRequest,
+    ) -> BoxFuture<'_, Result<AuthenticatedClient, GatewayError>>;
+}
+
 pub const MAXIMUM_AUTHORIZATION_BYTES: usize = 8 * 1024;

 /// API 从数据面 Authorization 头构造的有界认证信封
@@ -27,7 +27,7 @@ use crate::{
            ClientAdmissionRequest,
        },
        authentication::{
-            ClientAuthenticationRequest, FrontendAuthenticationDecision,
+            ClientAuthenticationRequest, ClientIdentityPort, FrontendAuthenticationDecision,
            FrontendAuthenticationExtensionIndex,
        },
        budget::{ClientBudgetCharge, ClientBudgetPort},
@@ -1705,7 +1705,7 @@ impl ExecutionService for DefaultExecutionService {
    }
 }

-impl NestedModelExecutionPort for DefaultExecutionService {
+impl ClientIdentityPort for DefaultExecutionService {
    fn resolve_client(
        &self,
        client_key_id: ClientApiKeyId,
@@ -1737,7 +1737,9 @@ impl NestedModelExecutionPort for DefaultExecutionService {
            }
        })
    }
+}

+impl NestedModelExecutionPort for DefaultExecutionService {
    fn models(
        &self,
        client: AuthenticatedClient,
@@ -13,8 +13,8 @@ use futures::future::BoxFuture;

 use crate::{
    account::ProviderAccountId, error::GatewayError, identity::ProviderKind,
-    lifecycle::CancellationToken, operation::Operation, policy::ClientApiKeyId,
-    provider_ports::ProviderSessionAffinityKey, routing::PublicModelId,
+    lifecycle::CancellationToken, operation::Operation, provider_ports::ProviderSessionAffinityKey,
+    routing::PublicModelId,
 };

 use super::{
@@ -111,19 +111,6 @@ impl fmt::Debug for NestedModelExecutionRequest {

 /// 嵌套模型执行只进入 Core 的既有执行会话；Runtime 不实现重试、commit 或计费
 pub trait NestedModelExecutionPort: Send + Sync {
-    /// 只解析身份与冻结权限，不建立模型执行图或记录使用事实
-    fn resolve_client(
-        &self,
-        client_key_id: ClientApiKeyId,
-        settings: Option<crate::routing::request_settings::RequestSettings>,
-    ) -> BoxFuture<'_, Result<AuthenticatedClient, GatewayError>>;
-
-    /// 复用入口认证计划及 principal 映射，不进行模型准入
-    fn verify_client(
-        &self,
-        request: super::authentication::ClientAuthenticationRequest,
-    ) -> BoxFuture<'_, Result<AuthenticatedClient, GatewayError>>;
-
    fn models(
        &self,
        client: AuthenticatedClient,
@@ -34,6 +34,7 @@ use engine::ExecutionStore;
 use engine::admission::{
    ClientAdmissionPort, ClientAdmissionRecoveryPort, restore_client_admission_startup,
 };
+use engine::authentication::ClientIdentityPort;
 use engine::continuation::NativeContinuationPort;
 use engine::execution::{
    ClientApiKeyUsageSink, ClientKeyVerifier, DefaultExecutionService, ExecutionService,
@@ -127,6 +128,7 @@ pub struct CoreBundle {
    client_key_verifier: Arc<dyn ClientKeyVerifier>,
    snapshot_control: Arc<dyn SnapshotControl>,
    account_probe: Arc<dyn AccountProbe>,
+    client_identity: Arc<dyn ClientIdentityPort>,
    nested_model_execution: Arc<dyn NestedModelExecutionPort>,
    affinity_lookup: Arc<dyn AffinityLookupPort>,
    health_probes: Vec<Arc<dyn HealthProbe>>,
@@ -181,6 +183,7 @@ impl CoreStartup {
        let execution: Arc<dyn ExecutionService> = service.clone();
        let client_key_verifier: Arc<dyn ClientKeyVerifier> = service.clone();
        let account_probe: Arc<dyn AccountProbe> = service.clone();
+        let client_identity: Arc<dyn ClientIdentityPort> = service.clone();
        let nested_model_execution: Arc<dyn NestedModelExecutionPort> = service.clone();
        let affinity_lookup: Arc<dyn AffinityLookupPort> = service;
        let snapshot_control: Arc<dyn SnapshotControl> = publisher;
@@ -191,6 +194,7 @@ impl CoreStartup {
            client_key_verifier,
            snapshot_control,
            account_probe,
+            client_identity,
            nested_model_execution,
            affinity_lookup,
            health_probes,
@@ -266,6 +270,11 @@ impl CoreBundle {
        Arc::clone(&self.account_probe)
    }

+    #[must_use]
+    pub fn client_identity_port(&self) -> Arc<dyn ClientIdentityPort> {
+        Arc::clone(&self.client_identity)
+    }
+
    #[must_use]
    pub fn nested_model_execution_port(&self) -> Arc<dyn NestedModelExecutionPort> {
        Arc::clone(&self.nested_model_execution)
@@ -395,6 +404,7 @@ pub struct CoreCommandPlaneStartup {
 pub struct CoreCommandPlaneBundle {
    control: CoreControlPlaneBundle,
    account_probe: Arc<dyn AccountProbe>,
+    client_identity: Arc<dyn ClientIdentityPort>,
    nested_model_execution: Arc<dyn NestedModelExecutionPort>,
    affinity_lookup: Arc<dyn AffinityLookupPort>,
 }
@@ -425,11 +435,13 @@ impl CoreCommandPlaneStartup {
            .with_snapshot_refresh(Arc::clone(&control.publisher)),
        );
        let account_probe: Arc<dyn AccountProbe> = service.clone();
+        let client_identity: Arc<dyn ClientIdentityPort> = service.clone();
        let nested_model_execution: Arc<dyn NestedModelExecutionPort> = service.clone();
        let affinity_lookup: Arc<dyn AffinityLookupPort> = service;
        Ok(CoreCommandPlaneBundle {
            control,
            account_probe,
+            client_identity,
            nested_model_execution,
            affinity_lookup,
        })
@@ -452,6 +464,11 @@ impl CoreCommandPlaneBundle {
        self.control.snapshot_control()
    }

+    #[must_use]
+    pub fn client_identity_port(&self) -> Arc<dyn ClientIdentityPort> {
+        Arc::clone(&self.client_identity)
+    }
+
    #[must_use]
    pub fn nested_model_execution_port(&self) -> Arc<dyn NestedModelExecutionPort> {
        Arc::clone(&self.nested_model_execution)
@@ -28,7 +28,7 @@ fn identity_queries_keep_frozen_permissions_without_using_execution_resources()
        let settings = service.request_settings().unwrap();
        let key = ClientApiKeyId::new("key_start_test").unwrap();
        let client =
-            NestedModelExecutionPort::resolve_client(&service, key.clone(), Some(settings.clone()))
+            ClientIdentityPort::resolve_client(&service, key.clone(), Some(settings.clone()))
                .await
                .unwrap();
        snapshots.publish(start_snapshot_with_policy(
@@ -59,12 +59,11 @@ fn identity_queries_keep_frozen_permissions_without_using_execution_resources()
        .await
        .unwrap();
        assert_eq!(models, vec![PublicModelId::new("gpt-start").unwrap()]);
-        let frozen =
-            NestedModelExecutionPort::resolve_client(&service, key.clone(), Some(settings))
-                .await
-                .unwrap();
+        let frozen = ClientIdentityPort::resolve_client(&service, key.clone(), Some(settings))
+            .await
+            .unwrap();
        assert_eq!(frozen.snapshot().revision().get(), 1);
-        let current = NestedModelExecutionPort::resolve_client(&service, key, None).await;
+        let current = ClientIdentityPort::resolve_client(&service, key, None).await;
        assert!(matches!(current, Err(error) if error.kind() == GatewayErrorKind::Unauthorized));

        let cancellation = CancellationToken::default();
@@ -108,7 +107,7 @@ fn execution_scope_limits_apply_only_when_starting_a_model() {
                scope.extending(format!("ancestor-{index}")).unwrap()
            });
        for extension_scope in [repeated, exhausted] {
-            let client = NestedModelExecutionPort::resolve_client(
+            let client = ClientIdentityPort::resolve_client(
                &service,
                ClientApiKeyId::new("key_start_test").unwrap(),
                None,
@@ -234,7 +233,7 @@ fn verification_keeps_parent_origin_and_requires_the_frontend_principal_mapping(
            let mut cancelled_origin = origin.clone();
            cancelled_origin.cancellation = CancellationToken::default();
            cancelled_origin.cancellation.cancel();
-            let cancelled = NestedModelExecutionPort::verify_client(
+            let cancelled = ClientIdentityPort::verify_client(
                &service,
                ClientAuthenticationRequest::new("External fixture")
                    .unwrap()
@@ -244,7 +243,7 @@ fn verification_keeps_parent_origin_and_requires_the_frontend_principal_mapping(
            .await;
            assert!(matches!(cancelled, Err(error) if error.kind() == GatewayErrorKind::Cancelled));
            assert_eq!(plan.calls.load(Ordering::SeqCst), 0);
-            let result = NestedModelExecutionPort::verify_client(
+            let result = ClientIdentityPort::verify_client(
                &service,
                ClientAuthenticationRequest::new("External fixture")
                    .unwrap()
@@ -13,6 +13,7 @@ use std::{
    time::{Duration, Instant, SystemTime},
 };

+use gateway_core::engine::authentication::ClientIdentityPort;
 use gateway_core::engine::budget::{ClientBudgetCharge, ClientBudgetError, ClientBudgetPort};
 use gateway_core::error::GatewayError;

@@ -1233,7 +1234,7 @@ fn explicit_key_model_execution_preserves_http_origin_and_leaves_background_orig
                Arc::new(crate::RecordingDiagnostics::default()),
            );
            let input = request(&service, ClientTransport::InternalPlugin);
-            let client = NestedModelExecutionPort::resolve_client(
+            let client = ClientIdentityPort::resolve_client(
                &service,
                ClientApiKeyId::new("key_start_test").unwrap(),
                None,
@@ -0,0 +1,154 @@
+//! 客户端认证、冻结权限查询与调用内身份复用
+
+use std::sync::{Arc, OnceLock, Weak};
+
+use gateway_admin::model::AdminError;
+use gateway_core::{
+    account::ProviderAccountId,
+    engine::{
+        authentication::{ClientAuthenticationRequest, ClientIdentityPort},
+        execution::AuthenticatedClient,
+    },
+    policy::ClientApiKeyId,
+    routing::UpstreamModelId,
+};
+use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault, call::data};
+
+use super::{CallResources, denied, invalid};
+use crate::RpcReply;
+
+const MAX_CLIENT_IDENTITIES_PER_CALL: usize = 16;
+
+pub(crate) struct PluginClientIdentityPortSlot {
+    port: OnceLock<Weak<dyn ClientIdentityPort>>,
+}
+
+impl PluginClientIdentityPortSlot {
+    pub(crate) const fn new() -> Self {
+        Self {
+            port: OnceLock::new(),
+        }
+    }
+
+    pub(crate) fn bind(&self, port: &Arc<dyn ClientIdentityPort>) -> Result<(), AdminError> {
+        self.port
+            .set(Arc::downgrade(port))
+            .map_err(|_| AdminError::conflict("插件身份端口已经绑定"))
+    }
+
+    fn upgrade(&self) -> Result<Arc<dyn ClientIdentityPort>, PluginFault> {
+        self.port.get().and_then(Weak::upgrade).ok_or_else(denied)
+    }
+}
+
+pub(super) struct PluginClients {
+    slot: Arc<PluginClientIdentityPortSlot>,
+}
+
+impl PluginClients {
+    pub(super) fn new(slot: Arc<PluginClientIdentityPortSlot>) -> Self {
+        Self { slot }
+    }
+
+    pub(super) async fn call(
+        &self,
+        context: &CallContext,
+        call: &Arc<CallResources>,
+        method: &str,
+        params: serde_json::Value,
+        payload: Vec<u8>,
+    ) -> Result<RpcReply, PluginFault> {
+        match method {
+            data::CLIENT_VERIFY => {
+                if params != serde_json::json!({}) {
+                    return Err(invalid());
+                }
+                let request: data::VerifyClientRequest =
+                    serde_json::from_slice(&payload).map_err(|_| invalid())?;
+                let mut authentication = ClientAuthenticationRequest::new(request.authorization)
+                    .map_err(|_| invalid())?
+                    .with_invocation(call.child_origin(context)?);
+                if let Some(settings) = call.request_settings() {
+                    authentication = authentication.with_settings(settings);
+                }
+                let port = self.slot.upgrade()?;
+                call.observe_dispatch()?;
+                let client = port
+                    .verify_client(authentication)
+                    .await
+                    .map_err(super::error::gateway)?;
+                super::admin::encode(&data::VerifiedClient {
+                    client_key_id: client.policy().key_id().as_str().to_owned(),
+                })
+            }
+            data::ACCESS_CHECK => {
+                if params != serde_json::json!({}) {
+                    return Err(invalid());
+                }
+                let request: data::AccessQuery =
+                    serde_json::from_slice(&payload).map_err(|_| invalid())?;
+                let client = self.identity(call, request.client_key_id).await?;
+                let scope = client.policy().account_scope();
+                let model = request
+                    .upstream_model
+                    .map(UpstreamModelId::new)
+                    .transpose()
+                    .map_err(|_| invalid())?;
+                let accounts = request
+                    .account_ids
+                    .into_iter()
+                    .map(|account_id| {
+                        let account =
+                            ProviderAccountId::new(account_id.clone()).map_err(|_| invalid())?;
+                        let allowed = model.as_ref().map_or_else(
+                            || scope.allows(&account),
+                            |model| scope.allows_model(&account, model.as_str()),
+                        );
+                        Ok(data::AccountAccessFacts {
+                            account_id,
+                            allowed,
+                        })
+                    })
+                    .collect::<Result<Vec<_>, PluginFault>>()?;
+                super::admin::encode(&data::AccessFacts {
+                    schema_version: 1,
+                    client_key_id: client.policy().key_id().as_str().to_owned(),
+                    accounts,
+                })
+            }
+            _ => Err(denied()),
+        }
+    }
+
+    pub(super) async fn identity(
+        &self,
+        call: &CallResources,
+        client_key_id: String,
+    ) -> Result<AuthenticatedClient, PluginFault> {
+        let port = self.slot.upgrade()?;
+        let key = ClientApiKeyId::new(client_key_id.clone()).map_err(|_| invalid())?;
+        let settings = call.request_settings();
+        let mut identities = call.client_identities.lock().await;
+        if let Some(client) = identities
+            .get(&client_key_id)
+            .filter(|client| client.request_settings() == settings.as_ref())
+        {
+            return Ok(client.clone());
+        }
+        if !identities.contains_key(&client_key_id)
+            && identities.len() >= MAX_CLIENT_IDENTITIES_PER_CALL
+        {
+            return Err(PluginFault::new(
+                ErrorCode::Capacity,
+                "client identity capacity is exhausted",
+            ));
+        }
+        let client = port
+            .resolve_client(key, settings)
+            .await
+            .map_err(super::error::gateway)?;
+        // 查询与执行共用调用内快照，执行链和子执行取消只在真正启动时建立
+        identities.insert(client_key_id, client.clone());
+        Ok(client)
+    }
+}
@@ -4,6 +4,7 @@ mod accounts;
 mod admin;
 mod affinity;
 mod call_resources;
+mod client;
 mod data;
 pub(crate) mod error;
 mod facts;
@@ -38,6 +39,7 @@ use crate::{CallbackHandler, RpcReply};
 pub(crate) use accounts::PluginAccountPortSlot;
 pub(crate) use affinity::PluginAffinityPortSlot;
 use call_resources::{CallResources, HttpStream};
+pub(crate) use client::PluginClientIdentityPortSlot;
 pub(crate) use keys::PluginClientKeyPortSlot;
 pub(crate) use middleware::{
    MiddlewareBinding, MiddlewareBodyAuthority, MiddlewareCallback, MiddlewareCompletionBody,
@@ -48,37 +50,14 @@ pub(crate) use resources::PluginResourcePorts;
 pub(crate) use scope::{CallbackScope, InvocationContext};

 pub(crate) struct PluginCallbackPorts {
-    services: Arc<services::ServicePorts>,
-    http: Arc<HttpClient>,
-    network: NetworkPolicy,
-    accounts: Arc<PluginAccountPortSlot>,
-    keys: Arc<PluginClientKeyPortSlot>,
-    models: Arc<PluginModelPortSlot>,
-    affinity: Arc<PluginAffinityPortSlot>,
-    resources: Arc<PluginResourcePorts>,
-}
-
-impl PluginCallbackPorts {
-    pub(crate) fn new(
-        services: Arc<services::ServicePorts>,
-        http: Arc<HttpClient>,
-        accounts: Arc<PluginAccountPortSlot>,
-        keys: Arc<PluginClientKeyPortSlot>,
-        models: Arc<PluginModelPortSlot>,
-        affinity: Arc<PluginAffinityPortSlot>,
-        resources: Arc<PluginResourcePorts>,
-    ) -> Self {
-        Self {
-            services,
-            http,
-            network: NetworkPolicy::unrestricted(),
-            accounts,
-            keys,
-            models,
-            affinity,
-            resources,
-        }
-    }
+    pub(crate) services: Arc<services::ServicePorts>,
+    pub(crate) http: Arc<HttpClient>,
+    pub(crate) accounts: Arc<PluginAccountPortSlot>,
+    pub(crate) keys: Arc<PluginClientKeyPortSlot>,
+    pub(crate) clients: Arc<PluginClientIdentityPortSlot>,
+    pub(crate) models: Arc<PluginModelPortSlot>,
+    pub(crate) affinity: Arc<PluginAffinityPortSlot>,
+    pub(crate) resources: Arc<PluginResourcePorts>,
 }

 pub(crate) struct PluginCallbacks {
@@ -91,6 +70,7 @@ pub(crate) struct PluginCallbacks {
    keys: Arc<keys::PluginClientKeys>,
    affinity: Arc<affinity::PluginAffinity>,
    log: Arc<log::PluginLog>,
+    clients: Arc<client::PluginClients>,
    models: Arc<model::PluginModels>,
    private_state: Arc<private_state::PluginPrivateState>,
    http: Arc<HttpClient>,
@@ -112,6 +92,7 @@ impl PluginCallbacks {
        ports: PluginCallbackPorts,
        request_limits: Arc<request_limits::RequestLimits>,
    ) -> Result<Self, AdminError> {
+        let clients = Arc::new(client::PluginClients::new(ports.clients));
        Ok(Self {
            registration_body_limit: None,
            services: ports.services,
@@ -132,14 +113,16 @@ impl PluginCallbacks {
                log::PluginLog::new(manifest, log_slots)
                    .map_err(|_| AdminError::invalid("插件清单身份不合法"))?,
            ),
+            clients: clients.clone(),
            models: Arc::new(model::PluginModels::new(
                ports.models,
+                clients,
                request_limits.clone(),
                maximum_payload,
            )),
            private_state,
            http: ports.http,
-            network: Arc::new(ports.network),
+            network: Arc::new(NetworkPolicy::unrestricted()),
            scopes: Mutex::new(BTreeMap::new()),
            pending_middleware: Mutex::new(BTreeMap::new()),
            calls: Mutex::new(BTreeMap::new()),
@@ -437,6 +420,7 @@ impl CallbackHandler for PluginCallbacks {
        let resources = self.resources.clone();
        let data = self.data.clone();
        let keys = self.keys.clone();
+        let clients = self.clients.clone();
        let models = self.models.clone();
        let request_limits = self.request_limits.clone();
        let affinity = self.affinity.clone();
@@ -513,10 +497,14 @@ impl CallbackHandler for PluginCallbacks {
            }
            if matches!(
                method.as_str(),
-                "host.models.list"
-                    | gateway_plugin_sdk::call::data::CLIENT_VERIFY
+                gateway_plugin_sdk::call::data::CLIENT_VERIFY
                    | gateway_plugin_sdk::call::data::ACCESS_CHECK
            ) {
+                return clients
+                    .call(&context, &call, &method, params, payload)
+                    .await;
+            }
+            if method == "host.models.list" {
                return models.call(&context, &call, &method, params, payload).await;
            }
            if matches!(
@@ -1,4 +1,4 @@
-//! 复用 Core 的身份查询与模型执行入口；不拥有权限、路由、重试或计费规则
+//! 模型目录与执行适配；身份解析交给客户端端口，路由与计费沿用 Core

 use std::sync::{Arc, OnceLock, Weak};

@@ -8,7 +8,6 @@ use gateway_core::{
    account::ProviderAccountId,
    engine::{
        CommitRequirement, ModelRequestId,
-        authentication::ClientAuthenticationRequest,
        execution::{
            AuthenticatedClient, ClientTransport, ExecutionRequestMetadata, ExecutionSession,
        },
@@ -22,12 +21,10 @@ use gateway_core::{
        GenerateRequest, ImageRequest, ImageRequestKind, Operation, ProtocolPayload,
        RawJsonPayload, StandaloneSearchRequest,
    },
-    policy::ClientApiKeyId,
-    routing::{PublicModelId, UpstreamModelId},
+    routing::PublicModelId,
 };
 use gateway_plugin_sdk::{
    CallContext, ErrorCode, PluginFault,
-    call::data,
    call::host::{
        ModelEventBatch, ModelExecuteRequest, ModelExecuteResult, ModelListRequest,
        ModelListResult, ModelOperation, ModelStreamCloseRequest, ModelStreamReadRequest,
@@ -39,7 +36,6 @@ use super::{CallResources, denied, invalid};
 use crate::RpcReply;

 const MAX_MODEL_STREAMS_PER_CALL: usize = 16;
-const MAX_CLIENT_IDENTITIES_PER_CALL: usize = 16;

 pub(crate) struct PluginModelPortSlot {
    port: OnceLock<Weak<dyn NestedModelExecutionPort>>,
@@ -65,6 +61,7 @@ impl PluginModelPortSlot {

 pub(super) struct PluginModels {
    slot: Arc<PluginModelPortSlot>,
+    clients: Arc<super::client::PluginClients>,
    request_limits: Arc<super::request_limits::RequestLimits>,
    maximum_payload: usize,
 }
@@ -72,11 +69,13 @@ pub(super) struct PluginModels {
 impl PluginModels {
    pub(super) fn new(
        slot: Arc<PluginModelPortSlot>,
+        clients: Arc<super::client::PluginClients>,
        request_limits: Arc<super::request_limits::RequestLimits>,
        maximum_payload: usize,
    ) -> Self {
        Self {
            slot,
+            clients,
            request_limits,
            maximum_payload,
        }
@@ -91,64 +90,6 @@ impl PluginModels {
        payload: Vec<u8>,
    ) -> Result<RpcReply, PluginFault> {
        match method {
-            data::CLIENT_VERIFY => {
-                if params != serde_json::json!({}) {
-                    return Err(invalid());
-                }
-                let request: data::VerifyClientRequest =
-                    serde_json::from_slice(&payload).map_err(|_| invalid())?;
-                let mut authentication = ClientAuthenticationRequest::new(request.authorization)
-                    .map_err(|_| invalid())?
-                    .with_invocation(call.child_origin(context)?);
-                if let Some(settings) = call.request_settings() {
-                    authentication = authentication.with_settings(settings);
-                }
-                let port = self.slot.upgrade()?;
-                call.observe_dispatch()?;
-                let client = port
-                    .verify_client(authentication)
-                    .await
-                    .map_err(super::error::gateway)?;
-                super::admin::encode(&data::VerifiedClient {
-                    client_key_id: client.policy().key_id().as_str().to_owned(),
-                })
-            }
-            data::ACCESS_CHECK => {
-                if params != serde_json::json!({}) {
-                    return Err(invalid());
-                }
-                let request: data::AccessQuery =
-                    serde_json::from_slice(&payload).map_err(|_| invalid())?;
-                let port = self.slot.upgrade()?;
-                let client = self.identity(&port, call, request.client_key_id).await?;
-                let scope = client.policy().account_scope();
-                let model = request
-                    .upstream_model
-                    .map(UpstreamModelId::new)
-                    .transpose()
-                    .map_err(|_| invalid())?;
-                let accounts = request
-                    .account_ids
-                    .into_iter()
-                    .map(|account_id| {
-                        let account =
-                            ProviderAccountId::new(account_id.clone()).map_err(|_| invalid())?;
-                        let allowed = model.as_ref().map_or_else(
-                            || scope.allows(&account),
-                            |model| scope.allows_model(&account, model.as_str()),
-                        );
-                        Ok(data::AccountAccessFacts {
-                            account_id,
-                            allowed,
-                        })
-                    })
-                    .collect::<Result<Vec<_>, PluginFault>>()?;
-                super::admin::encode(&data::AccessFacts {
-                    schema_version: 1,
-                    client_key_id: client.policy().key_id().as_str().to_owned(),
-                    accounts,
-                })
-            }
            "host.models.list" => {
                if !payload.is_empty() {
                    return Err(invalid());
@@ -162,7 +103,7 @@ impl PluginModels {
                    return Err(invalid());
                }
                let port = self.slot.upgrade()?;
-                let client = self.identity(&port, call, request.client_key_id).await?;
+                let client = self.clients.identity(call, request.client_key_id).await?;
                let models = port
                    .models(
                        client,
@@ -196,7 +137,7 @@ impl PluginModels {
                let port = self.slot.upgrade()?;
                let stream = method == "host.model.execute_stream";
                let started = if let Some(client_key_id) = request.client_key_id.take() {
-                    let client = self.identity(&port, call, client_key_id).await?;
+                    let client = self.clients.identity(call, client_key_id).await?;
                    check_size(client.snapshot().responses_max_decompressed_body_bytes())?;
                    let request = self.request_parts(request, payload, stream)?;
                    let request = Self::bound_request(context, call, request, client);
@@ -295,38 +236,6 @@ impl PluginModels {
        }
    }

-    async fn identity(
-        &self,
-        port: &Arc<dyn NestedModelExecutionPort>,
-        call: &CallResources,
-        client_key_id: String,
-    ) -> Result<AuthenticatedClient, PluginFault> {
-        let key = ClientApiKeyId::new(client_key_id.clone()).map_err(|_| invalid())?;
-        let settings = call.request_settings();
-        let mut identities = call.client_identities.lock().await;
-        if let Some(client) = identities
-            .get(&client_key_id)
-            .filter(|client| client.request_settings() == settings.as_ref())
-        {
-            return Ok(client.clone());
-        }
-        if !identities.contains_key(&client_key_id)
-            && identities.len() >= MAX_CLIENT_IDENTITIES_PER_CALL
-        {
-            return Err(PluginFault::new(
-                ErrorCode::Capacity,
-                "client identity capacity is exhausted",
-            ));
-        }
-        let client = port
-            .resolve_client(key, settings)
-            .await
-            .map_err(super::error::gateway)?;
-        // 查询与执行共用调用内快照，执行链和子执行取消只在真正启动时建立
-        identities.insert(client_key_id, client.clone());
-        Ok(client)
-    }
-
    fn request(
        &self,
        context: &CallContext,
@@ -161,15 +161,16 @@ impl PluginRuntime {
            &manifest,
            self.log_slots.clone(),
            private_state.clone(),
-            PluginCallbackPorts::new(
-                self.service_ports.clone(),
-                self.http.clone(),
-                self.account_ports.clone(),
-                self.client_key_ports.clone(),
-                self.model_ports.clone(),
-                self.affinity_ports.clone(),
-                self.resource_ports.clone(),
-            ),
+            PluginCallbackPorts {
+                services: self.service_ports.clone(),
+                http: self.http.clone(),
+                accounts: self.account_ports.clone(),
+                keys: self.client_key_ports.clone(),
+                clients: self.client_identity_port.clone(),
+                models: self.model_ports.clone(),
+                affinity: self.affinity_ports.clone(),
+                resources: self.resource_ports.clone(),
+            },
            self.request_limits.clone(),
        )?;
        callbacks.registration_body_limit = request_body_limit;
@@ -8,7 +8,8 @@ use super::{
 use crate::{
    PackageLimits, RpcLimits, RpcSession,
    callback::{
-        PluginAccountPortSlot, PluginAffinityPortSlot, PluginClientKeyPortSlot, PluginModelPortSlot,
+        PluginAccountPortSlot, PluginAffinityPortSlot, PluginClientIdentityPortSlot,
+        PluginClientKeyPortSlot, PluginModelPortSlot,
    },
 };
 use gateway_admin::{
@@ -21,7 +22,7 @@ use gateway_admin::{
 };
 use gateway_core::{
    engine::{
-        authentication::FrontendAuthenticationExtensionIndex,
+        authentication::{ClientIdentityPort, FrontendAuthenticationExtensionIndex},
        extensions::ExecutionExtensionIndex,
        nested::{AffinityLookupPort, NestedModelExecutionPort},
        observation::RequestObserverExtensionIndex,
@@ -69,6 +70,7 @@ pub struct PluginRuntime {
    pub(super) account_ports: Arc<PluginAccountPortSlot>,
    pub(super) client_key_ports: Arc<PluginClientKeyPortSlot>,
    pub(super) resource_ports: Arc<crate::callback::PluginResourcePorts>,
+    pub(super) client_identity_port: Arc<PluginClientIdentityPortSlot>,
    pub(super) model_ports: Arc<PluginModelPortSlot>,
    pub(super) affinity_ports: Arc<PluginAffinityPortSlot>,
    pub(super) observers: RequestObserverExtensionIndex,
@@ -113,6 +115,7 @@ impl PluginRuntime {
            account_ports: Arc::new(PluginAccountPortSlot::new()),
            client_key_ports: Arc::new(PluginClientKeyPortSlot::new()),
            resource_ports: Arc::new(crate::callback::PluginResourcePorts::new()),
+            client_identity_port: Arc::new(PluginClientIdentityPortSlot::new()),
            model_ports: Arc::new(PluginModelPortSlot::new()),
            affinity_ports: Arc::new(PluginAffinityPortSlot::new()),
            observers: RequestObserverExtensionIndex::default(),
@@ -168,6 +171,14 @@ impl PluginRuntime {
        self.client_key_ports.bind(access)
    }

+    /// 身份查询独立装配，复用 Core 的认证与冻结权限，不要求绑定模型执行端口
+    pub fn bind_client_identity_port(
+        &self,
+        clients: &Arc<dyn ClientIdentityPort>,
+    ) -> Result<(), AdminError> {
+        self.client_identity_port.bind(clients)
+    }
+
    /// Core 激活后一次性绑定嵌套执行与亲和查询端口；Runtime 仅保存 Weak，避免组合根强环
    pub fn bind_model_ports(
        &self,
@@ -276,6 +276,15 @@ async fn startup_failure_of_an_unrelated_plugin_preserves_native_authentication_

 #[tokio::test]
 async fn callback_verification_reenters_the_authentication_instance_without_model_binding() {
+    assert_callback_identity_queries(true).await;
+}
+
+#[tokio::test]
+async fn client_queries_do_not_require_model_execution_ports() {
+    assert_callback_identity_queries(false).await;
+}
+
+async fn assert_callback_identity_queries(include_models: bool) {
    use gateway_admin::{
        model::plugins::management::PluginManagementRequest,
        ports::plugin_management::PluginManagement,
@@ -292,10 +301,14 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode
        .directory
        .path()
        .join("authenticated-facts.json");
-    let facts = json!([
-        {"method":"host.data.access.check", "query":{"client_key_id":"key_frontend", "account_ids":[account.as_str(), "acct_missing"]}},
-        {"method":"host.models.list", "params":{"client_key_id":"key_frontend", "protocol":"codex", "client_version":"0.154.0"}}
-    ]);
+    let mut facts = vec![json!(
+        {"method":"host.data.access.check", "query":{"client_key_id":"key_frontend", "account_ids":[account.as_str(), "acct_missing"]}}
+    )];
+    if include_models {
+        facts.push(json!(
+            {"method":"host.models.list", "params":{"client_key_id":"key_frontend", "protocol":"codex", "client_version":"0.154.0"}}
+        ));
+    }
    let mut config = configuration(
        json!({"outcome":"authenticated","principal":"external-user"}),
        true,
@@ -303,20 +316,32 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode
    );
    config["sdk_session"] = json!(true);
    config["frontend_authentication_marker"] = json!(marker);
-    config["auth_queries"] = facts.clone();
+    config["auth_queries"] = json!(facts);
    config["management_registration"] = json!({"routes":[{"method":"GET","path":"facts","request_content_types":[],"response_content_types":["application/json"]}]});
-    config["data_queries"] = json!([
-        {"method":"host.client.verify", "query":{"authorization":"External controlled-fixture"}},
-        facts[0].clone(), facts[1].clone()
-    ]);
+    let mut queries = vec![json!(
+        {"method":"host.client.verify", "query":{"authorization":"External controlled-fixture"}}
+    )];
+    queries.extend(facts);
+    config["data_queries"] = json!(queries);
    environment.install_plugin(config).await;
    let (runtime, core) = environment
-        .runtime_with_limits(RpcLimits {
-            maximum_executing_calls: 1,
-            maximum_queue_timeout: std::time::Duration::from_millis(500),
-            ..RpcLimits::default()
-        })
+        .runtime_with_identity_port(
+            RpcLimits {
+                maximum_executing_calls: 1,
+                maximum_queue_timeout: std::time::Duration::from_millis(500),
+                ..RpcLimits::default()
+            },
+            crate::support::native::provider_registry(),
+        )
        .await;
+    if include_models {
+        runtime
+            .bind_model_ports(
+                &core.nested_model_execution_port(),
+                &core.affinity_lookup_port(),
+            )
+            .unwrap();
+    }
    let snapshot = core.snapshots().acquire().unwrap();
    let generation = snapshot.extensions().unwrap();
    let view = runtime.views(generation).await.unwrap().remove(0);
@@ -348,9 +373,11 @@ async fn callback_verification_reenters_the_authentication_instance_without_mode
            {"account_id":account.as_str(),"allowed":true}, {"account_id":"acct_missing","allowed":false}
        ])
    );
-    assert_eq!(authenticated_facts[1], results[2]);
-    // 此替身只登记执行能力，没有原生目录；查询应成功返回空目录而不是执行绑定错误
-    assert_eq!(results[2], json!({"models":[]}));
+    if include_models {
+        assert_eq!(authenticated_facts[1], results[2]);
+        // 此替身只登记执行能力，没有原生目录；查询应成功返回空目录而不是执行绑定错误
+        assert_eq!(results[2], json!({"models":[]}));
+    }
    assert!(
        environment
            .bound_model_requests("key_frontend")
@@ -463,6 +463,21 @@ impl Environment {
        &self,
        limits: RpcLimits,
        providers: gateway_core::engine::provider::ProviderRegistry,
+    ) -> (Arc<PluginRuntime>, gateway_core::CoreBundle) {
+        let (runtime, core) = self.runtime_with_identity_port(limits, providers).await;
+        runtime
+            .bind_model_ports(
+                &core.nested_model_execution_port(),
+                &core.affinity_lookup_port(),
+            )
+            .unwrap();
+        (runtime, core)
+    }
+
+    pub async fn runtime_with_identity_port(
+        &self,
+        limits: RpcLimits,
+        providers: gateway_core::engine::provider::ProviderRegistry,
    ) -> (Arc<PluginRuntime>, gateway_core::CoreBundle) {
        let runtime = self.plugin_runtime_with_limits(limits);
        let core = gateway_core::prepare(
@@ -487,10 +502,7 @@ impl Environment {
            .insert(Arc::as_ptr(&runtime) as usize, access);
        let core = core.activate().await.unwrap();
        runtime
-            .bind_model_ports(
-                &core.nested_model_execution_port(),
-                &core.affinity_lookup_port(),
-            )
+            .bind_client_identity_port(&core.client_identity_port())
            .unwrap();
        (runtime, core)
    }
@@ -511,6 +523,9 @@ impl Environment {
        .await
        .unwrap();
        runtime.bind_snapshots(&core.snapshots()).unwrap();
+        runtime
+            .bind_client_identity_port(&core.client_identity_port())
+            .unwrap();
        runtime
            .bind_model_ports(
                &core.nested_model_execution_port(),
@@ -112,6 +112,7 @@ flowchart LR
  Runtime <-->|SDK 协议| Plugin["插件进程"]
  Runtime -->|受管资源| Host["Host<br/>进程、网络、Worker"]
  Runtime -->|账号事务端口| Admin
+  Runtime -->|身份查询端口| Core
  Runtime -->|模型执行端口| Core
 ```

@@ -183,8 +184,9 @@ flowchart LR
  Provider 来源与执行事件快照可由插件读取，包含费用明细、上游观测、会话状态和原始 wire；读取快照不转移计量和资源所有权。
  数据面插件通过受管 HTTP 已发送、无法证明未发送，或进入主动 HTTP／公开服务调用链时，Runtime 把该事实并入 Core 的请求副作用水位，后续不能按
  Provider 的 `not_sent` 结果透明重放；RPC 返回或取消前同步封存未完成的发送，并禁止已关闭调用再分派 HTTP／公开服务
- **身份查询与执行准入分离**：Core 的 `AuthenticatedClient` 持有身份和冻结权限，模型目录与账号授权直接读取这份事实。
-  Runtime 只按父调用缓存身份快照，查询不建立模型执行图；真正发起模型执行时才检查深度、建立执行归属并申请原生资源。
+- **身份查询与执行准入分离**：Core 通过 `ClientIdentityPort` 独立提供身份解析和认证，`AuthenticatedClient` 持有身份和冻结权限，模型目录与账号授权直接读取这份事实。
+  Runtime 的客户端适配器承接认证与授权查询，模型适配器复用同一身份解析。
+  身份快照只按父调用缓存，查询不建立模型执行图；真正发起模型执行时才检查深度、建立执行归属并申请原生资源。
  认证复用原有认证计划与 principal 映射，通用 `InvocationOrigin` 传递父调用来源、取消、冻结计划和副作用归属，认证不依赖模型执行上下文
 - **挂起与恢复由共同边界管理**：SDK 在父处理 future 返回 `Pending` 后报告挂起，再次 poll 前等待 Runtime 的执行授权。
  Runtime 的调用表持有唯一准入状态，挂起保留总在途名额；结果接收独立于任务恢复，回调额度持续保留到消费、丢弃或父调用终结。