fix(plugins): separate execution capacity from downstream waits (#434)

* fix(plugins): delegate HTTP requests on RPC admission exhaustion

Co-authored-by: Codex <noreply@openai.com>

* fix(plugins): align default RPC capacity at 64

Raise host and SDK call and callback limits together, exercise 64 in-flight HTTP calls, and document plugin rebuild requirements.

Co-authored-by: Codex <noreply@openai.com>

* fix(plugins): preserve middleware execution under RPC saturation

Remove capacity-triggered delegation, retain the 64-call defaults, and verify that saturation never bypasses a healthy HTTP plugin.

Co-authored-by: Codex <noreply@openai.com>

* fix(plugins): separate execution capacity from downstream waits

Bound initial admission and total in-flight resources while preserving middleware continuations. Keep queue waits independent of processing deadlines and report overload without bypassing plugins.

Co-authored-by: Codex <noreply@openai.com>

---------

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-api/src/middleware/mod.rs

@@ -17,9 +17,10 @@ use axum::{
     http::{HeaderName, HeaderValue, Request, StatusCode},
     response::{IntoResponse as _, Response},
 };
-use gateway_core::engine::middleware::FrozenMiddlewarePlan;
 use gateway_core::engine::middleware::http as contract;
 use gateway_core::engine::middleware::http::Settings;
+use gateway_core::engine::middleware::{FrozenMiddlewarePlan, MiddlewareError};
+use gateway_core::error::GatewayErrorKind;
 use gateway_core::lifecycle::CancellationToken;
 use gateway_core::middleware::compose;
 use http_body::{Body as HttpBody, Frame, SizeHint};
@@ -270,6 +271,11 @@ async fn handle(
             }
             response.map(|body| Body::new(ResponseBody::new(body, guard)))
         }
+        Err(MiddlewareError::Gateway(error))
+            if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable =>
+        {
+            crate::openai::error::gateway_error_response(&error)
+        }
         Err(error) => {
             let status = if error.is_rejected() {
                 StatusCode::FORBIDDEN

backend/crates/gateway-api/src/openai/error.rs

@@ -279,6 +279,11 @@ pub const fn gateway_error_contract(
             "server_error",
             "provider_infrastructure_unavailable",
         ),
+        GatewayErrorKind::MiddlewareCapacityUnavailable => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "middleware_capacity_unavailable",
+        ),
         GatewayErrorKind::ConcurrencyQueueFull => (
             StatusCode::TOO_MANY_REQUESTS,
             "rate_limit_error",

backend/crates/gateway-api/tests/middleware/mod.rs

@@ -35,6 +35,7 @@ use tower::ServiceExt as _;
 
 #[derive(Clone, Copy, Debug)]
 enum Mode {
+    Overloaded,
     Rewrite,
     Upgrade {
         reject: bool,
@@ -142,6 +143,12 @@ impl MiddlewarePlan for Plan {
         let mode = self.mode;
         Box::pin(async move {
             match mode {
+                Mode::Overloaded => Err(MiddlewareError::Gateway(
+                    gateway_core::error::GatewayError::new(
+                        gateway_core::error::GatewayErrorKind::MiddlewareCapacityUnavailable,
+                        "Request middleware is temporarily busy",
+                    ),
+                )),
                 Mode::Service => unreachable!("service-only plan bypasses HTTP layers"),
                 Mode::Upgrade { reject, idle, .. } => {
                     let upgrade = request
@@ -659,3 +666,25 @@ async fn plugin_http_dispatch_uses_host_identity_without_client_credentials() {
         .unwrap();
     assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
 }
+
+#[tokio::test]
+async fn http_middleware_overload_is_an_explicit_503_before_native_routing() {
+    let (router, plan) = router(Mode::Overloaded, None).await;
+    for path in [
+        "/healthz",
+        "/assets/index.js",
+        "/api/settings",
+        "/v1/responses",
+    ] {
+        let response = router
+            .clone()
+            .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap())
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let body = response.into_body().collect().await.unwrap().to_bytes();
+        let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(body["error"]["code"], "middleware_capacity_unavailable");
+    }
+    assert_eq!(plan.calls.load(Ordering::Acquire), 4);
+}

backend/crates/gateway-api/tests/openai/error.rs

@@ -263,6 +263,18 @@ fn rate_limited_error_should_map_to_openai_retryable_status() {
     );
 }
 
+#[test]
+fn middleware_capacity_is_service_unavailable() {
+    assert_eq!(
+        gateway_error_contract(GatewayErrorKind::MiddlewareCapacityUnavailable),
+        (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "server_error",
+            "middleware_capacity_unavailable"
+        ),
+    );
+}
+
 #[test]
 fn queue_rejections_keep_distinct_retryable_http_codes() {
     for (kind, code) in [

backend/crates/gateway-core/src/error/mod.rs

@@ -1067,6 +1067,8 @@ pub enum GatewayErrorKind {
     NoAvailableProvider,
     /// 符合条件的上游账号暂时没有调度容量
     AccountCapacityUnavailable,
+    /// 中间件在途资源或执行等待队列已满
+    MiddlewareCapacityUnavailable,
     ConcurrencyQueueFull,
     ConcurrencyQueueTimeout,
     /// Provider 的本地账号基础设施不可用
@@ -1098,6 +1100,7 @@ impl GatewayErrorKind {
             Self::ModelNotFound => "model_not_found",
             Self::NoAvailableProvider => "no_available_provider",
             Self::AccountCapacityUnavailable => "account_capacity_unavailable",
+            Self::MiddlewareCapacityUnavailable => "middleware_capacity_unavailable",
             Self::ConcurrencyQueueFull => "concurrency_queue_full",
             Self::ConcurrencyQueueTimeout => "concurrency_queue_timeout",
             Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable",

backend/crates/gateway-plugin/runtime/src/adapter/command_line/execution.rs

@@ -139,7 +139,7 @@ fn rpc_failure(error: RpcError) -> &'static str {
         RpcError::Handshake(_) | RpcError::Protocol(_) | RpcError::InvalidResponse(..) => {
             "插件协议错误"
         }
-        RpcError::Capacity => "插件调用容量不足",
+        RpcError::Capacity | RpcError::Overloaded => "插件调用容量不足",
         RpcError::Context(_) => "调用上下文无效",
         RpcError::Remote(_) => "插件拒绝命令",
     }

backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs

@@ -221,6 +221,7 @@ async fn recover_rpc(
         RpcError::Remote(fault) if fault.code == ErrorCode::Rejected
     );
     let delegate_safe = !rejected
+        && !error.is_overloaded()
         && !matches!(
             &error,
             RpcError::Remote(fault) if fault.send_state != SendState::NotSent

backend/crates/gateway-plugin/runtime/src/callback/error.rs

@@ -20,6 +20,7 @@ pub(super) fn gateway(error: GatewayError) -> PluginFault {
         GatewayErrorKind::Unauthorized => ErrorCode::PermissionDenied,
         GatewayErrorKind::PolicyDenied => ErrorCode::Rejected,
         GatewayErrorKind::AccountCapacityUnavailable
+        | GatewayErrorKind::MiddlewareCapacityUnavailable
         | GatewayErrorKind::ConcurrencyQueueFull
         | GatewayErrorKind::ConcurrencyQueueTimeout
         | GatewayErrorKind::RateLimited
@@ -142,13 +143,22 @@ pub(crate) fn rpc(error: &crate::RpcError) -> PluginFault {
         RpcError::Remote(fault) => return fault.clone(),
         RpcError::Timeout => ErrorCode::Timeout,
         RpcError::Cancelled => ErrorCode::Cancelled,
-        RpcError::Capacity => ErrorCode::Capacity,
+        RpcError::Capacity | RpcError::Overloaded => ErrorCode::Capacity,
         _ => ErrorCode::Fault,
     };
     PluginFault::new(code, error.to_string())
 }
 
 pub(crate) fn rpc_middleware(error: crate::RpcError) -> MiddlewareError {
+    if error.is_overloaded() {
+        return MiddlewareError::Gateway(
+            GatewayError::new(
+                GatewayErrorKind::MiddlewareCapacityUnavailable,
+                "Request middleware is temporarily busy",
+            )
+            .with_source(gateway_core::error::ErrorSource::new(error)),
+        );
+    }
     let rejected =
         matches!(&error, crate::RpcError::Remote(fault) if fault.code == ErrorCode::Rejected);
     MiddlewareError::Remote {

backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs

@@ -194,7 +194,10 @@ impl PluginRuntime {
             .map_err(|_| AdminError::internal("插件集合 ID 无效"))?;
         let observers = crate::adapter::observer::PluginObserverPlan::compile(
             observer_entries,
-            self.config.rpc_limits.maximum_calls,
+            self.config
+                .rpc_limits
+                .maximum_calls
+                .min(self.config.rpc_limits.maximum_executing_calls),
             self.config.rpc_limits.maximum_call_timeout,
             self.config.rpc_limits.maximum_buffered_body_bytes,
         )

backend/crates/gateway-plugin/runtime/src/rpc/admission.rs

@@ -0,0 +1,50 @@
+//! 区分插件执行名额与初次准入队列，恢复中的调用沿用在途资源预算
+
+use std::{sync::Arc, time::Duration};
+
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+
+use super::RpcError;
+
+pub(super) struct ExecutionCapacity {
+    active: Arc<Semaphore>,
+    queued: Arc<Semaphore>,
+    wait_timeout: Duration,
+}
+
+impl ExecutionCapacity {
+    pub(super) fn new(active: usize, queued: usize, wait_timeout: Duration) -> Self {
+        Self {
+            active: Arc::new(Semaphore::new(active)),
+            queued: Arc::new(Semaphore::new(queued)),
+            wait_timeout,
+        }
+    }
+
+    pub(super) async fn enter(&self) -> Result<OwnedSemaphorePermit, RpcError> {
+        if let Ok(permit) = self.active.clone().try_acquire_owned() {
+            return Ok(permit);
+        }
+        let _queued = self.queued.try_acquire().map_err(|error| {
+            if matches!(error, tokio::sync::TryAcquireError::Closed) {
+                RpcError::Closed(None)
+            } else {
+                RpcError::Overloaded
+            }
+        })?;
+        self.resume().await
+    }
+
+    /// 恢复的等待者已持有在途名额，不再竞争初次准入队列
+    pub(super) async fn resume(&self) -> Result<OwnedSemaphorePermit, RpcError> {
+        tokio::time::timeout(self.wait_timeout, self.active.clone().acquire_owned())
+            .await
+            .map_err(|_| RpcError::Overloaded)?
+            .map_err(|_| RpcError::Closed(None))
+    }
+
+    pub(super) fn close(&self) {
+        self.active.close();
+        self.queued.close();
+    }
+}

backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs

@@ -125,7 +125,7 @@ async fn read_loop<R: AsyncRead + Unpin>(
                     Err(RpcError::Protocol(None))
                 } else if let Some(context) = shared.context(parent_id) {
                     last_callback = id;
-                    if let Some(permit) = shared.try_callback_slot() {
+                    if let Some(permit) = shared.try_callback_slot(&method) {
                         let handler = Arc::clone(&callbacks);
                         let response = Arc::clone(&shared);
                         let (ready, started) = oneshot::channel();
@@ -138,6 +138,12 @@ async fn read_loop<R: AsyncRead + Unpin>(
                                 return;
                             }
                             let result = handler.call(context, method, params, frame.payload).await;
+                            if pause.is_some()
+                                && let Err(error) = response.resume_next(parent_id).await
+                            {
+                                response.cancel(parent_id, error);
+                                return;
+                            }
                             drop(pause);
                             let mut reply = match result {
                                 Ok(reply) => Frame {

backend/crates/gateway-plugin/runtime/src/rpc/mod.rs

@@ -1,5 +1,6 @@
 //! 插件双向 RPC 会话、调用结果与生命周期的公共入口
 
+mod admission;
 mod dispatch;
 mod session;
 

backend/crates/gateway-plugin/runtime/src/rpc/session.rs

@@ -15,17 +15,21 @@ use gateway_plugin_sdk::{
     CallContext, Frame, Handshake, Message, PROTOCOL_VERSION, PluginFault, Stage,
     client::{CallDeadline, CallDeadlinePause, read_frame, validate_frame, write_frame},
 };
-use tokio::sync::{Mutex as AsyncMutex, Semaphore, mpsc, oneshot, watch};
+use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch};
 
 use crate::{PreparedPackage, RpcStream, stream::StreamIngress};
 
-use super::dispatch;
+use super::{admission::ExecutionCapacity, dispatch};
 
 #[derive(Debug, Clone, Copy)]
 pub struct RpcLimits {
     /// 受管回调与观察队列的缓冲预算，不是请求正文或 IPC 消息总量上限
     pub maximum_buffered_body_bytes: usize,
+    /// 包含初次排队、下游等待、取消确认和未消费流的总在途上限
     pub maximum_calls: usize,
+    pub maximum_executing_calls: usize,
+    pub maximum_queued_calls: usize,
+    pub maximum_queue_timeout: Duration,
     pub maximum_callbacks: usize,
     pub handshake_timeout: Duration,
     pub maximum_call_timeout: Duration,
@@ -36,8 +40,11 @@ impl Default for RpcLimits {
     fn default() -> Self {
         Self {
             maximum_buffered_body_bytes: 1024 * 1024,
-            maximum_calls: 16,
-            maximum_callbacks: 16,
+            maximum_calls: 256,
+            maximum_executing_calls: 64,
+            maximum_queued_calls: 64,
+            maximum_queue_timeout: Duration::from_secs(30),
+            maximum_callbacks: 64,
             handshake_timeout: Duration::from_secs(5),
             maximum_call_timeout: Duration::from_secs(120),
             maximum_stderr_bytes: 64 * 1024,
@@ -63,6 +70,8 @@ pub enum RpcError {
     Cancelled,
     #[error("plugin call capacity is exhausted")]
     Capacity,
+    #[error("plugin admission capacity is exhausted")]
+    Overloaded,
     #[error("plugin call context is invalid")]
     Context(#[source] Option<gateway_core::error::ErrorSource>),
     #[error("plugin returned an error")]
@@ -70,6 +79,11 @@ pub enum RpcError {
 }
 
 impl RpcError {
+    pub(crate) fn is_overloaded(&self) -> bool {
+        matches!(self, Self::Overloaded)
+            || matches!(self, Self::Remote(fault) if fault.code == gateway_plugin_sdk::ErrorCode::Capacity)
+    }
+
     /// 会话状态与请求错误共用安全分类，不展开插件返回正文
     pub(crate) fn diagnostic(&self) -> (&'static str, &'static str) {
         match self {
@@ -101,6 +115,7 @@ impl RpcError {
             RpcError::Timeout => ("timeout", "插件调用超时"),
             RpcError::Cancelled => ("cancelled", "插件会话已取消"),
             RpcError::Capacity => ("capacity", "插件调用容量已耗尽"),
+            RpcError::Overloaded => ("overloaded", "插件暂时繁忙"),
             RpcError::Context(_) => ("context", "插件调用上下文无效"),
             RpcError::Remote(_) => ("remote", "插件返回错误"),
         }
@@ -216,6 +231,8 @@ pub(crate) struct Shared {
     pub process: ProcessControl,
     pub stopped: watch::Sender<bool>,
     callback_slots: Arc<Semaphore>,
+    next_slots: Arc<Semaphore>,
+    execution: ExecutionCapacity,
 }
 
 struct State {
@@ -224,6 +241,8 @@ struct State {
 }
 
 struct PendingCall {
+    _slot: Arc<OwnedSemaphorePermit>,
+    execution: Option<OwnedSemaphorePermit>,
     deadline: CallDeadline,
     handler: Arc<dyn CallbackHandler>,
     transmitted: bool,
@@ -274,6 +293,7 @@ impl Shared {
                 .flatten();
             (std::mem::take(&mut state.calls), observer)
         };
+        self.execution.close();
         for call in calls.into_values() {
             call.finish(Err(error.clone()));
         }
@@ -303,6 +323,7 @@ impl Shared {
                 call.deadline.clear();
             }
             let initial = call.result.take().ok_or(RpcError::Protocol(None))?;
+            call.execution.take();
             call.handler.streaming(&call.context);
             let _ = initial.send(result);
         } else {
@@ -384,11 +405,40 @@ impl Shared {
     }
 
     pub fn pause_next(&self, id: u64) -> Option<CallDeadlinePause> {
-        self.state()
+        let mut state = self.state();
+        let call = state
             .calls
-            .get(&id)
+            .get_mut(&id)
+            .filter(|call| call.cancellation.is_none())?;
+        let pause = call.deadline.pause()?;
+        call.execution.take();
+        Some(pause)
+    }
+
+    pub async fn resume_next(&self, id: u64) -> Result<(), RpcError> {
+        {
+            let state = self.state();
+            let call = state
+                .calls
+                .get(&id)
+                .filter(|call| call.cancellation.is_none())
+                .ok_or(RpcError::Cancelled)?;
+            // 已交付响应头的流只占在途资源，不重新占用首结果执行名额
+            if call.result.is_none() {
+                return Ok(());
+            }
+        }
+        let permit = self.execution.resume().await?;
+        let mut state = self.state();
+        let call = state
+            .calls
+            .get_mut(&id)
             .filter(|call| call.cancellation.is_none())
-            .and_then(|call| call.deadline.pause())
+            .ok_or(RpcError::Cancelled)?;
+        if call.result.is_some() {
+            call.execution = Some(permit);
+        }
+        Ok(())
     }
 
     pub fn is_cancelling(&self, id: u64) -> bool {
@@ -495,21 +545,23 @@ impl Shared {
         }
     }
 
-    pub(super) fn try_callback_slot(&self) -> Option<tokio::sync::OwnedSemaphorePermit> {
-        self.callback_slots.clone().try_acquire_owned().ok()
+    pub(super) fn try_callback_slot(&self, method: &str) -> Option<OwnedSemaphorePermit> {
+        let slots = if method == gateway_plugin_sdk::call::middleware::NEXT_METHOD {
+            &self.next_slots
+        } else {
+            &self.callback_slots
+        };
+        slots.clone().try_acquire_owned().ok()
     }
 
-    async fn wait_for_callbacks(&self, maximum_callbacks: usize) {
-        let Ok(maximum_callbacks) = u32::try_from(maximum_callbacks) else {
-            return;
-        };
-        if let Ok(permits) = self
-            .callback_slots
-            .clone()
-            .acquire_many_owned(maximum_callbacks)
-            .await
-        {
-            drop(permits);
+    async fn wait_for_callbacks(&self, limits: RpcLimits) {
+        for (slots, maximum) in [
+            (&self.callback_slots, limits.maximum_callbacks),
+            (&self.next_slots, limits.maximum_calls),
+        ] {
+            if let Ok(permits) = slots.clone().acquire_many_owned(maximum as u32).await {
+                drop(permits);
+            }
         }
     }
 }
@@ -558,6 +610,10 @@ impl RpcSession {
     ) -> Result<Self, RpcError> {
         if limits.maximum_calls == 0
             || limits.maximum_calls > 256
+            || limits.maximum_executing_calls == 0
+            || limits.maximum_executing_calls > 256
+            || limits.maximum_queued_calls > 256
+            || limits.maximum_queue_timeout.is_zero()
             || limits.maximum_callbacks == 0
             || limits.maximum_callbacks > 256
             || limits.maximum_buffered_body_bytes < 1024
@@ -657,7 +713,7 @@ impl RpcSession {
         }
         let (data, data_receiver) = mpsc::channel(limits.maximum_calls);
         let (control, control_receiver) =
-            mpsc::channel(limits.maximum_callbacks + limits.maximum_calls + 8);
+            mpsc::channel(limits.maximum_callbacks + 2 * limits.maximum_calls + 8);
         let (stopped, _) = watch::channel(false);
         let callback_slots = Arc::new(Semaphore::new(limits.maximum_callbacks));
         let shared = Arc::new(Shared {
@@ -674,6 +730,12 @@ impl RpcSession {
             process,
             stopped,
             callback_slots,
+            next_slots: Arc::new(Semaphore::new(limits.maximum_calls)),
+            execution: ExecutionCapacity::new(
+                limits.maximum_executing_calls,
+                limits.maximum_queued_calls,
+                limits.maximum_queue_timeout,
+            ),
         });
         dispatch::start(
             input,
@@ -754,11 +816,7 @@ impl RpcSession {
         }
     }
 
-    fn call_deadline(
-        &self,
-        method: &str,
-        context: &CallContext,
-    ) -> Result<tokio::time::Instant, RpcError> {
+    fn call_timeout(&self, method: &str, context: &CallContext) -> Result<Duration, RpcError> {
         if context.instance_id != self.handshake.instance_id
             || context.generation != self.handshake.generation
             || context.incarnation != self.handshake.incarnation
@@ -769,7 +827,7 @@ impl RpcSession {
         {
             return Err(RpcError::Context(None));
         }
-        Ok(tokio::time::Instant::now() + Duration::from_millis(context.timeout_ms))
+        Ok(Duration::from_millis(context.timeout_ms))
     }
 
     fn begin_call(
@@ -777,6 +835,8 @@ impl RpcSession {
         context: CallContext,
         deadline: tokio::time::Instant,
         stream: Option<StreamIngress>,
+        slot: Arc<OwnedSemaphorePermit>,
+        execution: OwnedSemaphorePermit,
     ) -> Result<StartedCall, RpcError> {
         let id = context.call_id;
         let (result, received) = oneshot::channel();
@@ -784,14 +844,13 @@ impl RpcSession {
         if let Some(error) = &state.failure {
             return Err(error.clone());
         }
-        if state.calls.len() >= self.limits.maximum_calls {
-            return Err(RpcError::Capacity);
-        }
         let deadline = CallDeadline::new(deadline);
         self.callbacks.begin(&context, &deadline);
         state.calls.insert(
             id,
             PendingCall {
+                _slot: slot,
+                execution: Some(execution),
                 deadline: deadline.clone(),
                 handler: self.callbacks.clone(),
                 transmitted: false,
@@ -818,7 +877,16 @@ impl RpcSession {
         stream: Option<StreamIngress>,
         credit: Option<(u32, u32)>,
     ) -> Result<CompletedCall, RpcError> {
-        let deadline = self.call_deadline(method, &context)?;
+        let timeout = self.call_timeout(method, &context)?;
+        let slot = Arc::new(self.slots.clone().try_acquire_owned().map_err(|error| {
+            if matches!(error, tokio::sync::TryAcquireError::Closed) {
+                RpcError::Closed(None)
+            } else {
+                RpcError::Overloaded
+            }
+        })?);
+        let execution = self.shared.execution.enter().await?;
+        let deadline = tokio::time::Instant::now() + timeout;
         // FIFO 锁同时拥有 ID 与 data 入队顺序；Call/初始 Credit 入队后必须立即释放，
         // 不能把插件回复或重入回调纳入串行区，也不能为等待锁重置调用期限
         let mut next_call = tokio::time::timeout_at(deadline, self.next_call.lock())
@@ -841,7 +909,7 @@ impl RpcSession {
         validate_frame(&frame).map_err(|source| {
             RpcError::Context(Some(gateway_core::error::ErrorSource::new(source)))
         })?;
-        let started = self.begin_call(context, deadline, stream)?;
+        let started = self.begin_call(context, deadline, stream, slot.clone(), execution)?;
         let mut guard = CallGuard {
             shared: Arc::clone(&self.shared),
             id: started.id,
@@ -906,6 +974,7 @@ impl RpcSession {
             }
         };
         Ok(CompletedCall {
+            slot,
             id: started.id,
             deadline: started.deadline,
             reply,
@@ -919,7 +988,6 @@ impl RpcSession {
         params: serde_json::Value,
         payload: Vec<u8>,
     ) -> Result<RpcReply, RpcError> {
-        let _slot = self.slots.try_acquire().map_err(|_| RpcError::Capacity)?;
         self.send_call(method, context, params, payload, None, None)
             .await
             .map(|completed| completed.reply)
@@ -932,9 +1000,6 @@ impl RpcSession {
         params: serde_json::Value,
         payload: Vec<u8>,
     ) -> Result<RpcStream, RpcError> {
-        let slot = Arc::clone(&self.slots)
-            .try_acquire_owned()
-            .map_err(|_| RpcError::Capacity)?;
         let bytes = self.limits.maximum_buffered_body_bytes.min(256 * 1024) as u32;
         let frames = 32;
         let (stream, chunks, terminal) = StreamIngress::new(bytes, frames);
@@ -955,7 +1020,7 @@ impl RpcSession {
             deadline: completed.deadline,
             id: completed.id,
             shared: Arc::clone(&self.shared),
-            _slot: slot,
+            _slot: completed.slot,
             _session: Arc::clone(self),
         })
     }
@@ -988,9 +1053,7 @@ impl RpcSession {
         self.shared.process.exited().await;
         // 不等待监控任务调度；先撤销父调用并中止其回调，再确认回调 future 已析构
         self.shared.fail(RpcError::Closed(None));
-        self.shared
-            .wait_for_callbacks(self.limits.maximum_callbacks)
-            .await;
+        self.shared.wait_for_callbacks(self.limits).await;
     }
 }
 
@@ -1022,6 +1085,7 @@ struct StartedCall {
 }
 
 struct CompletedCall {
+    slot: Arc<OwnedSemaphorePermit>,
     id: u64,
     deadline: CallDeadline,
     reply: RpcReply,

backend/crates/gateway-plugin/runtime/src/stream/bridge.rs

@@ -62,7 +62,7 @@ pub struct RpcStream {
     pub(crate) deadline: CallDeadline,
     pub(crate) id: u64,
     pub(crate) shared: Arc<Shared>,
-    pub(crate) _slot: OwnedSemaphorePermit,
+    pub(crate) _slot: Arc<OwnedSemaphorePermit>,
     pub(crate) _session: Arc<RpcSession>,
 }
 

backend/crates/gateway-plugin/runtime/tests/callback/log.rs

@@ -99,6 +99,8 @@ impl tracing::Subscriber for Capture {
 
 #[tokio::test]
 async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() {
+    // 低于每秒日志上限，单独验证阻塞写入的槽位及跨代保留行为
+    const MAXIMUM_CALLBACKS: usize = 16;
     let capture = Capture::default();
     tracing::subscriber::set_global_default(capture.clone()).unwrap();
     let directory = tempfile::tempdir().unwrap();
@@ -155,7 +157,10 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() {
             cache_directory: directory.path().join("cache"),
             host_version: "1.0.0".parse().unwrap(),
             package_limits: PackageLimits::default(),
-            rpc_limits: RpcLimits::default(),
+            rpc_limits: RpcLimits {
+                maximum_callbacks: MAXIMUM_CALLBACKS,
+                ..RpcLimits::default()
+            },
             restart_circuit: Default::default(),
         },
         Arc::new(gateway_host::outbound::HttpClient::new().unwrap()),
@@ -270,7 +275,7 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() {
     blocked.config_revision = Revision::new(9).unwrap();
     blocked.instances[0].id = "log-blocked".into();
     blocked.instances[0].configuration = json!({
-        "log_method":"plugin.register", "log_entries":[{"params":{"event":"fixture.blocked"},"repeat":32}],
+        "log_method":"plugin.register", "log_entries":[{"params":{"event":"fixture.blocked"},"repeat":MAXIMUM_CALLBACKS * 2}],
         "log_marker":blocked_marker,
     });
     // 日志后端停住时，注册的业务结果和父调用结束仍须及时完成
@@ -288,7 +293,7 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() {
             .iter()
             .filter(|result| result["recorded"] == true)
             .count(),
-        RpcLimits::default().maximum_callbacks
+        MAXIMUM_CALLBACKS
     );
     assert!(
         blocked_results

backend/crates/gateway-plugin/runtime/tests/generation/policy/http/capacity.rs

@@ -0,0 +1,162 @@
+//! 验证 next 等待不占执行名额，硬上限拒绝请求且不跳过插件
+
+use super::*;
+use gateway_core::error::GatewayErrorKind;
+
+fn context() -> core::Context {
+    core::Context {
+        plugin_instance_id: None,
+        call_id: uuid::Uuid::new_v4().to_string(),
+        parent_call_id: None,
+        extensions: Default::default(),
+        plan: None,
+        request_id: uuid::Uuid::new_v4().to_string(),
+        cancellation: CancellationToken::new(),
+    }
+}
+
+fn plugin_request() -> core::Request {
+    ::http::Request::builder()
+        .uri("/v1/responses")
+        .header("authorization", "Bearer fixture-secret")
+        .body(core::empty_body())
+        .unwrap()
+}
+
+#[tokio::test]
+async fn http_waiting_next_releases_execution_for_delegate_plugin() {
+    assert_capacity(PluginFailurePolicy::Delegate, RpcLimits::default(), false).await;
+}
+
+#[tokio::test]
+async fn http_waiting_next_releases_execution_for_reject_plugin() {
+    assert_capacity(PluginFailurePolicy::Reject, RpcLimits::default(), false).await;
+}
+
+#[tokio::test]
+async fn http_in_flight_limit_never_delegates_and_recovers_after_cancellation() {
+    for policy in [PluginFailurePolicy::Delegate, PluginFailurePolicy::Reject] {
+        assert_capacity(
+            policy,
+            RpcLimits {
+                maximum_calls: 64,
+                ..Default::default()
+            },
+            true,
+        )
+        .await;
+    }
+}
+
+async fn assert_capacity(policy: PluginFailurePolicy, limits: RpcLimits, saturated: bool) {
+    const CONCURRENT_CALLS: usize = 64;
+    let (cache, runtime, generation, plan) =
+        http_plugin_with_policy("passthrough", limits, policy).await;
+    let release = Arc::new(tokio::sync::Semaphore::new(0));
+    let (started, mut entered) = tokio::sync::mpsc::unbounded_channel();
+    let mut slow_calls = tokio::task::JoinSet::new();
+    for index in 0..CONCURRENT_CALLS {
+        let plan = plan.clone();
+        let release = release.clone();
+        let started = started.clone();
+        slow_calls.spawn(async move {
+            plan.handle_http(
+                context(),
+                plugin_request(),
+                compose(Vec::new(), move |request: core::Request| {
+                    assert_eq!(request.uri(), "/rewritten?encoded=%2F");
+                    let release = release.clone();
+                    Box::pin(async move {
+                        started.send(()).unwrap();
+                        release.acquire().await.unwrap().forget();
+                        Ok(core::Response::new(core::empty_body()))
+                    })
+                }),
+            )
+            .await
+        });
+        tokio::time::timeout(Duration::from_secs(5), entered.recv())
+            .await
+            .unwrap_or_else(|_| panic!("第 {} 个调用未能进入下游", index + 1))
+            .unwrap();
+    }
+
+    for uri in [
+        "/healthz",
+        "/assets/index.js",
+        "/api/settings",
+        "/v1/responses",
+    ] {
+        let request = ::http::Request::builder()
+            .uri(uri)
+            .header("authorization", "Bearer fixture-secret")
+            .body(core::empty_body())
+            .unwrap();
+        let result = tokio::time::timeout(
+            Duration::from_secs(3),
+            plan.handle_http(
+                context(),
+                request,
+                compose(Vec::new(), move |request: core::Request| {
+                    assert!(!saturated, "容量耗尽时不得执行下游");
+                    assert_eq!(request.uri(), "/rewritten?encoded=%2F");
+                    Box::pin(async { Ok(core::Response::new(core::empty_body())) })
+                }),
+            ),
+        )
+        .await
+        .expect("下游等待不应阻塞其它请求");
+        if saturated {
+            assert!(matches!(result, Err(MiddlewareError::Gateway(error))
+                if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable));
+        } else {
+            assert_eq!(result.unwrap().headers()["x-http-plugin"], "active");
+        }
+    }
+
+    if saturated {
+        slow_calls.abort_all();
+        while let Some(result) = slow_calls.join_next().await {
+            assert!(matches!(result, Err(error) if error.is_cancelled()));
+        }
+    }
+    release.add_permits(CONCURRENT_CALLS);
+    while let Some(result) = slow_calls.join_next().await {
+        let response = result.unwrap().unwrap();
+        assert_eq!(response.headers()["x-http-plugin"], "active");
+    }
+    // 取消确认到达前仍持有在途名额，探测只允许遇到明确的暂时过载
+    let response = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let result = plan
+                .handle_http(
+                    context(),
+                    plugin_request(),
+                    compose(Vec::new(), |request: core::Request| {
+                        Box::pin(async move {
+                            assert_eq!(request.uri(), "/rewritten?encoded=%2F");
+                            Ok(core::Response::new(core::empty_body()))
+                        })
+                    }),
+                )
+                .await;
+            match result {
+                Ok(response) => break response,
+                Err(MiddlewareError::Gateway(error))
+                    if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable =>
+                {
+                    tokio::task::yield_now().await
+                }
+                Err(error) => panic!("unexpected error: {error}"),
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(response.headers()["x-http-plugin"], "active");
+    drop(response);
+    drop(plan);
+    drop(generation);
+    runtime.shutdown().await;
+    crate::generation::wait_until_empty(cache.path()).await;
+}

backend/crates/gateway-plugin/runtime/tests/generation/policy/http/mod.rs

@@ -1,5 +1,6 @@
 //! 验证真实插件 HTTP 中间件的请求改写、流式转换与短路响应
 
+mod capacity;
 mod dispatch;
 mod websocket;
 
@@ -27,6 +28,19 @@ async fn http_plugin_with_limits(
     PluginRuntime,
     ExtensionSetReference,
     gateway_core::engine::middleware::FrozenMiddlewarePlan,
+) {
+    http_plugin_with_policy(mode, limits, PluginFailurePolicy::Reject).await
+}
+
+async fn http_plugin_with_policy(
+    mode: &str,
+    limits: RpcLimits,
+    failure_policy: PluginFailurePolicy,
+) -> (
+    tempfile::TempDir,
+    PluginRuntime,
+    ExtensionSetReference,
+    gateway_core::engine::middleware::FrozenMiddlewarePlan,
 ) {
     let worker = std::fs::read(env!("CARGO_BIN_EXE_gateway-plugin-test-middleware")).unwrap();
     let package = crate::support::package_with_contributions(
@@ -42,12 +56,7 @@ async fn http_plugin_with_limits(
         vec![InstanceFixture {
             id: "http-plugin",
             configuration: serde_json::json!({"http":true,"mode":mode}),
-            bindings: vec![binding(
-                MIDDLEWARE_CONTRIBUTION,
-                "http",
-                0,
-                PluginFailurePolicy::Reject,
-            )],
+            bindings: vec![binding(MIDDLEWARE_CONTRIBUTION, "http", 0, failure_policy)],
         }],
         package,
         limits,

backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs

@@ -327,3 +327,79 @@ async fn sdk_waiting_next_obeys_request_deadline_and_client_cancellation() {
         super::super::wait_until_empty(cache.path()).await;
     }
 }
+
+#[tokio::test]
+async fn request_middleware_never_delegates_when_total_in_flight_capacity_is_full() {
+    for policy in [PluginFailurePolicy::Delegate, PluginFailurePolicy::Reject] {
+        let package = crate::support::package_with_contributions(
+            crate::support::worker(),
+            Contributions::from([crate::support::contribution(
+                Capability::Middleware,
+                vec![Stage::Request],
+                vec!["openai".into()],
+                vec!["openai".into()],
+            )]),
+        );
+        let (cache, runtime) = setup_package_with_limits(
+            vec![InstanceFixture {
+                id: "busy-middleware",
+                configuration: serde_json::json!({}),
+                bindings: vec![binding(MIDDLEWARE_CONTRIBUTION, "request", 0, policy)],
+            }],
+            package,
+            RpcLimits {
+                maximum_calls: 1,
+                ..Default::default()
+            },
+        )
+        .await;
+        let generation = prepare(&runtime).await;
+        let plan = runtime
+            .execution_registry()
+            .middleware(&generation)
+            .unwrap();
+        let (entered, waiting) = tokio::sync::oneshot::channel();
+        let first = tokio::spawn({
+            let plan = plan.clone();
+            async move {
+                plan.handle(
+                    middleware_context(ClientTransport::HttpJson),
+                    MiddlewareRequest::new(
+                        "openai",
+                        middleware_headers(),
+                        Bytes::from_static(b"{}"),
+                    ),
+                    gateway_core::middleware::compose(Vec::new(), move |_| {
+                        Box::pin(async move {
+                            entered.send(()).unwrap();
+                            std::future::pending().await
+                        })
+                    }),
+                )
+                .await
+            }
+        });
+        tokio::time::timeout(Duration::from_secs(3), waiting)
+            .await
+            .unwrap()
+            .unwrap();
+        let result = plan
+            .handle(
+                middleware_context(ClientTransport::HttpJson),
+                MiddlewareRequest::new("openai", middleware_headers(), Bytes::from_static(b"{}")),
+                gateway_core::middleware::compose(Vec::new(), |_| {
+                    Box::pin(async { panic!("RPC 过载不能跳过请求中间件") })
+                }),
+            )
+            .await;
+        assert!(
+            matches!(result, Err(MiddlewareError::Gateway(error)) if error.kind() == gateway_core::error::GatewayErrorKind::MiddlewareCapacityUnavailable)
+        );
+        first.abort();
+        assert!(matches!(first.await, Err(error) if error.is_cancelled()));
+        drop(plan);
+        drop(generation);
+        runtime.shutdown().await;
+        super::super::wait_until_empty(cache.path()).await;
+    }
+}

backend/crates/gateway-plugin/runtime/tests/rpc/admission.rs

@@ -0,0 +1,328 @@
+//! 验证执行排队、next 恢复、流式在途预算与取消回收
+
+use std::{
+    sync::{Arc, atomic::Ordering},
+    time::Duration,
+};
+
+use super::session::{Callbacks, LifecycleCallbacks, prepared_session};
+use futures::future::BoxFuture;
+use gateway_plugin_runtime::{CallbackHandler, RpcError, RpcLimits, RpcReply, RpcSession};
+use gateway_plugin_sdk::{CallContext, PluginFault, Stage, call::middleware::NEXT_METHOD};
+use serde_json::json;
+use tokio::sync::{Semaphore, mpsc};
+
+async fn limited<C: CallbackHandler + 'static>(
+    callbacks: Arc<C>,
+    limits: RpcLimits,
+) -> (tempfile::TempDir, Arc<RpcSession>) {
+    let (cache, prepared, handshake, processes) = prepared_session();
+    let session = RpcSession::start(prepared, handshake, limits, &processes, callbacks)
+        .await
+        .unwrap();
+    (cache, Arc::new(session))
+}
+
+fn limits() -> RpcLimits {
+    RpcLimits {
+        maximum_calls: 4,
+        maximum_executing_calls: 1,
+        maximum_queued_calls: 1,
+        maximum_queue_timeout: Duration::from_millis(200),
+        maximum_callbacks: 1,
+        ..Default::default()
+    }
+}
+
+async fn wait_begun(callbacks: &LifecycleCallbacks, count: usize) {
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while callbacks.begun.load(Ordering::Acquire) < count {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+}
+
+fn call(session: &RpcSession, method: &str) -> impl Future<Output = Result<RpcReply, RpcError>> {
+    session.call(
+        method,
+        session.context(Stage::Request, Duration::from_secs(2)),
+        json!({}),
+        Vec::new(),
+    )
+}
+
+#[tokio::test]
+async fn bounded_initial_queue_waits_and_releases_cancelled_or_expired_waiters() {
+    let callbacks = Arc::new(LifecycleCallbacks::default());
+    let (_cache, session) = limited(callbacks.clone(), limits()).await;
+    let active = tokio::spawn({
+        let session = session.clone();
+        async move { call(&session, "hang").await }
+    });
+    wait_begun(&callbacks, 1).await;
+
+    let mut queued = Box::pin(call(&session, "echo"));
+    assert!(futures::poll!(&mut queued).is_pending());
+    assert!(matches!(
+        call(&session, "echo").await,
+        Err(RpcError::Overloaded)
+    ));
+    assert_eq!(callbacks.begun.load(Ordering::Acquire), 1);
+    drop(queued);
+
+    let expired = session
+        .call(
+            "echo",
+            session.context(Stage::Request, Duration::from_millis(30)),
+            json!({}),
+            Vec::new(),
+        )
+        .await;
+    assert!(matches!(expired, Err(RpcError::Overloaded)));
+    assert_eq!(callbacks.begun.load(Ordering::Acquire), 1);
+
+    let mut queued = Box::pin(call(&session, "echo"));
+    assert!(futures::poll!(&mut queued).is_pending());
+    active.abort();
+    assert!(matches!(active.await, Err(error) if error.is_cancelled()));
+    assert!(queued.await.is_ok());
+    assert_eq!(callbacks.begun.load(Ordering::Acquire), 2);
+    assert!(session.is_ready());
+    session.shutdown(Duration::from_secs(1)).await;
+}
+
+struct NextCallbacks {
+    entered: mpsc::UnboundedSender<()>,
+    release: Arc<Semaphore>,
+}
+
+impl CallbackHandler for NextCallbacks {
+    fn call(
+        &self,
+        _context: CallContext,
+        method: String,
+        params: serde_json::Value,
+        payload: Vec<u8>,
+    ) -> BoxFuture<'static, Result<RpcReply, PluginFault>> {
+        let release = self.release.clone();
+        self.entered.send(()).unwrap();
+        Box::pin(async move {
+            if method == NEXT_METHOD {
+                release.acquire().await.unwrap().forget();
+            }
+            Ok(RpcReply {
+                result: params,
+                payload,
+            })
+        })
+    }
+}
+
+#[tokio::test]
+async fn next_resumption_does_not_need_a_vacant_initial_queue_slot() {
+    let (entered, mut started) = mpsc::unbounded_channel();
+    let release = Arc::new(Semaphore::new(0));
+    let (_cache, session) = limited(
+        Arc::new(NextCallbacks {
+            entered,
+            release: release.clone(),
+        }),
+        limits(),
+    )
+    .await;
+    let waiting = tokio::spawn({
+        let session = session.clone();
+        async move {
+            session
+                .call(
+                    "callback_method",
+                    session.context(Stage::Request, Duration::from_secs(2)),
+                    json!({"method": NEXT_METHOD}),
+                    Vec::new(),
+                )
+                .await
+        }
+    });
+    started.recv().await.unwrap();
+    // 等待 next 不挤占唯一的普通回调额度
+    let mut occupying = Box::pin(call(&session, "callback"));
+    assert!(futures::poll!(&mut occupying).is_pending());
+    started.recv().await.unwrap();
+    occupying.await.unwrap();
+    let mut active = Box::pin(call(&session, "hang"));
+    assert!(futures::poll!(&mut active).is_pending());
+    let mut queued = Box::pin(call(&session, "echo"));
+    assert!(futures::poll!(&mut queued).is_pending());
+    assert!(matches!(
+        call(&session, "echo").await,
+        Err(RpcError::Overloaded)
+    ));
+    release.add_permits(1);
+    // 恢复等待不能立即因初次队列已满而拒绝，也不能绕过执行名额完成
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    assert!(!waiting.is_finished());
+    drop(active);
+    assert!(queued.await.is_ok());
+    assert!(waiting.await.unwrap().is_ok());
+    assert!(session.is_ready());
+    session.shutdown(Duration::from_secs(1)).await;
+}
+
+#[tokio::test]
+async fn response_stream_releases_execution_but_retains_total_capacity_until_drop() {
+    let callbacks = Arc::new(Callbacks::default());
+    let (_cache, session) = limited(
+        callbacks,
+        RpcLimits {
+            maximum_calls: 2,
+            ..limits()
+        },
+    )
+    .await;
+    let mut streams = Vec::new();
+    for _ in 0..2 {
+        streams.push(
+            session
+                .call_stream(
+                    "stream",
+                    session.context(Stage::Request, Duration::from_secs(5)),
+                    json!({}),
+                    Vec::new(),
+                )
+                .await
+                .unwrap(),
+        );
+    }
+    assert!(matches!(
+        call(&session, "echo").await,
+        Err(RpcError::Overloaded)
+    ));
+    drop(streams.pop());
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            match call(&session, "echo").await {
+                Ok(_) => break,
+                Err(RpcError::Overloaded) => tokio::task::yield_now().await,
+                Err(error) => panic!("unexpected error: {error}"),
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert!(session.is_ready());
+    drop(streams);
+    session.shutdown(Duration::from_secs(1)).await;
+}
+
+#[tokio::test]
+async fn queued_resumption_is_bounded_and_cancellable_without_failing_session() {
+    for cancelled in [false, true] {
+        let (entered, mut started) = mpsc::unbounded_channel();
+        let release = Arc::new(Semaphore::new(0));
+        let (_cache, session) = limited(
+            Arc::new(NextCallbacks {
+                entered,
+                release: release.clone(),
+            }),
+            limits(),
+        )
+        .await;
+        let waiting = tokio::spawn({
+            let session = session.clone();
+            async move {
+                session
+                    .call(
+                        "callback_method",
+                        session.context(Stage::Request, Duration::from_millis(200)),
+                        json!({"method": NEXT_METHOD}),
+                        Vec::new(),
+                    )
+                    .await
+            }
+        });
+        started.recv().await.unwrap();
+        let mut active = Box::pin(call(&session, "hang"));
+        assert!(futures::poll!(&mut active).is_pending());
+        release.add_permits(1);
+        if cancelled {
+            tokio::time::sleep(Duration::from_millis(30)).await;
+            assert!(!waiting.is_finished());
+            waiting.abort();
+            assert!(matches!(waiting.await, Err(error) if error.is_cancelled()));
+        } else {
+            assert!(matches!(waiting.await.unwrap(), Err(RpcError::Overloaded)));
+        }
+        drop(active);
+        assert!(call(&session, "echo").await.is_ok());
+        assert!(session.is_ready());
+        session.shutdown(Duration::from_secs(1)).await;
+    }
+}
+
+#[tokio::test]
+async fn initial_queue_wait_does_not_consume_the_plugin_processing_budget() {
+    let (_cache, session) = limited(
+        Arc::new(Callbacks::default()),
+        RpcLimits {
+            maximum_queue_timeout: Duration::from_secs(2),
+            ..limits()
+        },
+    )
+    .await;
+    let mut active = Box::pin(call(&session, "hang"));
+    assert!(futures::poll!(&mut active).is_pending());
+    let mut queued = Box::pin(session.call(
+        "echo",
+        session.context(Stage::Request, Duration::from_millis(200)),
+        json!({}),
+        Vec::new(),
+    ));
+    assert!(futures::poll!(&mut queued).is_pending());
+    tokio::time::sleep(Duration::from_millis(350)).await;
+    drop(active);
+    assert!(queued.await.is_ok());
+    assert!(session.is_ready());
+    session.shutdown(Duration::from_secs(1)).await;
+}
+
+#[tokio::test]
+async fn resumption_queue_wait_preserves_the_remaining_plugin_processing_budget() {
+    let (entered, mut started) = mpsc::unbounded_channel();
+    let release = Arc::new(Semaphore::new(0));
+    let (_cache, session) = limited(
+        Arc::new(NextCallbacks {
+            entered,
+            release: release.clone(),
+        }),
+        RpcLimits {
+            maximum_queue_timeout: Duration::from_secs(2),
+            ..limits()
+        },
+    )
+    .await;
+    let waiting = tokio::spawn({
+        let session = session.clone();
+        async move {
+            session
+                .call(
+                    "callback_method",
+                    session.context(Stage::Request, Duration::from_millis(200)),
+                    json!({"method": NEXT_METHOD}),
+                    Vec::new(),
+                )
+                .await
+        }
+    });
+    started.recv().await.unwrap();
+    let mut active = Box::pin(call(&session, "hang"));
+    assert!(futures::poll!(&mut active).is_pending());
+    release.add_permits(1);
+    tokio::time::sleep(Duration::from_millis(350)).await;
+    assert!(!waiting.is_finished());
+    drop(active);
+    assert!(waiting.await.unwrap().is_ok());
+    assert!(session.is_ready());
+    session.shutdown(Duration::from_secs(1)).await;
+}

backend/crates/gateway-plugin/runtime/tests/rpc/mod.rs

@@ -1,4 +1,5 @@
 //! 插件双向 RPC 分派与会话生命周期的测试入口
 
+mod admission;
 mod dispatch;
 mod session;

backend/crates/gateway-plugin/runtime/tests/rpc/session.rs

@@ -17,7 +17,7 @@ use serde_json::json;
 use tokio::sync::{Barrier, Notify};
 
 #[derive(Default)]
-struct Callbacks {
+pub(super) struct Callbacks {
     called: AtomicUsize,
     nested: Mutex<Option<Weak<RpcSession>>>,
 }
@@ -29,8 +29,8 @@ struct BlockingCallbacks {
 }
 
 #[derive(Default)]
-struct LifecycleCallbacks {
-    begun: AtomicUsize,
+pub(super) struct LifecycleCallbacks {
+    pub(super) begun: AtomicUsize,
     finished: AtomicUsize,
 }
 
@@ -193,7 +193,7 @@ where
     (cache, Arc::new(session))
 }
 
-fn prepared_session() -> (
+pub(super) fn prepared_session() -> (
     tempfile::TempDir,
     Arc<gateway_plugin_runtime::PreparedPackage>,
     Handshake,

backend/crates/gateway-plugin/sdk/README.md

@@ -52,6 +52,12 @@ use gateway_plugin_sdk::client::{read_frame, write_frame};
 `PluginSession::accept/run` 和 `PluginHandler`。组合插件从作者清单建立类型化方法，SDK 自动生成
 `plugin.register`；作者不需要复制贡献项或手写方法名、阶段和敏感载荷编解码。
 会话管理握手、有界并发、父调用关联、回调、流信用、取消、排空与关闭。
+
+`SessionConfig.maximum_calls` 默认 256，限制包含下游等待和响应流在内的在途调用；`maximum_callbacks` 默认 64，
+限制普通宿主回调。`next` 使用独立回调预算，上限为 `maximum_calls`，不挤占普通回调容量。
+宿主负责执行与准入调度，等待与处理独立计时；默认预算与过载行为见[插件使用](../../../../docs/plugins.md#故障处理)。
+插件显式覆盖配置时需与宿主协调在途预算；较小的插件侧上限会提前拒绝新调用
+
 `ResponseStream::channel` 提供有界流生产入口，`ResponseStream::pull` 按消费进度驱动生产者；`PluginCall` 的宿主客户端只代表当前父调用，
 不能把一个调用的资源误用到另一个调用。插件程序自行选择运行时和启动入口，不会启动网关。
 `SessionConfig.maximum_stream_chunk_bytes` 默认 1 MiB，只约束单个业务流分块；

backend/crates/gateway-plugin/sdk/docs/capabilities.md

@@ -243,7 +243,9 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限
 `next.run(input)` 消费自身，最多执行一次；结果按相反顺序经过外层。重试或独立子请求通过宿主执行服务发起
 
 调用上下文的 `timeout_ms` 限制插件自身处理时间，默认上限为 120 秒。等待宿主 `next` 时暂停这份预算，
-收到结果或错误后恢复剩余时间；入站处理与返回处理合计使用同一预算，普通宿主回调不会暂停计时。
+宿主等待期间归还执行名额，返回后先重新取得名额再交付结果或错误，插件继续同一次调用。
+初次准入和恢复排队各有独立等待期限，默认 30 秒，不扣插件处理预算；恢复等待受总在途上限约束，不占初次准入队列。
+入站处理与返回处理合计使用同一预算，普通宿主回调不会暂停计时。
 请求总期限、下游操作超时和取消信号独立生效，客户端取消或插件退出仍会结束等待并回收资源。
 宿主与插件 SDK 必须同时支持此语义；仍使用固定绝对期限的 SDK 构建的插件需重新构建，单独升级宿主不能解除插件侧期限
 

backend/crates/gateway-plugin/sdk/src/client/session/config.rs

@@ -25,8 +25,8 @@ impl Default for SessionConfig {
         Self {
             // 流分块还需匹配宿主授予的信用窗口
             maximum_stream_chunk_bytes: 1024 * 1024,
-            maximum_calls: 32,
-            maximum_callbacks: 32,
+            maximum_calls: 256,
+            maximum_callbacks: 64,
             maximum_buffered_stream_chunks: 1_024,
             handshake_timeout: Duration::from_secs(5),
             maximum_call_timeout: Duration::from_secs(120),

backend/crates/gateway-plugin/sdk/src/client/session/driver.rs

@@ -91,6 +91,7 @@ where
             .config
             .maximum_calls
             .checked_add(self.config.maximum_callbacks)
+            .and_then(|capacity| capacity.checked_add(self.config.maximum_calls))
             .ok_or(SessionError::Configuration)?;
         let control_reserve = self
             .config
@@ -105,7 +106,10 @@ where
             sender: outbound,
             data_slots: Arc::new(Semaphore::new(data_capacity)),
         };
-        let callbacks = Arc::new(CallbackRegistry::new(self.config.maximum_callbacks));
+        let callbacks = Arc::new(CallbackRegistry::new(
+            self.config.maximum_callbacks,
+            self.config.maximum_calls,
+        ));
         let handler: Arc<dyn PluginHandler> = Arc::new(handler);
         let lifecycle = LifecycleHooks::new(Arc::clone(&handler), control_reserve)?;
         write_frame(

backend/crates/gateway-plugin/sdk/src/client/session/registry.rs

@@ -67,7 +67,7 @@ impl HostClient {
             ordered = self.callbacks.order.lock() => ordered,
         };
         let managed_next = method == NEXT_METHOD;
-        let (id, received) = self.callbacks.begin(self.parent_id)?;
+        let (id, received) = self.callbacks.begin(self.parent_id, managed_next)?;
         let frame = Frame {
             message: Message::Callback {
                 id,
@@ -114,6 +114,7 @@ pub(super) struct CallbackRegistry {
     state: Mutex<CallbackState>,
     order: tokio::sync::Mutex<()>,
     maximum: usize,
+    maximum_next: usize,
     maximum_retired: usize,
 }
 
@@ -126,12 +127,13 @@ struct CallbackState {
 
 struct PendingCallback {
     parent_id: u64,
+    managed_next: bool,
     pause: Option<CallDeadlinePause>,
     result: oneshot::Sender<Result<HostReply, SessionError>>,
 }
 
 impl CallbackRegistry {
-    pub(super) fn new(maximum: usize) -> Self {
+    pub(super) fn new(maximum: usize, maximum_next: usize) -> Self {
         Self {
             state: Mutex::new(CallbackState {
                 next_id: 2,
@@ -141,19 +143,32 @@ impl CallbackRegistry {
             }),
             order: tokio::sync::Mutex::new(()),
             maximum,
-            maximum_retired: maximum.saturating_mul(2),
+            maximum_next,
+            maximum_retired: (maximum + maximum_next).saturating_mul(2),
         }
     }
 
     fn begin(
         &self,
         parent_id: u64,
+        managed_next: bool,
     ) -> Result<(u64, oneshot::Receiver<Result<HostReply, SessionError>>), SessionError> {
         let mut state = self
             .state
             .lock()
             .unwrap_or_else(std::sync::PoisonError::into_inner);
-        if state.pending.len() >= self.maximum {
+        let maximum = if managed_next {
+            self.maximum_next
+        } else {
+            self.maximum
+        };
+        if state
+            .pending
+            .values()
+            .filter(|pending| pending.managed_next == managed_next)
+            .count()
+            >= maximum
+        {
             return Err(SessionError::Capacity);
         }
         let id = state.next_id;
@@ -163,6 +178,7 @@ impl CallbackRegistry {
             id,
             PendingCallback {
                 parent_id,
+                managed_next,
                 result,
                 pause: None,
             },

backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs

@@ -140,10 +140,15 @@ impl PluginHandler for TestHandler {
                     tokio::time::sleep(Duration::from_millis(80)).await;
                     Ok(CallReply::unary(call.params, call.payload))
                 }
-                "callback" => {
+                "callback" | "next_callback" => {
+                    let method = if call.method == "next_callback" {
+                        NEXT_METHOD
+                    } else {
+                        "host.log"
+                    };
                     let reply = call
                         .host
-                        .call("host.log", call.params, call.payload)
+                        .call(method, call.params, call.payload)
                         .await
                         .map_err(SessionError::into_plugin_fault)?;
                     Ok(CallReply::unary(reply.result, reply.payload))
@@ -248,23 +253,31 @@ pub(super) async fn start_session<H: PluginHandler>(
 async fn start_session_with_capacity<H: PluginHandler>(
     handler: H,
     capacity: usize,
+) -> (HostPeer, JoinHandle<Result<(), SessionError>>) {
+    start_configured_session(
+        handler,
+        capacity,
+        SessionConfig {
+            maximum_stream_chunk_bytes: MAXIMUM_STREAM_CHUNK_BYTES,
+            maximum_calls: 4,
+            maximum_callbacks: 4,
+            maximum_buffered_stream_chunks: 16,
+            handshake_timeout: Duration::from_secs(1),
+            maximum_call_timeout: Duration::from_secs(2),
+        },
+    )
+    .await
+}
+
+async fn start_configured_session<H: PluginHandler>(
+    handler: H,
+    capacity: usize,
+    config: SessionConfig,
 ) -> (HostPeer, JoinHandle<Result<(), SessionError>>) {
     let (host, plugin) = tokio::io::duplex(capacity);
     let (plugin_reader, plugin_writer) = tokio::io::split(plugin);
     let task = tokio::spawn(async move {
-        let session = PluginSession::accept(
-            plugin_reader,
-            plugin_writer,
-            SessionConfig {
-                maximum_stream_chunk_bytes: MAXIMUM_STREAM_CHUNK_BYTES,
-                maximum_calls: 4,
-                maximum_callbacks: 4,
-                maximum_buffered_stream_chunks: 16,
-                handshake_timeout: Duration::from_secs(1),
-                maximum_call_timeout: Duration::from_secs(2),
-            },
-        )
-        .await?;
+        let session = PluginSession::accept(plugin_reader, plugin_writer, config).await?;
         session.run(handler).await
     });
     let (mut reader, mut writer) = tokio::io::split(host);

backend/crates/gateway-plugin/sdk/tests/client/session/registry.rs

@@ -297,3 +297,57 @@ async fn typed_key_and_quota_calls_keep_payloads_and_do_not_retry_failures() {
         }
     }
 }
+
+#[tokio::test]
+async fn managed_next_does_not_consume_the_bounded_ordinary_callback_budget() {
+    let (mut host, task) = start_configured_session(
+        TestHandler::default(),
+        MAXIMUM_STREAM_CHUNK_BYTES * 2,
+        SessionConfig {
+            maximum_calls: 4,
+            maximum_callbacks: 1,
+            ..Default::default()
+        },
+    )
+    .await;
+    send_call(&mut host, 1, "next_callback", json!({}), Vec::new()).await;
+    let Message::Callback {
+        id: next,
+        parent_id: 1,
+        method,
+        ..
+    } = receive(&mut host).await.message
+    else {
+        panic!("expected next callback")
+    };
+    assert_eq!(method, NEXT_METHOD);
+    send_call(&mut host, 3, "callback", json!({}), Vec::new()).await;
+    let Message::Callback {
+        id: ordinary,
+        parent_id: 3,
+        method,
+        ..
+    } = receive(&mut host).await.message
+    else {
+        panic!("next must not occupy ordinary callback capacity")
+    };
+    assert_eq!(method, "host.log");
+    send_call(&mut host, 5, "callback", json!({}), Vec::new()).await;
+    assert!(
+        matches!(receive(&mut host).await.message, Message::Error { id: 5, error } if error.code == ErrorCode::Capacity)
+    );
+    for (callback, parent) in [(ordinary, 3), (next, 1)] {
+        send_control(
+            &mut host,
+            Message::Result {
+                id: callback,
+                result: json!({}),
+            },
+        )
+        .await;
+        assert!(
+            matches!(receive(&mut host).await.message, Message::Result { id, .. } if id == parent)
+        );
+    }
+    shutdown(&mut host, task).await;
+}

docs/architecture.md

@@ -197,7 +197,9 @@ flowchart LR
 #### 调用资源与状态
 
 - **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。受管流建立后不沿用握手的绝对期限，网络操作与模型执行各自计时。
-  Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停，返回后恢复剩余预算；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。
+  Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停；Runtime 同时归还执行名额，返回后取得名额再以剩余处理预算恢复原调用。
+  执行调度由 Runtime 统一负责，准入和恢复等待使用独立期限；初次排队、总在途上下文和普通回调各有上限；`next` 回调按在途上限独立计数。
+  流、取消确认与消费方共同持有同一在途租约，执行名额只覆盖首结果前的处理；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。
   显式 Key 的身份快照随调用保留，每个模型执行拥有独立子调用图与结算；省略 Key 时继承现有模型父请求。页面 Responses 桥复核精确实例目标和 revision
   `frontend_authentication` 使用 principal 到 Key 的映射；公开登录回调的一次性票据用于关联流程，不限制插件回调能力。接口合同见 [SDK 回调](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#宿主资源回调)
 - **观察合同统一**：`observer` 通过一个类型化 RPC 入口接收完成与实际上游 WebSocket 事件，发布时按事件类型编译独立作用域。完成事件直接携带终态、用量、费用和耗时，不按能力声明裁剪字段；异步投递的并发、队列和载荷有界，观察失败不改变业务结果

docs/plugins.md

@@ -124,6 +124,14 @@ flowchart LR
 上游 429、用户请求错误和插件主动拒绝请求等业务结果不会直接停止插件。其他插件、原生转发和管理端继续运行，
 不因单个插件故障撤销整个宿主的运行配置。命中故障插件绑定的请求仍遵循既有拒绝或委托策略，入口认证插件失败不会绕过认证
 
+宿主每个插件会话默认允许 64 个执行中的 RPC 调用、64 个初次排队调用和 256 个总在途上下文。
+等待 `next` 时释放执行名额，返回后重新取得名额并继续原调用；已返回响应头的流继续占用在途预算，不占首结果执行名额。
+初次排队和恢复等待各限 30 秒，不扣插件自身默认上限 120 秒的处理预算。达到容量或等待上限时，中间件明确返回过载错误，HTTP 响应为 503（`middleware_capacity_unavailable`），不跳过插件、不重放请求。
+这些数值是有界资源预算的初值，实际容量需结合插件耗时、正文和流的资源占用评估
+
+普通宿主回调默认上限 64，`next` 使用总在途上限对应的独立回调预算。实际容量同时受插件 SDK 配置约束；
+使用较小上限或未区分 `next` 回调预算的插件需由作者调整并重新构建，升级宿主不会改变已安装插件的容量
+
 ### 停用与删除
 
 | 操作 | 删除什么 | 保留什么 |