fix(plugins): retain request middleware resource streams (#492)

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

@@ -69,6 +69,7 @@ pub(super) async fn invoke(
         },
         maximum_timeout,
     );
+    call_context.resource_stream = true;
     call_context.request = crate::adapter::request_context(context.extension_scope().request());
     call_context.request_id = Some(context.request_id().as_str().to_owned());
     call_context.attempt_id = context

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

@@ -214,6 +214,8 @@ async fn nested_sdk_middleware_waits_beyond_budget_and_can_read_response_callbac
         .expect("host next must not consume either plugin's processing budget");
         let (_, status, _, mut body, _) = response.into_parts();
         assert_eq!(status, 200);
+        // 正文消费可晚于初始处理预算，惰性 body_read 仍须归属原资源流
+        tokio::time::sleep(Duration::from_millis(900)).await;
         // 正文改写还会发起新的 body_read，证明 callback 有效期也随暂停恢复
         let frame = body.next_frame().await.unwrap().unwrap();
         assert_eq!(

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

@@ -195,32 +195,34 @@ async fn managed_next_can_outlive_the_initial_budget_with_or_without_resource_st
 
 #[tokio::test(start_paused = true)]
 async fn response_processing_uses_only_the_budget_left_before_next() {
-    let plugin = MiddlewarePlugin::new(
-        &middleware_contributions(),
-        |call: RequestCall| async move {
-            tokio::time::sleep(Duration::from_millis(200)).await;
-            let response = call.next.run(call.request).await?;
-            tokio::time::sleep(Duration::from_millis(850)).await;
-            Ok(response)
-        },
-    )
-    .unwrap();
-    let (mut host, task) = start_session(plugin).await;
-    send_middleware_call(&mut host, 1, false).await;
-    let next = receive_next(&mut host, 1).await;
-    tokio::time::advance(Duration::from_secs(10)).await;
-    reply_next(&mut host, next).await;
-    assert!(matches!(
-        receive(&mut host).await.message,
-        Message::Error {
-            id: 1,
-            error: PluginFault {
-                code: ErrorCode::Timeout,
-                ..
+    for resource_stream in [false, true] {
+        let plugin = MiddlewarePlugin::new(
+            &middleware_contributions(),
+            |call: RequestCall| async move {
+                tokio::time::sleep(Duration::from_millis(200)).await;
+                let response = call.next.run(call.request).await?;
+                tokio::time::sleep(Duration::from_millis(850)).await;
+                Ok(response)
+            },
+        )
+        .unwrap();
+        let (mut host, task) = start_session(plugin).await;
+        send_middleware_call(&mut host, 1, resource_stream).await;
+        let next = receive_next(&mut host, 1).await;
+        tokio::time::advance(Duration::from_secs(10)).await;
+        reply_next(&mut host, next).await;
+        assert!(matches!(
+            receive(&mut host).await.message,
+            Message::Error {
+                id: 1,
+                error: PluginFault {
+                    code: ErrorCode::Timeout,
+                    ..
+                }
             }
-        }
-    ));
-    shutdown(&mut host, task).await;
+        ));
+        shutdown(&mut host, task).await;
+    }
 }
 
 #[tokio::test(start_paused = true)]

docs/architecture.md

@@ -199,7 +199,7 @@ flowchart LR
 
 #### 调用资源与状态
 
-- **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。受管流建立后不沿用握手的绝对期限，网络操作与模型执行各自计时。
+- **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。HTTP 与模型 request/attempt 中间件在首个流式响应头交付后，由正文 owner 管理取消与终态，不沿用初始处理期限；网络操作与模型执行各自计时。
   Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停；Runtime 同时归还执行名额，返回后取得名额再以剩余处理预算恢复原调用。
   执行调度由 Runtime 统一负责，准入和恢复等待使用独立期限；初次排队、总在途上下文和普通回调各有上限；`next` 回调按在途上限独立计数。
   流、取消确认与消费方共同持有同一在途租约，执行名额只覆盖首结果前的处理；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。