Run request hooks on every body sent upstream; cap live reply instances

crates/tw-api/msg-codes.txt

@@ -282,6 +282,7 @@ gw.output_limit.withheld
 gw.plugin.answer_unreadable
 gw.plugin.api
 gw.plugin.bad_output
+gw.plugin.cannot_read
 gw.plugin.changed
 gw.plugin.cpu_limit
 gw.plugin.engine
@@ -296,6 +297,7 @@ gw.plugin.output_limit
 gw.plugin.permission_violation
 gw.plugin.reason passthrough
 gw.plugin.rejected
+gw.plugin.reply_busy
 gw.plugin.reply_failed
 gw.plugin.request_failed
 gw.plugin.request_unreadable

crates/tw-gateway/src/client_api.rs

@@ -92,6 +92,21 @@ impl ClientApi {
             || (p.contains("/models/") && p.ends_with(":countTokens"))
     }
 
+    /// 请求体和生成回答**同一种形状**、回来的却不是一次回答的接口：数 token（见
+    /// [`ClientApi::counts_tokens`]），Responses 的压缩和数 token（`/responses/compact`、
+    /// `/responses/input_tokens`，Codex 后端的 `/backend-api/codex/responses/compact`）。
+    ///
+    /// 插件的请求钩子照样看得懂它们（见 [`crate::plugin::request::Shape`]）
+    pub fn like_generation(path: &str) -> bool {
+        if Self::counts_tokens(path) {
+            return true;
+        }
+        let p = path.trim_end_matches('/');
+        let tail = p.strip_prefix("/v1").unwrap_or(p);
+        matches!(tail, "/responses/compact" | "/responses/input_tokens")
+            || p == "/backend-api/codex/responses/compact"
+    }
+
     /// 转换库里对应的格式
     pub fn dialect(&self) -> Dialect {
         match self {
@@ -252,6 +267,27 @@ mod tests {
         }
     }
 
+    #[test]
+    fn counting_and_compacting_take_a_generation_shaped_body() {
+        for (path, like) in [
+            ("/v1/messages/count_tokens", true),
+            ("/v1beta/models/gemini-2.5-pro:countTokens", true),
+            ("/v1/responses/compact", true),
+            ("/responses/compact/", true),
+            ("/v1/responses/input_tokens", true),
+            ("/backend-api/codex/responses/compact", true),
+            // 生成回答本身不算：它是「生成」那一类
+            ("/v1/messages", false),
+            ("/v1/responses", false),
+            ("/v1/embeddings", false),
+            ("/v1/completions", false),
+            ("/v1/responses/resp_1/cancel", false),
+            ("/v1beta/models/gemini-embedding-001:embedContent", false),
+        ] {
+            assert_eq!(ClientApi::like_generation(path), like, "{path}");
+        }
+    }
+
     #[test]
     fn a_path_we_do_not_know_is_not_guessed() {
         for path in ["/v1/models", "/v1/files", "/healthz", "/v1/messagesx", "/"] {

crates/tw-gateway/src/plugin/mod.rs

@@ -22,9 +22,12 @@
 //! - [`request`]：请求钩子。排在路由之后，**每发往一个上游跑一次**（契约附录二的 I7、
 //!   I8）：按这一次的客户端、发出去的模型和上游挑插件，从客户端的原话起改；换上游从
 //!   原话重来，同一家重发不重跑。改过的请求再过一遍内容审查，然后才转换格式、脱敏。
+//!   **发往上游的每个请求体都过**：数 token、Responses 的压缩也改，插件看不懂的接口按
+//!   插件的 `on_error` 处置（见 [`request::Shape`]）。
 //! - [`reply`]：回答钩子。排在格式转换之后、工具调用审查和输出长度之前（I7）——
 //!   这两道防护看的就是插件改过的那一版。
 //! - [`pool`]：插件调用都是阻塞的、吃 CPU 的，放在专用线程池上跑，不占 tokio 的线程。
+//!   同时活着的回答实例也在这里限数（[`pool::MAX_LIVE_REPLIES`]）。
 //! - [`trial`]：对着存下来的请求和回答试跑一个插件。
 //!
 //! [`defaults`] 是随 core 一起发的那几个插件（清单和源码）；[`manifests`] 是编过的插件的

crates/tw-gateway/src/plugin/pool.rs

@@ -1,15 +1,29 @@
-//! 跑插件的专用线程池。
+//! 跑插件的专用线程池，和回答实例的名额。
 //!
 //! 插件调用是阻塞的、吃 CPU 的（一次请求钩子最多跑两百毫秒）。放在 tokio 的工作线程
 //! 上调，几个慢插件就能把整个数据面的线程占满 —— 那时连不走插件的请求也一起卡住。
 //! 所以一律交给这里：几根自己的线程，**排队的数量有上限**，满了的话调用方异步地等，
 //! 不占着 tokio 的线程。
 //!
 //! 线程**第一次用到时才起**：绝大多数用户一个插件都没装，不该为它多几根闲着的线程。
+//!
+//! # 回答实例的名额
+//!
+//! 回答钩子的实例从回答开始活到回答结束（不变式 I3），一个最多占
+//! `tw_plugin::Limits::reply_memory`（64 MiB）。流式的回答一开就是几分钟，同时在流的
+//! 回答越多，活着的实例就越多 —— 不设上限的话，内存跟着并发的流一起涨。所以整个进程
+//! 同时活着的回答实例最多 [`MAX_LIVE_REPLIES`] 个：起实例之前先拿一个名额
+//! （[`Pool::reply_slot`]）。**拿不到不等**：回答已经到了，等一个名额就是让客户端干等，
+//! 满了就按那个插件的 `on_error` 处置 —— 拒绝是这个请求失败，跳过是这次回答绕过它。
+//! 名额跟着实例走（[`Slot`]），实例扔掉的那一刻还回来。
 
 use std::sync::mpsc;
 use std::sync::{Arc, Mutex, OnceLock};
 
+/// 同时活着的回答实例最多这么多个（见模块说明）。按一个实例 64 MiB 的上限算，最坏
+/// 2 GiB；实际的插件一个实例多半不到 1 MiB
+pub const MAX_LIVE_REPLIES: usize = 32;
+
 type Job = Box<dyn FnOnce() + Send>;
 
 /// 池子坏了：任务 panic 了，或者线程起不来。
@@ -32,24 +46,64 @@ pub struct Pool {
     threads: usize,
     /// 正在跑的加排着的，最多这么多
     permits: Arc<tokio::sync::Semaphore>,
+    /// 回答实例的名额（见模块说明）
+    replies: Arc<tokio::sync::Semaphore>,
+    /// 名额一共几个
+    reply_cap: usize,
     tx: OnceLock<Result<Mutex<mpsc::Sender<Job>>, String>>,
 }
 
+/// 一个回答实例占着的名额（见 [`Pool::reply_slot`]）。**和实例放在一起**：实例在哪儿被
+/// 扔掉 —— 回答收尾、插件出错被拿掉、客户端走了、插件线程上 panic —— 名额就在哪儿还回去
+#[derive(Debug)]
+pub struct Slot {
+    _permit: tokio::sync::OwnedSemaphorePermit,
+}
+
 /// 一根线程的栈。**跑沙箱的线程至少要 2 MiB**（wasm 自己最多用 1 MiB，外面还有宿主
 /// 那一侧的调用），这里明着给足，不靠平台的默认值（Windows 上只有 1 MiB）
 pub(crate) const STACK: usize = 8 * 1024 * 1024;
 
 impl Pool {
-    /// `threads` 根线程，最多 `queue` 个任务在跑或者排着。
+    /// `threads` 根线程，最多 `queue` 个任务在跑或者排着；回答实例的名额是
+    /// [`MAX_LIVE_REPLIES`] 个。
     pub fn new(threads: usize, queue: usize) -> Self {
+        Self::with_replies(threads, queue, MAX_LIVE_REPLIES)
+    }
+
+    /// 同 [`Pool::new`]，回答实例的名额是 `replies` 个。**测试用**：拿一个小数，不用开几十
+    /// 条流就能把名额占满
+    pub fn with_replies(threads: usize, queue: usize, replies: usize) -> Self {
         let threads = threads.max(1);
         Self {
             threads,
             permits: Arc::new(tokio::sync::Semaphore::new(queue.max(threads))),
+            replies: Arc::new(tokio::sync::Semaphore::new(replies)),
+            reply_cap: replies,
             tx: OnceLock::new(),
         }
     }
 
+    /// 给一个回答实例拿一个名额。**满了是 `None`，不等**（理由见模块说明）
+    pub fn reply_slot(&self) -> Option<Slot> {
+        self.replies
+            .clone()
+            .try_acquire_owned()
+            .ok()
+            .map(|p| Slot { _permit: p })
+    }
+
+    /// 回答实例的名额一共几个
+    pub fn reply_cap(&self) -> usize {
+        self.reply_cap
+    }
+
+    /// 此刻活着的回答实例（占着的名额）
+    pub fn live_replies(&self) -> usize {
+        self.reply_cap
+            .saturating_sub(self.replies.available_permits())
+    }
+
     /// 按机器的核数定：至少两根，最多八根；排队的是线程数的四倍。
     pub fn default_size() -> Self {
         let cores = std::thread::available_parallelism().map_or(2, |n| n.get());
@@ -142,6 +196,21 @@ mod tests {
         assert_eq!(pool.run(|| 7).await, Ok(7));
     }
 
+    #[test]
+    fn reply_slots_run_out_without_waiting_and_come_back_when_dropped() {
+        let pool = Pool::with_replies(1, 1, 2);
+        let a = pool.reply_slot().expect("the first slot");
+        let b = pool.reply_slot().expect("the second slot");
+        assert_eq!(pool.live_replies(), 2);
+        assert!(pool.reply_slot().is_none(), "a third slot past the cap");
+        drop(a);
+        assert_eq!(pool.live_replies(), 1);
+        let c = pool.reply_slot().expect("the slot that came back");
+        drop((b, c));
+        assert_eq!(pool.live_replies(), 0);
+        assert_eq!(Pool::default_size().reply_cap(), MAX_LIVE_REPLIES);
+    }
+
     #[tokio::test]
     async fn more_calls_than_threads_wait_their_turn() {
         let pool = Arc::new(Pool::new(2, 2));