fix(plugins): honor configured model and HTTP request body limits (#496)

* fix(plugins): honor configured request body limits

* test(plugin-sdk): mirror upstream adapter module layout

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

* fix(plugins): preserve frozen request body budgets across callbacks

Carry the execution body budget through policy and attempt callbacks, and validate explicit-key model requests against their bound snapshot. Add regression coverage for configuration publication and document budget ownership and lifetime.

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

---------

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

backend/apps/gateway/src/bootstrap/command.rs

@@ -96,6 +96,10 @@ async fn execute(
         Some(plugin_runtime.policy_registry()),
         Some(plugin_runtime.execution_registry()),
     );
+    or_shutdown!(
+        plugin_runtime,
+        plugin_runtime.bind_snapshots(&core.snapshots())
+    );
     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);

backend/apps/gateway/src/bootstrap/server.rs

@@ -31,6 +31,10 @@ pub async fn run() -> Result<(), BootstrapError> {
         Some(plugin_runtime.execution_registry()),
         Some(plugin_runtime.frontend_authentication_registry()),
     );
+    or_shutdown!(
+        plugin_runtime,
+        plugin_runtime.bind_snapshots(&core.snapshots())
+    );
     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);

backend/crates/gateway-core/src/engine/coordinator.rs

@@ -1132,6 +1132,7 @@ where
                 .with_pricing(self.plan.pricing())
                 .with_request_location(self.plan.request_location().cloned())
                 .with_privacy(self.plan.privacy())
+                .with_request_body_limit(self.plan.request_body_limit())
                 .with_concurrency_wait_budget(self.concurrency_wait_budget.clone())
                 .with_connection_budget(if connection_rotation_reserved {
                     self.connection_budget.clone().reserve_next_opening()

backend/crates/gateway-core/src/engine/execution/scope.rs

@@ -197,6 +197,15 @@ pub struct BoundModelExecutionContext {
 }
 
 impl BoundModelExecutionContext {
+    /// 与实际执行共用绑定时的快照，不能回退到后续发布的配置
+    #[must_use]
+    pub fn request_body_limit(&self) -> usize {
+        self.authority
+            .client
+            .snapshot
+            .responses_max_decompressed_body_bytes()
+    }
+
     #[must_use]
     pub fn request_settings(&self) -> Option<&RequestSettings> {
         self.authority.client.request_settings()

backend/crates/gateway-core/src/engine/execution/service.rs

@@ -448,6 +448,12 @@ impl DefaultExecutionService {
                     account_group_ids.clone(),
                 )
                 .with_extension_scope(authorization.extension_scope.clone())
+                .with_request_body_limit(
+                    request
+                        .client
+                        .snapshot
+                        .responses_max_decompressed_body_bytes(),
+                )
                 .with_execution_effects(Arc::clone(&authorization.graph.as_ref()?.effects)),
             )
         });

backend/crates/gateway-core/src/engine/middleware/contract.rs

@@ -559,9 +559,22 @@ pub struct MiddlewareContext {
     deadline: crate::lifecycle::Deadline,
     extension_scope: ExtensionCallScope,
     execution_effects: Option<Arc<ExecutionEffects>>,
+    request_body_limit: Option<usize>,
 }
 
 impl MiddlewareContext {
+    /// attempt 只继承执行预算，不开放 request 设置改写
+    #[must_use]
+    pub const fn with_request_body_limit(mut self, bytes: Option<usize>) -> Self {
+        self.request_body_limit = bytes;
+        self
+    }
+
+    #[must_use]
+    pub const fn request_body_limit(&self) -> Option<usize> {
+        self.request_body_limit
+    }
+
     #[must_use]
     pub fn plan(&self) -> Option<FrozenMiddlewarePlan> {
         self.plan.clone()
@@ -590,6 +603,7 @@ impl MiddlewareContext {
         } = authority;
         Self {
             plan: None,
+            request_body_limit: None,
             request_id,
             mount,
             attempt_index,

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

@@ -286,6 +286,7 @@ impl ContinuationAttempt {
 /// Provider 每次执行可见的 request-local context
 #[derive(Debug, Clone)]
 pub struct RequestAttemptContext {
+    request_body_limit: Option<usize>,
     lease_metadata: Option<Arc<crate::provider_ports::ProviderLeaseMetadata>>,
     response_control: Option<response_control::ResponseControl>,
     pricing: Arc<crate::metering::PricingOverrides>,
@@ -311,6 +312,12 @@ pub struct RequestAttemptContext {
 }
 
 impl RequestAttemptContext {
+    #[must_use]
+    pub const fn with_request_body_limit(mut self, bytes: usize) -> Self {
+        self.request_body_limit = Some(bytes);
+        self
+    }
+
     #[must_use]
     pub fn with_lease_metadata(
         mut self,
@@ -387,6 +394,7 @@ impl RequestAttemptContext {
     pub fn new(request_id: ModelRequestId, client_api_key_ref: ClientApiKeyId) -> Self {
         Self {
             response_control: None,
+            request_body_limit: None,
             lease_metadata: None,
             request_id,
             client_api_key_ref,
@@ -542,6 +550,11 @@ pub struct AttemptContext {
 }
 
 impl AttemptContext {
+    #[must_use]
+    pub const fn request_body_limit(&self) -> Option<usize> {
+        self.request.request_body_limit
+    }
+
     #[must_use]
     pub fn lease_metadata(&self) -> Option<Arc<crate::provider_ports::ProviderLeaseMetadata>> {
         self.request.lease_metadata.clone()
@@ -747,7 +760,8 @@ impl AttemptContext {
                 extension_scope: self.request.extension_scope.clone(),
                 execution_effects: self.request.execution_effects.as_ref().map(Arc::clone),
             },
-        );
+        )
+        .with_request_body_limit(self.request.request_body_limit);
         provider::execute_attempt_middleware(
             self.request.middleware.as_ref(),
             context,

backend/crates/gateway-core/src/engine/policy.rs

@@ -34,9 +34,15 @@ pub struct ModelRouteInput {
     requested_model: PublicModelId,
     available_providers: BTreeSet<ProviderKind>,
     execution_effects: Arc<ExecutionEffects>,
+    request_body_limit: Option<usize>,
 }
 
 impl ModelRouteInput {
+    #[must_use]
+    pub const fn request_body_limit(&self) -> Option<usize> {
+        self.request_body_limit
+    }
+
     #[must_use]
     pub const fn request_id(&self) -> &super::ModelRequestId {
         &self.request_id
@@ -188,9 +194,15 @@ pub struct AccountScheduleInput {
     model: Option<String>,
     candidates: Vec<AccountScheduleCandidate>,
     execution_effects: Arc<ExecutionEffects>,
+    request_body_limit: Option<usize>,
 }
 
 impl AccountScheduleInput {
+    #[must_use]
+    pub const fn request_body_limit(&self) -> Option<usize> {
+        self.request_body_limit
+    }
+
     #[must_use]
     pub const fn request_id(&self) -> &super::ModelRequestId {
         &self.request_id
@@ -282,9 +294,15 @@ pub struct RetryInput {
     pub extension_scope: ExtensionCallScope,
     pub facts: RetryFacts,
     execution_effects: Arc<ExecutionEffects>,
+    request_body_limit: Option<usize>,
 }
 
 impl RetryInput {
+    #[must_use]
+    pub const fn request_body_limit(&self) -> Option<usize> {
+        self.request_body_limit
+    }
+
     #[must_use]
     pub fn execution_effects(&self) -> Arc<ExecutionEffects> {
         Arc::clone(&self.execution_effects)
@@ -339,6 +357,7 @@ pub struct RequestPolicyContext {
     account_group_ids: Arc<[AccountGroupId]>,
     extension_scope: ExtensionCallScope,
     execution_effects: Arc<ExecutionEffects>,
+    request_body_limit: Option<usize>,
 }
 
 impl fmt::Debug for RequestPolicyContext {
@@ -353,6 +372,13 @@ impl fmt::Debug for RequestPolicyContext {
 }
 
 impl RequestPolicyContext {
+    /// 路由、调度和重试沿用同一次模型执行的冻结预算
+    #[must_use]
+    pub const fn with_request_body_limit(mut self, bytes: usize) -> Self {
+        self.request_body_limit = Some(bytes);
+        self
+    }
+
     pub async fn retry_decision(&self, facts: RetryFacts) -> RetryDecision {
         self.plan
             .retry_decision(RetryInput {
@@ -362,6 +388,7 @@ impl RequestPolicyContext {
                 extension_scope: self.extension_scope.clone(),
                 facts,
                 execution_effects: Arc::clone(&self.execution_effects),
+                request_body_limit: self.request_body_limit,
             })
             .await
             .unwrap_or_else(|_| {
@@ -391,6 +418,7 @@ impl RequestPolicyContext {
             account_group_ids: account_group_ids.into(),
             extension_scope: ExtensionCallScope::default(),
             execution_effects: Arc::new(ExecutionEffects::default()),
+            request_body_limit: None,
         }
     }
 
@@ -428,6 +456,7 @@ impl RequestPolicyContext {
                 requested_model,
                 available_providers,
                 execution_effects: Arc::clone(&self.execution_effects),
+                request_body_limit: self.request_body_limit,
             })
             .await
     }
@@ -474,6 +503,7 @@ impl RequestPolicyContext {
                 model: model.map(str::to_owned),
                 candidates: projected,
                 execution_effects: Arc::clone(&self.execution_effects),
+                request_body_limit: self.request_body_limit,
             })
             .await
             .map_err(|_| AccountPolicyError::Fault)?;

backend/crates/gateway-core/src/routing/extensions.rs

@@ -88,6 +88,15 @@ pub struct ExtensionPreparationError;
 
 /// 准备阶段读取与请求快照相同的持久 revision；不得自行发布当前代次
 pub trait ExtensionPreparationPort: Send + Sync {
+    /// 初始发布尚无可读快照，注册阶段沿用本次编译的正文预算
+    fn prepare_with_request_body_limit(
+        &self,
+        revision: ConfigRevision,
+        _request_body_limit: usize,
+    ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
+        self.prepare(revision)
+    }
+
     fn prepare(
         &self,
         revision: ConfigRevision,

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

@@ -561,6 +561,7 @@ impl ProviderCandidate {
 /// 一次请求冻结的 Provider 尝试顺序
 #[derive(Debug, Clone)]
 pub struct RoutingPlan {
+    request_body_limit: usize,
     pricing: Arc<crate::metering::PricingOverrides>,
     request_location: Option<crate::account::RequestLocation>,
     privacy: Option<Arc<dyn crate::settings::privacy::CompiledPrivacyPolicy>>,
@@ -573,6 +574,11 @@ pub struct RoutingPlan {
 }
 
 impl RoutingPlan {
+    #[must_use]
+    pub const fn request_body_limit(&self) -> usize {
+        self.request_body_limit
+    }
+
     #[must_use]
     pub fn pricing(&self) -> Arc<crate::metering::PricingOverrides> {
         Arc::clone(&self.pricing)

backend/crates/gateway-core/src/routing/snapshot.rs

@@ -331,7 +331,13 @@ impl RuntimeSnapshotCompiler {
                             });
                     if must_prepare {
                         let prepared = preparer
-                            .prepare(revision)
+                            .prepare_with_request_body_limit(
+                                revision,
+                                usize::try_from(
+                                    facts.settings.responses_max_decompressed_body_bytes,
+                                )
+                                .unwrap_or(usize::MAX),
+                            )
                             .await
                             .map_err(|_| RuntimeSnapshotCompileError::ExtensionsUnavailable)?;
                         prepared_extensions = Some((revision, prepared));
@@ -1201,6 +1207,7 @@ impl RuntimeSnapshot {
         }
 
         Ok(RoutingPlan {
+            request_body_limit: self.responses_max_decompressed_body_bytes(),
             config_revision: self.revision,
             pricing: Arc::clone(&self.settings.values.pricing),
             request_location: self.settings.request_location.clone(),
@@ -1282,6 +1289,7 @@ impl RuntimeSnapshot {
             account_scope: Arc::clone(&account_scope),
         };
         Ok(RoutingPlan {
+            request_body_limit: self.responses_max_decompressed_body_bytes(),
             config_revision: self.revision,
             pricing: Arc::clone(&self.settings.values.pricing),
             request_location: self.settings.request_location.clone(),

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

@@ -3,7 +3,9 @@
 mod publications;
 mod publisher;
 
-pub use publications::{RuntimeSnapshotHandle, RuntimeSnapshotUnavailable};
+pub use publications::{
+    RuntimeSnapshotHandle, RuntimeSnapshotUnavailable, WeakRuntimeSnapshotHandle,
+};
 pub use publisher::{
     RuntimeSnapshotPublisher, SnapshotControl, SnapshotRevisionStream, SnapshotSubscriptionError,
     SnapshotSubscriptionPort, runtime_revision_needs_refresh,

backend/crates/gateway-core/src/runtime/publications.rs

@@ -8,7 +8,7 @@ use crate::{
 use futures::future::BoxFuture;
 use std::{
     collections::BTreeMap,
-    sync::{Arc, Mutex as SyncMutex, RwLock},
+    sync::{Arc, Mutex as SyncMutex, RwLock, Weak},
 };
 
 /// 请求级冻结失败；此状态必须 fail closed
@@ -23,7 +23,27 @@ pub struct RuntimeSnapshotHandle {
     publications: Arc<SyncMutex<Vec<futures::channel::mpsc::Sender<()>>>>,
 }
 
+/// 不延长发布集合生命周期的只读句柄，供集合持有的扩展读取后续配置
+#[derive(Clone, Default)]
+pub struct WeakRuntimeSnapshotHandle {
+    current: Weak<RwLock<Option<Arc<RuntimeSnapshot>>>>,
+}
+
+impl WeakRuntimeSnapshotHandle {
+    pub fn acquire(&self) -> Result<Arc<RuntimeSnapshot>, RuntimeSnapshotUnavailable> {
+        let current = self.current.upgrade().ok_or(RuntimeSnapshotUnavailable)?;
+        acquire_current(&current)
+    }
+}
+
 impl RuntimeSnapshotHandle {
+    #[must_use]
+    pub fn downgrade(&self) -> WeakRuntimeSnapshotHandle {
+        WeakRuntimeSnapshotHandle {
+            current: Arc::downgrade(&self.current),
+        }
+    }
+
     #[must_use]
     pub fn new(initial: RuntimeSnapshot) -> Self {
         Self {
@@ -86,10 +106,7 @@ impl RuntimeSnapshotHandle {
 
     /// 冻结当前 Arc；后续发布不改变已经开始的请求
     pub fn acquire(&self) -> Result<Arc<RuntimeSnapshot>, RuntimeSnapshotUnavailable> {
-        read_unpoisoned(&self.current)
-            .clone()
-            .filter(|snapshot| snapshot.extensions().is_none_or(|set| set.can_serve()))
-            .ok_or(RuntimeSnapshotUnavailable)
+        acquire_current(&self.current)
     }
 }
 
@@ -109,6 +126,15 @@ impl HealthProbe for RuntimeSnapshotHandle {
     }
 }
 
+fn acquire_current(
+    current: &RwLock<Option<Arc<RuntimeSnapshot>>>,
+) -> Result<Arc<RuntimeSnapshot>, RuntimeSnapshotUnavailable> {
+    read_unpoisoned(current)
+        .clone()
+        .filter(|snapshot| snapshot.extensions().is_none_or(|set| set.can_serve()))
+        .ok_or(RuntimeSnapshotUnavailable)
+}
+
 fn read_unpoisoned<T>(lock: &RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
     lock.read()
         .unwrap_or_else(std::sync::PoisonError::into_inner)

backend/crates/gateway-core/tests/routing/snapshot.rs

@@ -1253,7 +1253,8 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication()
         ));
     }
     let handle = RuntimeSnapshotHandle::new(compile(1, 64 * 1024 * 1024).unwrap());
-    let frozen = handle.acquire().unwrap();
+    let reader = handle.downgrade();
+    let frozen = reader.acquire().unwrap();
     handle.publish(compile(2, 128 * 1024 * 1024).unwrap());
     assert_eq!(
         handle
@@ -1266,6 +1267,13 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication()
         frozen.responses_max_decompressed_body_bytes(),
         64 * 1024 * 1024
     );
+    assert_eq!(
+        reader
+            .acquire()
+            .unwrap()
+            .responses_max_decompressed_body_bytes(),
+        128 * 1024 * 1024
+    );
     handle.publish(compile(3, 1024 * 1024).unwrap());
     assert_eq!(
         handle
@@ -1274,6 +1282,10 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication()
             .responses_max_decompressed_body_bytes(),
         1024 * 1024
     );
+    handle.suspend();
+    assert!(reader.acquire().is_err());
+    drop(handle);
+    assert!(reader.acquire().is_err());
 }
 
 #[test]

backend/crates/gateway-host/src/outbound/http.rs

@@ -40,6 +40,8 @@ pub struct HttpRequest {
     pub url: String,
     pub headers: Vec<(String, Vec<u8>)>,
     pub body: Vec<u8>,
+    /// 调用方从有效配置派生的请求体上限；零只允许空正文
+    pub maximum_body_bytes: usize,
 }
 
 pub struct HttpResponse {
@@ -191,7 +193,7 @@ impl HttpClient {
         scope: Option<&str>,
         websocket_key: Option<&str>,
     ) -> Result<PendingResponse, HttpError> {
-        if request.body.len() > 16 * 1024 * 1024
+        if request.body.len() > request.maximum_body_bytes
             || request.headers.len() > 128
             || timeout.is_zero()
             || timeout > Duration::from_secs(120)

backend/crates/gateway-host/src/plugin_distribution/http.rs

@@ -96,6 +96,7 @@ impl Downloads {
                         url: url.to_string(),
                         headers,
                         body: vec![],
+                        maximum_body_bytes: 0,
                     },
                     egress.map(|egress| &egress.proxy),
                     &self.network,

backend/crates/gateway-host/tests/outbound/connector.rs

@@ -16,6 +16,7 @@ fn request(port: u16, secure: bool) -> HttpRequest {
         ),
         headers: vec![],
         body: vec![],
+        maximum_body_bytes: 0,
     }
 }
 

backend/crates/gateway-host/tests/outbound/http.rs

@@ -12,6 +12,7 @@ fn request(url: String) -> HttpRequest {
         url,
         headers: vec![],
         body: vec![],
+        maximum_body_bytes: 0,
     }
 }
 
@@ -316,3 +317,40 @@ async fn cancelling_dns_releases_the_request_slot_without_starting_http() {
         );
     }
 }
+
+#[tokio::test]
+async fn managed_http_enforces_the_callers_body_limit_before_sending() {
+    let server = MockServer::start().await;
+    Mock::given(any())
+        .respond_with(ResponseTemplate::new(200))
+        .expect(2)
+        .mount(&server)
+        .await;
+    let client = HttpClient::new().unwrap();
+    // 覆盖旧固定上限之外的正文，并验证同一个客户端逐次使用调用方的预算
+    let maximum = 17 * 1024 * 1024;
+    for (size, limit, accepted) in [
+        (maximum, maximum, true),
+        (maximum + 1, maximum, false),
+        (maximum + 1, maximum + 1, true),
+    ] {
+        let mut request = request(server.uri());
+        request.method = "POST".into();
+        request.body = vec![b'x'; size];
+        request.maximum_body_bytes = limit;
+        let result = client
+            .open(request, None, &local_network(), Duration::from_secs(10))
+            .await;
+        if accepted {
+            assert_eq!(result.unwrap().status, 200);
+        } else {
+            let error = result.err().unwrap();
+            assert_eq!(error.kind(), HttpErrorKind::InvalidRequest);
+            assert_eq!(error.send_state, UpstreamSendState::NotSent);
+        }
+    }
+    let requests = server.received_requests().await.unwrap();
+    assert_eq!(requests.len(), 2);
+    assert_eq!(requests[0].body.len(), maximum);
+    assert_eq!(requests[1].body.len(), maximum + 1);
+}

backend/crates/gateway-host/tests/outbound/network.rs

@@ -98,6 +98,7 @@ fn request(url: String) -> HttpRequest {
         url,
         headers: vec![],
         body: vec![],
+        maximum_body_bytes: 0,
     }
 }
 

backend/crates/gateway-host/tests/outbound/websocket.rs

@@ -16,6 +16,7 @@ fn request(url: String) -> HttpRequest {
         url,
         headers: vec![],
         body: vec![],
+        maximum_body_bytes: 0,
     }
 }
 fn network() -> NetworkPolicy {

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

@@ -86,6 +86,7 @@ pub(super) async fn invoke(
             context.extension_scope().clone(),
             context.plan(),
             context.cancellation().clone(),
+            context.request_body_limit(),
         )
         .map_err(|_| MiddlewareError::Fault)?;
     let _binding = invocation_ports.callbacks.bind_middleware(

backend/crates/gateway-plugin/runtime/src/adapter/policy/retry.rs

@@ -91,6 +91,7 @@ async fn call(
             input.extension_scope.clone(),
             None,
             gateway_core::lifecycle::CancellationToken::new(),
+            input.request_body_limit(),
         )
         .map_err(|_| ())?;
     let reply = invocation

backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs

@@ -137,6 +137,7 @@ async fn route_with_entry(
             input.extension_scope().clone(),
             None,
             gateway_core::lifecycle::CancellationToken::new(),
+            input.request_body_limit(),
         )
         .map_err(|_| ())?;
     let reply = invocation
@@ -212,6 +213,7 @@ async fn schedule_with_entry(
             input.extension_scope().clone(),
             None,
             gateway_core::lifecycle::CancellationToken::new(),
+            input.request_body_limit(),
         )
         .map_err(|_| ())?;
     let reply = invocation

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

@@ -212,10 +212,21 @@ impl Execution {
             .extension_scope()
             .extending(self.adapter.instance_id.clone())
             .ok_or_else(invalid)?;
+        let request_body_limit = invocation
+            .context
+            .request_body_limit()
+            .map(Ok)
+            .unwrap_or_else(|| self.adapter.callbacks.request_body_limit())
+            .map_err(|error| fault_error(error, UpstreamSendState::NotSent))?;
         let scope = self
             .adapter
             .callbacks
-            .prepare_upstream(&context, Arc::clone(&managed), extension_scope)
+            .prepare_upstream(
+                &context,
+                Arc::clone(&managed),
+                extension_scope,
+                request_body_limit,
+            )
             .map_err(|error| fault_error(error, UpstreamSendState::NotSent))?;
         let request = UpstreamAdapterRequest {
             adapter_id: declaration.id.clone(),
@@ -239,14 +250,18 @@ impl Execution {
                 .collect(),
             continuation: previous.map(|state| state.continuation),
         };
+        let body = invocation
+            .operation
+            .middleware_body()
+            .map_err(|source| invalid().with_source(source))?;
+        if body.len() > gateway_plugin_sdk::call::model::request_payload_limit(request_body_limit) {
+            return Err(fault_error(
+                crate::callback::request_limits::too_large(),
+                UpstreamSendState::NotSent,
+            ));
+        }
         let payload = request
-            .encode(
-                invocation
-                    .operation
-                    .middleware_body()
-                    .map_err(|source| invalid().with_source(source))?
-                    .to_vec(),
-            )
+            .encode(body.to_vec())
             .map_err(|source| invalid().with_source(source))?;
         invocation.context.trace().record("plugin.upstream", serde_json::json!({ "instanceId": self.adapter.instance_id, "adapterId": declaration.id, "generation": context.generation, "transport": declaration.transport.as_str() }));
         let mut stream = tokio::select! {

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

@@ -17,9 +17,18 @@ pub(super) struct HttpCallbacks<'a> {
     pub(super) scope: &'a Arc<CallbackScope>,
     pub(super) call: &'a CallResources,
     pub(super) maximum_payload: usize,
+    pub(super) request_limits: &'a super::request_limits::RequestLimits,
 }
 
 impl HttpCallbacks<'_> {
+    pub(super) fn body_limit(&self, body: &[u8]) -> Result<usize, PluginFault> {
+        // 空正文不占请求预算，也允许尚未发布设置的注册阶段执行无正文查询
+        if body.is_empty() {
+            return Ok(0);
+        }
+        self.request_limits.body_bytes(self.call)
+    }
+
     pub(super) async fn send(
         &self,
         request: HttpRequest,
@@ -32,7 +41,11 @@ impl HttpCallbacks<'_> {
             scope,
             call,
             maximum_payload,
+            ..
         } = *self;
+        if request.body.len() > request.maximum_body_bytes {
+            return Err(super::request_limits::too_large());
+        }
         let timeout = call.timeout()?;
         let attempt = call.start_upstream(upstream_purpose)?;
         let response = client
@@ -116,6 +129,7 @@ impl HttpCallbacks<'_> {
                             .into_iter()
                             .map(|(name, value)| (name, value.into_bytes()))
                             .collect(),
+                        maximum_body_bytes: self.body_limit(&payload)?,
                         body: payload,
                     },
                     method == "host.http.do_stream",

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

@@ -15,6 +15,7 @@ mod log;
 mod middleware;
 mod model;
 pub(crate) mod private_state;
+pub(crate) mod request_limits;
 mod resources;
 mod scope;
 pub(crate) mod services;
@@ -80,6 +81,7 @@ impl PluginCallbackPorts {
 }
 
 pub(crate) struct PluginCallbacks {
+    pub(crate) registration_body_limit: Option<usize>,
     services: Arc<services::ServicePorts>,
     service_actor: gateway_admin::model::usage_data::UsageActor,
     accounts: Arc<accounts::PluginAccounts>,
@@ -96,6 +98,7 @@ pub(crate) struct PluginCallbacks {
     pending_middleware: Mutex<BTreeMap<String, Arc<dyn MiddlewareCallback>>>,
     calls: Mutex<BTreeMap<u64, Arc<CallResources>>>,
     maximum_payload: usize,
+    request_limits: Arc<request_limits::RequestLimits>,
 }
 
 impl PluginCallbacks {
@@ -106,8 +109,10 @@ impl PluginCallbacks {
         log_slots: Arc<tokio::sync::Semaphore>,
         private_state: Arc<private_state::PluginPrivateState>,
         ports: PluginCallbackPorts,
+        request_limits: Arc<request_limits::RequestLimits>,
     ) -> Result<Self, AdminError> {
         Ok(Self {
+            registration_body_limit: None,
             services: ports.services,
             service_actor: gateway_admin::model::usage_data::UsageActor::Plugin {
                 publisher: manifest.publisher.clone(),
@@ -126,7 +131,11 @@ impl PluginCallbacks {
                 log::PluginLog::new(manifest, log_slots)
                     .map_err(|_| AdminError::invalid("插件清单身份不合法"))?,
             ),
-            models: Arc::new(model::PluginModels::new(ports.models, maximum_payload)),
+            models: Arc::new(model::PluginModels::new(
+                ports.models,
+                request_limits.clone(),
+                maximum_payload,
+            )),
             private_state,
             http: ports.http,
             network: Arc::new(ports.network),
@@ -134,6 +143,7 @@ impl PluginCallbacks {
             pending_middleware: Mutex::new(BTreeMap::new()),
             calls: Mutex::new(BTreeMap::new()),
             maximum_payload,
+            request_limits,
         })
     }
 
@@ -181,6 +191,7 @@ impl PluginCallbacks {
         extensions: gateway_core::engine::extensions::ExtensionCallScope,
         plan: Option<gateway_core::engine::middleware::FrozenMiddlewarePlan>,
         cancellation: CancellationToken,
+        request_body_limit: Option<usize>,
     ) -> Result<Arc<CallbackScope>, PluginFault> {
         if !matches!(
             context.stage,
@@ -197,6 +208,7 @@ impl PluginCallbacks {
             scope = scope.with_execution_effects(effects);
         }
         scope.extension_scope = extensions;
+        scope.request_body_limit = request_body_limit;
         scope.origin = plan.map(|plan| InvocationContext {
             request_id: context
                 .request_id
@@ -212,11 +224,16 @@ impl PluginCallbacks {
         self.register_scope(context, scope)
     }
 
+    pub(crate) fn request_body_limit(&self) -> Result<usize, PluginFault> {
+        self.request_limits.current_body_bytes()
+    }
+
     pub(crate) fn prepare_upstream(
         &self,
         context: &CallContext,
         managed: Arc<upstream::ManagedUpstream>,
         extension_scope: gateway_core::engine::extensions::ExtensionCallScope,
+        request_body_limit: usize,
     ) -> Result<Arc<CallbackScope>, PluginFault> {
         if context.stage != gateway_plugin_sdk::Stage::Upstream
             || context.account_id.as_deref() != Some(managed.account.account_id().as_str())
@@ -227,6 +244,7 @@ impl PluginCallbacks {
         let mut scope = CallbackScope::new(context, managed.account.outbound_proxy().cloned())
             .with_upstream(managed);
         scope.extension_scope = extension_scope;
+        scope.request_body_limit = Some(request_body_limit);
         self.register_scope(context, scope)
     }
 
@@ -352,7 +370,13 @@ impl CallbackHandler for PluginCallbacks {
             .unwrap_or_else(std::sync::PoisonError::into_inner)
             .get(&context.resource_scope_id)
             .and_then(Weak::upgrade)
-            .unwrap_or_else(|| Arc::new(CallbackScope::for_call(context)));
+            .unwrap_or_else(|| {
+                let mut scope = CallbackScope::for_call(context);
+                if context.stage == gateway_plugin_sdk::Stage::Registration {
+                    scope.request_body_limit = self.registration_body_limit;
+                }
+                Arc::new(scope)
+            });
         let middleware = self
             .pending_middleware
             .lock()
@@ -401,6 +425,7 @@ impl CallbackHandler for PluginCallbacks {
         let data = self.data.clone();
         let keys = self.keys.clone();
         let models = self.models.clone();
+        let request_limits = self.request_limits.clone();
         let affinity = self.affinity.clone();
         let services = self.services.clone();
         let service_actor = self.service_actor.clone();
@@ -514,6 +539,7 @@ impl CallbackHandler for PluginCallbacks {
                 scope,
                 call: &call,
                 maximum_payload,
+                request_limits: &request_limits,
             };
             if method.starts_with("host.upstream.") {
                 return upstream::dispatch(&http, &method, params, payload).await;

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

@@ -63,13 +63,19 @@ impl PluginModelPortSlot {
 
 pub(super) struct PluginModels {
     slot: Arc<PluginModelPortSlot>,
+    request_limits: Arc<super::request_limits::RequestLimits>,
     maximum_payload: usize,
 }
 
 impl PluginModels {
-    pub(super) fn new(slot: Arc<PluginModelPortSlot>, maximum_payload: usize) -> Self {
+    pub(super) fn new(
+        slot: Arc<PluginModelPortSlot>,
+        request_limits: Arc<super::request_limits::RequestLimits>,
+        maximum_payload: usize,
+    ) -> Self {
         Self {
             slot,
+            request_limits,
             maximum_payload,
         }
     }
@@ -121,17 +127,27 @@ impl PluginModels {
             "host.model.execute" | "host.model.execute_stream" => {
                 let mut request: ModelExecuteRequest =
                     serde_json::from_value(params).map_err(|_| invalid())?;
-                if payload.is_empty() || payload.len() > self.maximum_payload {
+                if payload.is_empty() {
                     return Err(invalid());
                 }
+                let check_size = |limit| {
+                    if payload.len() > gateway_plugin_sdk::call::model::request_payload_limit(limit)
+                    {
+                        Err(super::request_limits::too_large())
+                    } else {
+                        Ok(())
+                    }
+                };
                 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 bound = self.binding(&port, context, call, client_key_id).await?;
+                    check_size(bound.request_body_limit())?;
                     port.start_bound(self.bound_request(request, payload, stream, bound)?)
                         .await
                         .map_err(super::error::gateway)?
                 } else {
+                    check_size(self.request_limits.body_bytes(call)?)?;
                     let request = self.request(context, request, payload, stream)?;
                     port.start(request).await.map_err(super::error::gateway)?
                 };

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

@@ -0,0 +1,49 @@
+//! 从当前请求或已发布设置读取正文预算，不持有发布集合的强引用
+
+use super::CallResources;
+use gateway_core::runtime::{RuntimeSnapshotHandle, WeakRuntimeSnapshotHandle};
+use gateway_plugin_sdk::{ErrorCode, PluginFault};
+use std::sync::OnceLock;
+
+#[derive(Default)]
+pub(crate) struct RequestLimits {
+    snapshots: OnceLock<WeakRuntimeSnapshotHandle>,
+}
+
+impl RequestLimits {
+    pub(crate) fn bind(
+        &self,
+        snapshots: &RuntimeSnapshotHandle,
+    ) -> Result<(), gateway_admin::model::AdminError> {
+        self.snapshots
+            .set(snapshots.downgrade())
+            .map_err(|_| gateway_admin::model::AdminError::conflict("插件设置读取端口已经绑定"))
+    }
+
+    pub(super) fn body_bytes(&self, call: &CallResources) -> Result<usize, PluginFault> {
+        if let Some(settings) = call.request_settings() {
+            return Ok(settings.snapshot().responses_max_decompressed_body_bytes());
+        }
+        if let Some(bytes) = call.scope.request_body_limit {
+            return Ok(bytes);
+        }
+        self.current_body_bytes()
+    }
+
+    pub(crate) fn current_body_bytes(&self) -> Result<usize, PluginFault> {
+        self.snapshots
+            .get()
+            .and_then(|snapshots| snapshots.acquire().ok())
+            .map(|snapshot| snapshot.responses_max_decompressed_body_bytes())
+            .ok_or_else(|| PluginFault::new(ErrorCode::Fault, "request body limit is unavailable"))
+    }
+}
+
+pub(crate) fn too_large() -> PluginFault {
+    let mut fault = PluginFault::new(
+        ErrorCode::InvalidInput,
+        "request body exceeds the configured limit",
+    );
+    fault.http_status = Some(413);
+    fault
+}

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

@@ -19,6 +19,7 @@ pub(crate) struct InvocationContext {
 /// 一次回调的调用链、账号与网络事实，由操作持有到完成
 pub(crate) struct CallbackScope {
     pub(super) origin: Option<InvocationContext>,
+    pub(super) request_body_limit: Option<usize>,
     stage: Stage,
     account_id: Option<String>,
     credential_revision: Option<u64>,
@@ -61,6 +62,7 @@ impl CallbackScope {
     pub(super) fn new(context: &CallContext, proxy: Option<OutboundProxy>) -> Self {
         Self {
             origin: None,
+            request_body_limit: None,
             stage: context.stage,
             account_id: context.account_id.clone(),
             credential_revision: context.credential_revision,

backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs

@@ -133,6 +133,7 @@ impl ManagedUpstream {
         &self,
         request: adapter::UpstreamHttpRequest,
         body: Vec<u8>,
+        maximum_body_bytes: usize,
     ) -> Result<(HttpRequest, UpstreamPathPurpose), PluginFault> {
         let (target, purpose) = self.target.resolve(&request.path, &request.query)?;
         if request.headers.len() > 128 {
@@ -170,6 +171,7 @@ impl ManagedUpstream {
                 url: target.into(),
                 headers,
                 body,
+                maximum_body_bytes,
             },
             purpose,
         ))
@@ -188,6 +190,7 @@ pub(super) async fn dispatch(
         scope,
         call,
         maximum_payload,
+        ..
     } = *http;
     let managed = scope.upstream.as_ref().ok_or_else(denied)?;
     let timeout = || {
@@ -197,7 +200,8 @@ pub(super) async fn dispatch(
     match method {
         "host.upstream.http.do" | "host.upstream.http.do_stream" => {
             let request = serde_json::from_value(params).map_err(|_| invalid())?;
-            let (request, purpose) = managed.request(request, payload)?;
+            let maximum_body_bytes = http.body_limit(&payload)?;
+            let (request, purpose) = managed.request(request, payload, maximum_body_bytes)?;
             http.send(request, method.ends_with("do_stream"), Some(purpose))
                 .await
         }
@@ -227,6 +231,7 @@ pub(super) async fn dispatch(
                     headers: request.headers,
                 },
                 vec![],
+                0,
             )?;
             if purpose != UpstreamPathPurpose::Inference {
                 return Err(denied());

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

@@ -21,6 +21,7 @@ impl PluginRuntime {
         &self,
         instance: PluginInstance,
         target_revision: Revision,
+        request_body_limit: Option<usize>,
     ) -> Result<Arc<PreparedInstance>, AdminError> {
         let mut observer_entries = Vec::new();
         let mut commands = Vec::new();
@@ -151,7 +152,7 @@ impl PluginRuntime {
             configuration,
             contributes: manifest.contributes.clone(),
         };
-        let callbacks = Arc::new(PluginCallbacks::new(
+        let mut callbacks = PluginCallbacks::new(
             &instance,
             self.config.rpc_limits.maximum_buffered_body_bytes,
             &manifest,
@@ -166,7 +167,10 @@ impl PluginRuntime {
                 self.affinity_ports.clone(),
                 self.resource_ports.clone(),
             ),
-        )?);
+            self.request_limits.clone(),
+        )?;
+        callbacks.registration_body_limit = request_body_limit;
+        let callbacks = Arc::new(callbacks);
         let session = Arc::new(
             RpcSession::start_supervised(
                 package.clone(),

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

@@ -13,7 +13,7 @@ impl PluginRuntime {
             .load_instances()
             .await
             .map_err(|_| AdminError::unavailable("无法读取已安装插件"))?;
-        let reference = self.prepare_snapshot(snapshot, None).await?;
+        let reference = self.prepare_snapshot(snapshot, None, None).await?;
         let set = self.prepared_set(&reference).await?;
         Ok(crate::PluginCommandSession::new(
             reference,

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

@@ -52,6 +52,7 @@ pub struct PluginRuntimeConfig {
 
 /// 只索引有主人的候选；当前代次仅由 Core 的发布视图持有
 pub struct PluginRuntime {
+    pub(super) request_limits: Arc<crate::callback::request_limits::RequestLimits>,
     pub(super) service_ports: Arc<crate::callback::services::ServicePorts>,
     pub(super) store: Arc<dyn PluginStore>,
     pub(super) state: Arc<dyn PluginStateStore>,
@@ -109,6 +110,7 @@ impl PluginRuntime {
             processes,
             validators: Arc::new(Semaphore::new(MAXIMUM_PARALLEL_PREPARATIONS)),
             configuration_schemas: Arc::default(),
+            request_limits: Arc::default(),
             service_ports: Arc::new(crate::callback::services::ServicePorts::default()),
             account_ports: Arc::new(PluginAccountPortSlot::new()),
             client_key_ports: Arc::new(PluginClientKeyPortSlot::new()),
@@ -138,6 +140,14 @@ impl PluginRuntime {
         self.service_ports.bind_http(dispatcher)
     }
 
+    /// 读取当前发布设置，同时避免回调反向保活自身所在的插件集合
+    pub fn bind_snapshots(
+        &self,
+        snapshots: &gateway_core::runtime::RuntimeSnapshotHandle,
+    ) -> Result<(), AdminError> {
+        self.request_limits.bind(snapshots)
+    }
+
     pub fn bind_services(
         &self,
         registry: &Arc<gateway_admin::public_service::Registry>,

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

@@ -35,10 +35,13 @@ impl PluginRuntime {
         &self,
         snapshot: PluginInstanceSnapshot,
         required_revision: Option<Revision>,
+        request_body_limit: Option<usize>,
     ) -> Result<ExtensionSetReference, AdminError> {
         let target_revision = snapshot.config_revision.get();
         self.record_preparation(target_revision, PreparationDiagnostic::Preparing);
-        let result = self.prepare_inner(snapshot, required_revision).await;
+        let result = self
+            .prepare_inner(snapshot, required_revision, request_body_limit)
+            .await;
         match &result {
             Ok(reference) => self.record_preparation(
                 target_revision,
@@ -58,6 +61,7 @@ impl PluginRuntime {
         &self,
         snapshot: PluginInstanceSnapshot,
         required_revision: Option<Revision>,
+        request_body_limit: Option<usize>,
     ) -> Result<ExtensionSetReference, AdminError> {
         if self.shutting_down.load(Ordering::Acquire) {
             return Err(AdminError::unavailable("插件运行时正在关闭"));
@@ -124,7 +128,11 @@ impl PluginRuntime {
                     tokio::time::timeout(
                         // 沿用准备阶段原有的 30 秒预算；打包校验与能力恢复也在预算内
                         Duration::from_secs(30),
-                        self.prepare_instance(instance.clone(), snapshot.config_revision),
+                        self.prepare_instance(
+                            instance.clone(),
+                            snapshot.config_revision,
+                            request_body_limit,
+                        ),
                     )
                     .await
                     .unwrap_or_else(|_| Err(AdminError::unavailable("插件启动超时")))
@@ -363,15 +371,33 @@ impl PluginPreparation for PluginRuntime {
         snapshot: PluginInstanceSnapshot,
     ) -> Result<ExtensionSetReference, AdminError> {
         let required_revision = snapshot.config_revision;
-        self.prepare_snapshot(snapshot, Some(required_revision))
+        self.prepare_snapshot(snapshot, Some(required_revision), None)
             .await
     }
 }
 
 impl ExtensionPreparationPort for PluginRuntime {
+    fn prepare_with_request_body_limit(
+        &self,
+        revision: ConfigRevision,
+        request_body_limit: usize,
+    ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
+        self.prepare_revision(revision, Some(request_body_limit))
+    }
+
     fn prepare(
         &self,
         revision: ConfigRevision,
+    ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
+        self.prepare_revision(revision, None)
+    }
+}
+
+impl PluginRuntime {
+    fn prepare_revision(
+        &self,
+        revision: ConfigRevision,
+        request_body_limit: Option<usize>,
     ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
         Box::pin(async move {
             let snapshot = self
@@ -382,7 +408,7 @@ impl ExtensionPreparationPort for PluginRuntime {
             if snapshot.config_revision.get() != revision.get() {
                 return Err(ExtensionPreparationError);
             }
-            self.prepare_snapshot(snapshot, None)
+            self.prepare_snapshot(snapshot, None, request_body_limit)
                 .await
                 .map_err(|_| ExtensionPreparationError)
         })

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

@@ -6,6 +6,116 @@ use serde_json::json;
 
 use crate::support::environment::Environment;
 
+#[tokio::test]
+async fn model_requests_have_a_body_limit_independent_of_the_response_buffer() {
+    use gateway_plugin_sdk::call::model::request_payload_limit;
+
+    for stream in [false, true] {
+        let Some(mut environment) = Environment::create_command().await else {
+            eprintln!("SKIP: plugin integration environment absent");
+            return;
+        };
+        let account = environment.account(None).await;
+        let key_id = format!("key_{}", uuid::Uuid::new_v4().simple());
+        environment
+            .client_key(&key_id, "sk-large-body-fixture")
+            .await;
+        let body_path = environment.directory.path().join("model-body.json");
+        let mut body = serde_json::to_vec(&json!({
+            "model":crate::support::native::MODEL,
+            "input":"x".repeat(2 * 1024 * 1024),
+        }))
+        .unwrap();
+        let raw_limit = 2 * 1024 * 1024;
+        body.resize(request_payload_limit(raw_limit), b' ');
+        std::fs::write(&body_path, &body).unwrap();
+        let marker = environment.directory.path().join("model-results.jsonl");
+        environment
+            .install_plugin(json!({
+                "plugin_id":"test.large-model-request",
+                "command_registration":{"commands":[{"name":"run","description":"执行模型"}]},
+                "command_nested_model_fixture":{
+                    "request":{
+                        "client_key_id":key_id,
+                        "model":crate::support::native::MODEL,
+                        "protocol":"openai",
+                        "operation":"generate",
+                        "provider":"openai",
+                        "account_id":account.as_str(),
+                    },
+                    "body_file":body_path,
+                    "stream":stream,
+                    "expect_error":true,
+                },
+                "command_nested_model_marker":marker,
+                "command_result":{"stdout":"ok\n","stderr":"","exit_code":0},
+            }))
+            .await;
+        let (runtime, core) = environment.runtime().await;
+        let snapshots = core.snapshots();
+        let configure = |bytes: usize| {
+            let snapshot = snapshots.acquire().unwrap();
+            let settings = snapshot
+                .settings()
+                .clone()
+                .with_responses_max_decompressed_body_bytes(bytes as u64);
+            let updated = gateway_core::routing::request_settings::RequestSettings::new(snapshot)
+                .replace(settings, "fixture")
+                .unwrap();
+            snapshots.publish((*updated.snapshot()).clone());
+        };
+        configure(raw_limit);
+        environment.store.start_command_line_writes().unwrap();
+        let instances = environment
+            .store
+            .admin_ports()
+            .plugins()
+            .load_instances()
+            .await
+            .unwrap();
+        let commands = runtime.prepare_command_line().await.unwrap();
+        let instance = &instances.instances[0];
+        commands.execute(&instance.id, "run", &[]).await.unwrap();
+        environment.wait_for_bound_model_requests(&key_id, 1).await;
+        let first: serde_json::Value = serde_json::from_str(
+            std::fs::read_to_string(&marker)
+                .unwrap()
+                .lines()
+                .next()
+                .unwrap(),
+        )
+        .unwrap();
+        assert!(first["request_id"].is_string(), "{first}");
+        assert!(first["events"].as_u64().unwrap() > 0);
+
+        // 合法 JSON 仅多一个空白字节，超限必须在进入 Core 和账本前拒绝
+        let mut oversized = body.clone();
+        oversized.push(b'\n');
+        std::fs::write(&body_path, oversized).unwrap();
+        commands.execute(&instance.id, "run", &[]).await.unwrap();
+        let records = std::fs::read_to_string(&marker).unwrap();
+        let rejected: serde_json::Value =
+            serde_json::from_str(records.lines().last().unwrap()).unwrap();
+        assert_eq!(rejected["error"], "invalid_input");
+        assert_eq!(rejected["http_status"], 413);
+        assert_eq!(environment.bound_model_requests(&key_id).await.len(), 1);
+
+        // 不重新准备插件实例：发布更大的配置后，同一个请求必须恢复成功
+        configure(raw_limit + 1);
+        commands.execute(&instance.id, "run", &[]).await.unwrap();
+        environment.wait_for_bound_model_requests(&key_id, 2).await;
+        commands.shutdown().await;
+        environment
+            .store
+            .shutdown_command_line_writes()
+            .await
+            .unwrap();
+        drop(core);
+        drop(runtime);
+        environment.close().await;
+    }
+}
+
 #[tokio::test]
 async fn command_selects_key_at_call_time_and_uses_normal_model_ledger() {
     let Some(mut environment) = Environment::create_command().await else {
@@ -277,3 +387,131 @@ async fn command_accounts_domain_does_not_need_separate_login_or_write_grants()
     drop(runtime);
     environment.close().await;
 }
+
+#[tokio::test]
+async fn bound_model_body_budget_stays_frozen_until_the_parent_call_ends() {
+    use gateway_plugin_sdk::call::model::request_payload_limit;
+
+    for stream in [false, true] {
+        for growing in [false, true] {
+            let Some(mut environment) = Environment::create_command().await else {
+                eprintln!("SKIP: plugin integration environment absent");
+                return;
+            };
+            let account = environment.account(None).await;
+            let key_id = format!("key_{}", uuid::Uuid::new_v4().simple());
+            environment
+                .client_key(&key_id, "sk-frozen-body-fixture")
+                .await;
+            let old_limit = if growing { 64 * 1024 } else { 128 * 1024 };
+            let new_limit = if growing { 128 * 1024 } else { 64 * 1024 };
+            let body_path = environment.directory.path().join("model-body.json");
+            let body = json!({"model":crate::support::native::MODEL,"input":"hello"});
+            let mut large_body = serde_json::to_vec(&body).unwrap();
+            large_body.resize(
+                request_payload_limit(old_limit) + usize::from(growing),
+                b' ',
+            );
+            std::fs::write(&body_path, large_body).unwrap();
+            let marker = environment.directory.path().join("model-results.jsonl");
+            let gate = environment.directory.path().join("resume-model");
+            let request = json!({
+                "client_key_id":key_id,"model":crate::support::native::MODEL,
+                "protocol":"openai","operation":"generate","provider":"openai",
+                "account_id":account.as_str(),
+            });
+            environment
+                .install_plugin(json!({
+                    "plugin_id":"test.frozen-model-request",
+                    "command_registration":{"commands":[{"name":"run","description":"执行模型"}]},
+                    "command_nested_model_fixture":[
+                        {"request":request,"body":body,"stream":stream},
+                        {"request":request,"body_file":body_path,"stream":stream,
+                            "expect_error":true,"wait_for_file":gate},
+                    ],
+                    "command_nested_model_marker":marker,
+                    "command_result":{"stdout":"ok\n","stderr":"","exit_code":0},
+                }))
+                .await;
+            let (runtime, core) = environment.runtime().await;
+            let snapshots = core.snapshots();
+            let configure = |bytes: usize| {
+                let snapshot = snapshots.acquire().unwrap();
+                let settings = snapshot
+                    .settings()
+                    .clone()
+                    .with_responses_max_decompressed_body_bytes(bytes as u64);
+                let updated =
+                    gateway_core::routing::request_settings::RequestSettings::new(snapshot)
+                        .replace(settings, "fixture")
+                        .unwrap();
+                snapshots.publish((*updated.snapshot()).clone());
+            };
+            configure(old_limit);
+            environment.store.start_command_line_writes().unwrap();
+            let instances = environment
+                .store
+                .admin_ports()
+                .plugins()
+                .load_instances()
+                .await
+                .unwrap();
+            let commands = runtime.prepare_command_line().await.unwrap();
+            let instance = &instances.instances[0];
+            let update = async {
+                tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                    while std::fs::read_to_string(&marker)
+                        .unwrap_or_default()
+                        .lines()
+                        .count()
+                        < 1
+                    {
+                        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+                    }
+                })
+                .await
+                .unwrap();
+                configure(new_limit);
+                std::fs::write(&gate, b"resume").unwrap();
+            };
+            let (result, ()) = tokio::join!(commands.execute(&instance.id, "run", &[]), update);
+            result.unwrap();
+            let expected = if growing { 1 } else { 2 };
+            environment
+                .wait_for_bound_model_requests(&key_id, expected)
+                .await;
+            assert_eq!(
+                environment.bound_model_requests(&key_id).await.len(),
+                expected
+            );
+
+            // 父调用结束后重新绑定，下一次调用才使用新发布的预算
+            commands.execute(&instance.id, "run", &[]).await.unwrap();
+            environment.wait_for_bound_model_requests(&key_id, 3).await;
+            let records = std::fs::read_to_string(&marker)
+                .unwrap()
+                .lines()
+                .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
+                .collect::<Vec<_>>();
+            assert_eq!(records.len(), 4);
+            for (index, rejected) in [(1, growing), (3, !growing)] {
+                if rejected {
+                    assert_eq!(records[index]["http_status"], 413, "{records:?}");
+                    assert_eq!(records[index]["error"], "invalid_input");
+                } else {
+                    assert!(records[index]["request_id"].is_string(), "{records:?}");
+                }
+            }
+            assert_eq!(environment.bound_model_requests(&key_id).await.len(), 3);
+            commands.shutdown().await;
+            environment
+                .store
+                .shutdown_command_line_writes()
+                .await
+                .unwrap();
+            drop(core);
+            drop(runtime);
+            environment.close().await;
+        }
+    }
+}

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

@@ -187,3 +187,164 @@ async fn mixed_mounts_keep_independent_order_and_fresh_continuations() {
     drop(generation);
     runtime.shutdown().await;
 }
+
+#[tokio::test]
+async fn data_plane_http_callbacks_keep_the_frozen_body_budget_across_publications() {
+    use gateway_core::{
+        engine::policy::RetryFacts,
+        error::ProviderErrorKind,
+        policy::ClientPolicy,
+        routing::{RuntimeSnapshot, request_settings::RequestSettings},
+        runtime::RuntimeSnapshotHandle,
+        settings::SettingsValues,
+        upstream::UpstreamSendState,
+    };
+
+    let snapshot = |limit: u64, revision| {
+        RuntimeSnapshot::new(
+            ConfigRevision::new(revision).unwrap(),
+            SettingsValues::new(1, "smart", Default::default(), None, None)
+                .with_responses_max_decompressed_body_bytes(limit),
+            vec![],
+            vec![],
+            vec![],
+        )
+        .unwrap()
+    };
+    for body_bytes in [128, 129] {
+        let server = MockServer::start().await;
+        Mock::given(any())
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+        let records = tempfile::tempdir().unwrap();
+        let marker = records.path().join("http-results.jsonl");
+        let (cache, runtime) = setup(vec![InstanceFixture {
+            id: "frozen-limits",
+            configuration: serde_json::json!({
+                "http_probe_methods":["middleware.handle","policy.route_model","policy.schedule_account","policy.retry_decision"],
+                "http_probe_url":server.uri(),"http_probe_body_bytes":body_bytes,"http_probe_marker":marker,
+                "retry_decision":{"decision":"delegate"},
+                "route_decision":{"decision":"unhandled"},
+                "schedule_decision":{"decision":"delegate"},
+            }),
+            bindings: vec![
+                binding(MIDDLEWARE_CONTRIBUTION,"request",0,PluginFailurePolicy::Reject),
+                binding(MIDDLEWARE_CONTRIBUTION,"attempt",0,PluginFailurePolicy::Reject),
+                binding(MODEL_ROUTER_CONTRIBUTION,"routing",0,PluginFailurePolicy::Reject),
+                binding(SCHEDULER_CONTRIBUTION,"scheduling",0,PluginFailurePolicy::Reject),
+                binding("test.example.retryPolicy","retry",0,PluginFailurePolicy::Delegate),
+            ],
+        }]).await;
+        let snapshots = RuntimeSnapshotHandle::new(snapshot(128, 1));
+        runtime.bind_snapshots(&snapshots).unwrap();
+        let settings = RequestSettings::new(snapshots.acquire().unwrap()).with_execution(
+            &ClientPolicy::new(
+                ClientApiKeyId::new("key-middleware").unwrap(),
+                gateway_core::policy::PlaintextClientApiKey::new("frozen-budget-fixture").unwrap(),
+                Arc::new(gateway_core::routing::FrozenAccountScope::new(
+                    Arc::default(),
+                    gateway_core::routing::ClientRoutingScope::all_accounts(),
+                )),
+                true,
+                Default::default(),
+            ),
+            Some(60_000),
+        );
+        let generation = prepare(&runtime).await;
+        let plan = runtime
+            .execution_registry()
+            .middleware(&generation)
+            .unwrap();
+        let policy = policy_context(&runtime, generation.clone(), "req_frozen_limits")
+            .with_request_body_limit(128);
+        for (revision, current_limit) in [(2, 64), (3, 256)] {
+            snapshots.publish(snapshot(current_limit, revision));
+            for attempt in [false, true] {
+                let mut request = MiddlewareRequest::new(
+                    "openai",
+                    middleware_headers(),
+                    Bytes::from_static(br#"{"input":"hello"}"#),
+                );
+                let context = if attempt {
+                    attempt_middleware_context(Arc::new(ExecutionEffects::default()))
+                        .with_request_body_limit(Some(128))
+                } else {
+                    request = request.with_settings(settings.clone());
+                    middleware_context(ClientTransport::HttpJson)
+                };
+                let response = plan
+                    .handle(
+                        context,
+                        request,
+                        Downstream {
+                            calls: Arc::default(),
+                            closes: Arc::default(),
+                            reads: Arc::default(),
+                            error: false,
+                        }
+                        .into_next(),
+                    )
+                    .await
+                    .unwrap();
+                let (_, _, _, mut body, _) = response.into_parts();
+                while body.next_frame().await.unwrap().is_some() {}
+                body.close().await;
+            }
+            policy
+                .route_model(
+                    operation(),
+                    PublicModelId::new("public-a").unwrap(),
+                    BTreeSet::from([ProviderKind::new("example").unwrap()]),
+                )
+                .await
+                .unwrap();
+            policy
+                .select_account(
+                    NonZeroU32::MIN,
+                    &ProviderKind::new("example").unwrap(),
+                    None,
+                    &[candidate("acct_high", 100)],
+                    &mut selection_context(),
+                )
+                .await
+                .unwrap();
+            policy
+                .retry_decision(RetryFacts {
+                    attempt_index: NonZeroU32::MIN,
+                    provider: ProviderKind::new("example").unwrap(),
+                    model: None,
+                    error_kind: ProviderErrorKind::RateLimited,
+                    upstream_status: Some(429),
+                    send_state: UpstreamSendState::NotSent,
+                    remaining_routing_attempts: 1,
+                    remaining_deadline: Duration::from_secs(3),
+                    retry_allowed: true,
+                })
+                .await;
+        }
+        let records = std::fs::read_to_string(marker)
+            .unwrap()
+            .lines()
+            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
+            .collect::<Vec<_>>();
+        assert_eq!(records.len(), 10, "{records:?}");
+        let expected = if body_bytes == 128 { 200 } else { 413 };
+        assert!(
+            records.iter().all(|record| record["status"] == expected),
+            "{records:?}"
+        );
+        let received = server.received_requests().await.unwrap();
+        assert_eq!(received.len(), if body_bytes == 128 { 10 } else { 0 });
+        assert!(
+            received
+                .iter()
+                .all(|request| request.body == vec![b'x'; body_bytes])
+        );
+        drop(policy);
+        drop(plan);
+        drop(generation);
+        runtime.shutdown().await;
+        super::super::wait_until_empty(cache.path()).await;
+    }
+}

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

@@ -9,6 +9,44 @@ use gateway_plugin_sdk::Capability;
 use gateway_plugin_sdk::Contributions;
 use gateway_plugin_sdk::Stage;
 
+#[tokio::test]
+async fn initial_registration_http_uses_the_prepared_settings_without_a_publication() {
+    use wiremock::{Mock, MockServer, ResponseTemplate, matchers::method};
+    for limit in [63, 64] {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200))
+            .expect(u64::from(limit == 64))
+            .mount(&server)
+            .await;
+        let (cache, store, runtime) =
+            super::super::setup_with_contributions(Contributions::from([
+                crate::support::contribution(
+                    Capability::ModelCatalog,
+                    vec![Stage::Registration],
+                    vec![],
+                    vec![],
+                ),
+            ]))
+            .await;
+        store.snapshot.lock().unwrap().instances[0].configuration = serde_json::json!({
+            "registration_http": {"params": {"method":"POST", "url":server.uri(), "headers":[]}, "body":"x".repeat(64)},
+            "model_catalog":{"models":[{"id":"fixture-alias","provider":"openai","model":"fixture-upstream"}]}
+        });
+        let prepared = ExtensionPreparationPort::prepare_with_request_body_limit(
+            &runtime,
+            ConfigRevision::new(1).unwrap(),
+            limit,
+        )
+        .await
+        .unwrap();
+        assert_eq!(prepared.model_aliases().len(), usize::from(limit == 64));
+        drop(prepared);
+        runtime.shutdown().await;
+        super::super::wait_until_empty(cache.path()).await;
+    }
+}
+
 #[tokio::test]
 async fn model_catalog_is_frozen_without_entering_the_request_chain() {
     let (cache, store, runtime) = super::super::setup_with_contributions(Contributions::from([

backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs

@@ -159,6 +159,16 @@ fn context_with_fast_mode(
     generation: &ExtensionSetReference,
     key: &str,
     mode: gateway_core::account::FastMode,
+) -> AttemptContext {
+    context_with_body_limit(runtime, generation, key, mode, 64 * 1024 * 1024)
+}
+
+fn context_with_body_limit(
+    runtime: &gateway_plugin_runtime::PluginRuntime,
+    generation: &ExtensionSetReference,
+    key: &str,
+    mode: gateway_core::account::FastMode,
+    bytes: usize,
 ) -> AttemptContext {
     AttemptContext::new(
         RequestAttemptContext::new(
@@ -166,7 +176,8 @@ fn context_with_fast_mode(
             ClientApiKeyId::new(key).unwrap(),
         )
         .with_upstream_adapters(runtime.execution_registry().upstream_adapters(generation))
-        .with_fast_mode(mode),
+        .with_fast_mode(mode)
+        .with_request_body_limit(bytes),
         NonZeroU32::new(1).unwrap(),
         SystemTime::now() + Duration::from_secs(5),
         AccountSelectionPolicy::new(
@@ -1225,3 +1236,120 @@ async fn malformed_adapter_events_fail_only_the_current_stream() {
         super::wait_until_empty(cache.path()).await;
     }
 }
+
+#[tokio::test]
+async fn large_adapter_requests_use_the_frozen_budget_for_both_http_capabilities() {
+    for streaming in [false, true] {
+        for upstream in [false, true] {
+            let server = MockServer::start().await;
+            Mock::given(method("POST"))
+                .respond_with(ResponseTemplate::new(200).set_body_string(if streaming {
+                    "data: {}\n\n"
+                } else {
+                    "{}"
+                }))
+                .expect(if upstream { 2 } else { 4 })
+                .mount(&server)
+                .await;
+            let mut config = configuration(&server.uri(), streaming);
+            if !upstream {
+                let mut inference = config["upstream_callbacks"][0].clone();
+                inference["body"] = json!("{}");
+                config["upstream_callbacks"]
+                    .as_array_mut()
+                    .unwrap()
+                    .push(inference);
+                config["upstream_callbacks"][0]["method"] = json!(if streaming {
+                    "host.http.do_stream"
+                } else {
+                    "host.http.do"
+                });
+                config["upstream_callbacks"][0]["params"] = json!({"method":"POST", "url":format!("{}/auxiliary", server.uri()), "headers":[]});
+            }
+            let (cache, _, runtime) = setup(config).await;
+            let generation =
+                ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap())
+                    .await
+                    .unwrap();
+            let account = Account::new();
+            let operation = Operation::Generate(GenerateRequest::from_protocol_payload(
+                ProtocolPayload::json_object(
+                    "openai",
+                    json!({"model":"native-model","input":"x".repeat(17 * 1024 * 1024)})
+                        .as_object()
+                        .unwrap()
+                        .clone(),
+                )
+                .unwrap(),
+            ));
+            let expected = operation.middleware_body().unwrap();
+            for accepted in [true, false, true] {
+                let context = context_with_body_limit(
+                    &runtime,
+                    &generation,
+                    "key-one",
+                    Default::default(),
+                    expected.len() - usize::from(!accepted),
+                );
+                let provider = ProviderKind::new("openai").unwrap();
+                let model = UpstreamModelId::new("native-model").unwrap();
+                let adapter = context
+                    .upstream_adapter(&provider, &model)
+                    .unwrap()
+                    .unwrap();
+                let metadata = ProviderCallMetadata::new(
+                    provider,
+                    model,
+                    account.id.clone(),
+                    UpstreamTransport::new(adapter.transport()).unwrap(),
+                );
+                let mut stream = adapter.execute(UpstreamAdapterInvocation {
+                    operation: operation.clone(),
+                    headers: vec![],
+                    context,
+                    metadata,
+                    account: account.clone(),
+                });
+                let mut completed = false;
+                let mut failure = None;
+                while let Some(event) = stream.next().await {
+                    match event {
+                        Ok(event) => {
+                            completed |= event
+                                .canonical_facts()
+                                .iter()
+                                .any(|fact| matches!(fact, GatewayEvent::Completed(_)))
+                        }
+                        Err(error) => {
+                            failure = Some(error);
+                            break;
+                        }
+                    }
+                }
+                if accepted {
+                    assert!(
+                        failure.is_none(),
+                        "streaming={streaming} upstream={upstream}: {failure:?}"
+                    );
+                    assert!(completed);
+                } else {
+                    assert_eq!(failure.unwrap().send_state(), UpstreamSendState::NotSent);
+                }
+            }
+            let requests = server
+                .received_requests()
+                .await
+                .unwrap()
+                .into_iter()
+                .filter(|request| request.body.len() > 1024 * 1024)
+                .collect::<Vec<_>>();
+            assert_eq!(requests.len(), 2);
+            for request in requests {
+                assert_eq!(request.body.as_slice(), expected.as_ref());
+            }
+            drop(generation);
+            runtime.shutdown().await;
+            super::wait_until_empty(cache.path()).await;
+        }
+    }
+}

backend/crates/gateway-plugin/runtime/tests/support/environment.rs

@@ -474,6 +474,7 @@ impl Environment {
             Some(runtime.execution_registry()),
             Some(runtime.frontend_authentication_registry()),
         );
+        runtime.bind_snapshots(&core.snapshots()).unwrap();
         let access = gateway_admin::initialize_plugin_accounts(
             super::native::admin_registry(),
             self.store.admin_ports().accounts(),
@@ -509,6 +510,7 @@ impl Environment {
         .activate()
         .await
         .unwrap();
+        runtime.bind_snapshots(&core.snapshots()).unwrap();
         runtime
             .bind_model_ports(
                 &core.nested_model_execution_port(),

backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs

@@ -284,15 +284,51 @@ impl Peer {
         let Some(fixture) = self.configuration.get(fixture_field) else {
             return Ok(());
         };
+        for fixture in fixture
+            .as_array()
+            .map_or(std::slice::from_ref(fixture), Vec::as_slice)
+        {
+            if let Some(gate) = fixture["wait_for_file"].as_str() {
+                while !std::path::Path::new(gate).exists() {
+                    tokio::time::sleep(Duration::from_millis(5)).await;
+                }
+            }
+            self.execute_nested_model_fixture(parent, fixture, marker_field)
+                .await?;
+        }
+        Ok(())
+    }
+
+    async fn execute_nested_model_fixture(
+        &self,
+        parent: u64,
+        fixture: &Value,
+        marker_field: &str,
+    ) -> Result<(), PluginFault> {
+        let body = if let Some(path) = fixture["body_file"].as_str() {
+            std::fs::read(path).unwrap()
+        } else {
+            serde_json::to_vec(&fixture["body"]).unwrap()
+        };
         if fixture["stream"] == true {
-            let (result, payload) = self
+            let response = self
                 .callback_payload(
                     parent,
                     "host.model.execute_stream",
                     fixture["request"].clone(),
-                    serde_json::to_vec(&fixture["body"]).unwrap(),
+                    body,
                 )
-                .await?;
+                .await;
+            let (result, payload) = match response {
+                Err(error) if fixture["expect_error"] == true => {
+                    self.append_observation_marker(
+                        marker_field,
+                        &json!({"error":error.code,"http_status":error.http_status}),
+                    );
+                    return Ok(());
+                }
+                result => result?,
+            };
             assert!(payload.is_empty());
             let stream: gateway_plugin_sdk::call::host::ModelStreamResult =
                 serde_json::from_value(result).unwrap();
@@ -348,7 +384,7 @@ impl Peer {
                 parent,
                 "host.model.execute",
                 fixture["request"].clone(),
-                serde_json::to_vec(&fixture["body"]).unwrap(),
+                body,
             )
             .await;
         match callback {
@@ -366,7 +402,10 @@ impl Peer {
                 Ok(())
             }
             Err(error) if fixture["expect_error"] == true => {
-                self.append_observation_marker(marker_field, &json!({"error":error.code}));
+                self.append_observation_marker(
+                    marker_field,
+                    &json!({"error":error.code,"http_status":error.http_status}),
+                );
                 Ok(())
             }
             Err(error) => {
@@ -439,6 +478,37 @@ impl Peer {
             self.send(Message::End { id, error: None }, vec![]).await;
             return;
         }
+        if self.configuration["http_probe_methods"]
+            .as_array()
+            .is_some_and(|methods| {
+                methods
+                    .iter()
+                    .any(|probe| probe.as_str() == Some(method.as_str()))
+            })
+        {
+            let body = vec![
+                b'x';
+                self.configuration["http_probe_body_bytes"]
+                    .as_u64()
+                    .unwrap() as usize
+            ];
+            let response = self
+                .callback_payload(
+                    id,
+                    "host.http.do",
+                    json!({"method":"POST", "url":self.configuration["http_probe_url"]}),
+                    body,
+                )
+                .await;
+            let status = match response {
+                Ok((result, _)) => result["status"].as_u64().unwrap(),
+                Err(error) => u64::from(error.http_status.unwrap()),
+            };
+            self.append_observation_marker(
+                "http_probe_marker",
+                &json!({"method":method,"status":status}),
+            );
+        }
         match method.as_str() {
             "upstream_adapter.register" => {
                 self.send(
@@ -1304,6 +1374,20 @@ impl Peer {
                 return;
             }
             "plugin.register" => {
+                if let Some(request) = self.configuration.get("registration_http") {
+                    let result = self
+                        .callback_payload(
+                            id,
+                            "host.http.do",
+                            request["params"].clone(),
+                            request["body"].as_str().unwrap().as_bytes().to_vec(),
+                        )
+                        .await;
+                    if let Err(error) = result {
+                        self.send(Message::Error { id, error }, vec![]).await;
+                        return;
+                    }
+                }
                 if self.configuration["exit_during_registration"] == true {
                     std::process::exit(31);
                 }
@@ -1598,13 +1682,19 @@ impl Peer {
                 )
             }))
             .await?;
-            for (result, payload) in results {
+            for (callback, (result, payload)) in callbacks.iter().zip(results) {
                 if let Some(stream) = result["stream"].as_str() {
+                    response_body.clear();
+                    let read_method = if callback["method"] == "host.http.do_stream" {
+                        "host.http.stream_read"
+                    } else {
+                        "host.upstream.http.stream_read"
+                    };
                     loop {
                         let (result, chunk) = self
                             .callback_payload(
                                 id,
-                                "host.upstream.http.stream_read",
+                                read_method,
                                 json!({"stream":stream,"maximum_bytes":65536}),
                                 vec![],
                             )

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

@@ -209,6 +209,17 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限
 - `host.model.execute[_stream]`：交给 Core 的路由、准入、租约、重试、账本和计费链路执行
 - `host.model.stream_read/close`：读取或关闭当前父调用创建的有界流
 
+模型请求正文使用独立的二进制载荷，不共用响应缓冲预算。宿主从有效设置
+`responses_max_decompressed_body_bytes` 取得原始正文预算 `L`，通过
+`call::model::request_payload_limit(L)` 为内部请求预留 JSON 字符串转义和元数据空间
+（`2 × L + 64 KiB`，受请求编码的 32 位分段长度约束）。超限返回带 HTTP 413 的
+`invalid_input`，不启动模型执行。上游适配器的请求 `GPAQ` 也使用请求预算，响应事件仍受
+原有 8 MiB 编码上限、Runtime 缓冲预算及流控窗口限制
+
+HTTP 和 request 中间件中的模型回调沿用有效请求设置；attempt、路由、调度、重试和上游适配器
+沿用该次模型请求冻结的正文预算。显式 Key 模型调用的预检与执行使用同一绑定快照，
+同一父调用内复用 Key 时不受后续配置发布影响；父调用结束后重新绑定才读取新配置，无需重新启用插件
+
 `ModelExecuteRequest.client_key_id` 可显式选择执行 Key，不与触发插件的父请求 Key 求交。省略时继承现有模型父请求身份；没有模型父请求的调用必须提供 Key。所选 Key 的启用状态、模型规则、账号组、并发和预算继续由 Core 执行
 
 模型事件使用 `call::model::ExecutionEvent` 的 `GPE2` 封套，将 canonical 事实、宿主事实和原始 wire 分段编码。`host` 包含费用及定价明细、完整上游观测、会话状态和改写来源；当前 wire 与原始 wire 都保留二进制字节。读取或回传快照不触发第二次计费、续接提交或发送状态更新。
@@ -227,6 +238,13 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限
 账号上游使用同一正文对象，使用方式见[受管上游](upstream-adapters.md)
 
 `host.http.do/do_stream` 和流读取／关闭方法提供受管 HTTP。正文使用独立二进制载荷，Host 统一执行代理、超时和背压，不按插件身份限制目标地址段。DNS 由宿主解析，连接保留目标 Host 与 TLS 域名校验；重定向不会自动跟随。
+受管 HTTP（包括 `host.upstream.http.do[_stream]`）的请求正文按有效设置
+`responses_max_decompressed_body_bytes` 检查。HTTP/request 中间件使用有效请求设置；
+attempt、路由、调度、重试和上游适配器使用执行冻结预算，其余调用读取当前发布配置。
+超限返回带 HTTP 413 的 `invalid_input` 和 `not_sent`，不向目标发送请求。
+注册阶段沿用正在编译的配置预算，首次发布前也可发送非空正文。
+非空正文无法取得有效配置时拒绝调用；无正文查询不消耗正文预算。
+这里检查实际出站正文，内部模型请求的转义余量不增加出站预算。响应缓冲及流窗口不随请求配置扩大
 普通调用使用父调用剩余期限；已建立的受管 HTTP 流或 WebSocket 会话按每次网络操作计时，仍受父资源取消约束
 
 期限耗尽返回 `timeout`，解析、连接或响应读取失败返回 `upstream`。错误消息提供分类提示，响应正文仍由插件读取；协议故障不应直接展示内部消息

backend/crates/gateway-plugin/sdk/src/call/model/codec.rs

@@ -45,29 +45,44 @@ pub(in crate::call) fn pack<T: Serialize, const N: usize>(
     prefix: [u8; 4],
     metadata: &T,
     payloads: [Vec<u8>; N],
+) -> Result<Vec<u8>, ExecutionEncodingError> {
+    pack_with_limit(prefix, metadata, payloads, MAX_EXECUTION_PAYLOAD_BYTES)
+}
+
+pub(in crate::call) fn pack_with_limit<T: Serialize, const N: usize>(
+    prefix: [u8; 4],
+    metadata: &T,
+    payloads: [Vec<u8>; N],
+    maximum: usize,
 ) -> Result<Vec<u8>, ExecutionEncodingError> {
     let header_bytes = 4 + 4 * (N + 1);
     let raw_bytes = payloads
         .iter()
-        .fold(0usize, |total, bytes| total.saturating_add(bytes.len()));
-    if raw_bytes > MAX_EXECUTION_PAYLOAD_BYTES {
+        .try_fold(0usize, |total, bytes| total.checked_add(bytes.len()))
+        .ok_or(ExecutionEncodingError)?;
+    if raw_bytes > maximum {
         return Err(ExecutionEncodingError);
     }
     let metadata = serde_json::to_vec(metadata).map_err(|_| ExecutionEncodingError)?;
     let total = header_bytes
-        .saturating_add(metadata.len())
-        .saturating_add(raw_bytes);
-    if total > MAX_EXECUTION_PAYLOAD_BYTES {
+        .checked_add(metadata.len())
+        .and_then(|total| total.checked_add(raw_bytes))
+        .ok_or(ExecutionEncodingError)?;
+    let lengths = std::iter::once(metadata.len())
+        .chain(payloads.iter().map(Vec::len))
+        .map(u32::try_from)
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|_| ExecutionEncodingError)?;
+    if total > maximum {
         return Err(ExecutionEncodingError);
     }
-    let mut result = Vec::with_capacity(total);
+    let mut result = Vec::new();
+    result
+        .try_reserve_exact(total)
+        .map_err(|_| ExecutionEncodingError)?;
     result.extend_from_slice(&prefix);
-    for length in std::iter::once(metadata.len()).chain(payloads.iter().map(Vec::len)) {
-        result.extend_from_slice(
-            &u32::try_from(length)
-                .map_err(|_| ExecutionEncodingError)?
-                .to_be_bytes(),
-        );
+    for length in lengths {
+        result.extend_from_slice(&length.to_be_bytes());
     }
     result.extend_from_slice(&metadata);
     for bytes in payloads {
@@ -79,12 +94,17 @@ pub(in crate::call) fn pack<T: Serialize, const N: usize>(
 pub(in crate::call) fn unpack<T: DeserializeOwned, const N: usize>(
     prefix: [u8; 4],
     bytes: &[u8],
+) -> Result<(T, [&[u8]; N]), ExecutionEncodingError> {
+    unpack_with_limit(prefix, bytes, MAX_EXECUTION_PAYLOAD_BYTES)
+}
+
+pub(in crate::call) fn unpack_with_limit<T: DeserializeOwned, const N: usize>(
+    prefix: [u8; 4],
+    bytes: &[u8],
+    maximum: usize,
 ) -> Result<(T, [&[u8]; N]), ExecutionEncodingError> {
     let header_bytes = 4 + 4 * (N + 1);
-    if bytes.len() < header_bytes
-        || bytes.len() > MAX_EXECUTION_PAYLOAD_BYTES
-        || bytes[..4] != prefix
-    {
+    if bytes.len() < header_bytes || bytes.len() > maximum || bytes[..4] != prefix {
         return Err(ExecutionEncodingError);
     }
     let mut cursor = header_bytes;

backend/crates/gateway-plugin/sdk/src/call/model/mod.rs

@@ -6,6 +6,15 @@ pub(in crate::call) mod codec;
 pub mod facts;
 pub use codec::{ExecutionEncodingError, MAX_EXECUTION_PAYLOAD_BYTES};
 
+/// 内部请求允许 JSON 字符串转义与元数据开销，出站正文仍按原始配置上限检查
+#[must_use]
+pub fn request_payload_limit(body_bytes: usize) -> usize {
+    body_bytes
+        .saturating_mul(2)
+        .saturating_add(64 * 1024)
+        .min(u32::MAX as usize)
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum ContentKind {

backend/crates/gateway-plugin/sdk/src/call/upstream_adapter/codec.rs

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 use super::{UpstreamAdapterEvent, UpstreamAdapterRequest, UpstreamContinuation, UpstreamFailure};
 use crate::call::model::{
     ExecutionEncodingError, ExecutionEvent,
-    codec::{pack, unpack},
+    codec::{pack, pack_with_limit, unpack, unpack_with_limit},
 };
 
 #[derive(Serialize, Deserialize)]
@@ -38,12 +38,13 @@ pub(super) fn encode_request(
     request: UpstreamAdapterRequest,
     body: Vec<u8>,
 ) -> Result<Vec<u8>, ExecutionEncodingError> {
-    pack(*b"GPAQ", &request, [body])
+    // 请求预算由宿主配置检查，线协议只约束 u32 长度；响应事件仍受独立预算限制
+    pack_with_limit(*b"GPAQ", &request, [body], usize::MAX)
 }
 
 pub(super) fn decode_request(
     bytes: &[u8],
 ) -> Result<(UpstreamAdapterRequest, Vec<u8>), ExecutionEncodingError> {
-    let (request, [body]) = unpack(*b"GPAQ", bytes)?;
+    let (request, [body]) = unpack_with_limit(*b"GPAQ", bytes, usize::MAX)?;
     Ok((request, body.to_vec()))
 }

backend/crates/gateway-plugin/sdk/tests/call/mod.rs

@@ -8,3 +8,4 @@ mod management;
 mod middleware;
 mod observation;
 mod policy;
+mod upstream_adapter;

backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/codec.rs

@@ -0,0 +1,46 @@
+//! 验证上游请求的大正文往返与响应事件预算保持独立
+
+use gateway_plugin_sdk::call::{
+    model::{ExecutionEvent, MAX_EXECUTION_PAYLOAD_BYTES, WireEvent, WirePayload},
+    upstream_adapter::UpstreamAdapterRequest,
+};
+use serde_json::json;
+
+fn request() -> UpstreamAdapterRequest {
+    serde_json::from_value(json!({
+        "adapter_id":"example", "provider":"openai", "upstream_model":"fixture",
+        "client_key_id":"key", "account_id":"account", "credential_revision":1,
+        "protocol":"openai", "client_transport":"http_json", "fast_mode":"default",
+        "headers":[["x-fixture",[97,98]]], "continuation":null
+    }))
+    .unwrap()
+}
+
+#[test]
+fn large_request_preserves_bytes_and_rejects_malformed_lengths() {
+    let body = (0..MAX_EXECUTION_PAYLOAD_BYTES + 1)
+        .map(|i| (i % 251) as u8)
+        .collect::<Vec<_>>();
+    let encoded = request().encode(body.clone()).unwrap();
+    let (decoded, actual) = UpstreamAdapterRequest::decode(&encoded).unwrap();
+    assert_eq!(actual, body);
+    assert_eq!(decoded.headers, request().headers);
+    assert!(UpstreamAdapterRequest::decode(&encoded[..encoded.len() - 1]).is_err());
+    let mut corrupted = encoded;
+    corrupted[8..12].copy_from_slice(&u32::MAX.to_be_bytes());
+    assert!(UpstreamAdapterRequest::decode(&corrupted).is_err());
+}
+
+#[test]
+fn large_requests_do_not_expand_response_event_budget() {
+    let event = ExecutionEvent {
+        wire: Some(WireEvent {
+            protocol: "openai".into(),
+            payload: WirePayload::RawBody {
+                body: vec![b'x'; MAX_EXECUTION_PAYLOAD_BYTES],
+            },
+        }),
+        ..Default::default()
+    };
+    assert!(event.encode().is_err());
+}

backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/mod.rs

@@ -0,0 +1,3 @@
+//! 上游适配器调用合同的测试入口
+
+mod codec;

docs/api.md

@@ -265,7 +265,8 @@ Codex 的 review 等子代理请求仍使用 `/v1/responses`，并通过 `x-open
 多帧连续解码，整体展开结果受运行设置 `responsesMaxDecompressedBodyBytes` 约束（默认 64 MiB），
 超限在继续展开前返回 `400 request_too_large`，错误信息包含当前请求的上限字节数。zstd
 回溯窗口独立固定为最多 64 MiB，不能满足该限制的帧按解码失败处理。这个限制保护入站解压资源，不是
-模型上下文或 Token 上限；未压缩正文不受此长度限制。
+模型上下文或 Token 上限；未压缩正文不经过此解压检查。插件受管出站仍受同一设置约束，
+见[请求解压上限](#请求解压上限)。
 不支持的编码、逗号分隔的叠加编码和重复 `Content-Encoding` 头返回
 `400 unsupported_content_encoding`；压缩正文损坏、截断或解压后不是合法 JSON 返回
 `400 invalid_json`。本地错误不包含原始正文或解压库细节。WebSocket 文本帧不经过这条解压路径
@@ -1627,10 +1628,15 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi
 
 ### 请求解压上限
 
-`responsesMaxDecompressedBodyBytes` 是压缩 Responses HTTP 请求的解压输出上限，单位字节，默认
-67108864（64 MiB）。必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。
-保存并发布成功后，新请求使用新值；已鉴权请求沿用原快照，无需重启。调高上限会增加大请求的内存占用，
-它不代表整个进程的内存预算
+`responsesMaxDecompressedBodyBytes` 限制压缩 Responses HTTP 请求的解压输出，以及插件受管 HTTP
+的实际出站正文（包括账号上游请求，无论是否压缩），单位字节，默认 67108864（64 MiB）。
+必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。
+插件模型调用从同一设置派生内部请求编码预算，具体边界见 [SDK 模型请求](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#key模型与模型调用)
+
+保存并发布成功后，新请求使用新值；在途模型请求及其路由、调度、attempt 和重试回调沿用冻结预算。
+显式 Key 的模型调用沿用当前父调用内绑定的快照，父调用结束后重新绑定时使用新值；无模型请求上下文的
+受管 HTTP 调用读取当前发布值，无需重启插件。调小该值可能使插件上传或转发返回 413，且不发送出站请求。
+调高上限会增加大请求的内存占用，不代表扩大模型 Token 上限、响应缓冲或整个进程的内存预算
 
 ### 智能调度
 

docs/architecture.md

@@ -197,6 +197,10 @@ flowchart LR
   子调用保留显式改写，在目标 Key 作用域重算默认值；Key 限额、Fast 和模型超时的覆盖绑定原 Key，同级调用互不改写。
   原生 Responses WebSocket 每轮从当前快照重新认证，保留握手的显式改写；已经开始的执行不随全局发布改变。具体字段与来源合同见 [SDK 中间件](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#http-入口)
 
+- **正文预算随执行冻结**：Core 将有效正文预算传给路由、调度、重试、attempt 和上游适配器；显式 Key 的模型预检与执行共用绑定快照。
+  Runtime 优先读取有效请求设置或执行预算，其余调用读取当前发布配置，注册准备使用编译中的预算。
+  组合根在激活 Core 前绑定弱快照读取句柄，不延长发布集合寿命；Host 仅检查调用方传入的实际出站正文预算
+
 #### 调用资源与状态
 
 - **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。HTTP 与模型 request/attempt 中间件在首个流式响应头交付后，由正文 owner 管理取消与终态，不沿用初始处理期限；网络操作与模型执行各自计时。