Run plugins in the tw-plugin sandbox

crates/tw-gateway/Cargo.toml

@@ -45,6 +45,9 @@ rand = { workspace = true }
 uuid = { workspace = true }
 tw-guard = { workspace = true }
 tw-dialect = { workspace = true }
+# 脚本插件的沙箱。**编它要一个能出 wasm 的 clang**：依赖它的只能是网关和 twcore
+# （tw-plugin 的 tests/boundary.rs 守着）
+tw-plugin = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

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

@@ -1,8 +1,9 @@
 //! 网关要从插件运行时那里拿到的东西：把一份源码编成插件，读出它的 manifest。
 //!
-//! **这里是一道接缝**：沙箱在 `tw-plugin` 里，网关只认这个 trait。真的运行时接上之前，
-//! 由 [`Unavailable`] 顶着 —— 每个插件都「加载不了」，有一个就拒一个请求（出错时
-//! 拒绝是出厂的做法），而不是悄悄放过。测试拿一个假的引擎接在这里。
+//! **这里是一道接缝**：沙箱在 `tw-plugin` 里（接上它的是 [`crate::plugin::sandbox`]），
+//! 网关只认这个 trait。没有运行时可用时由 [`Unavailable`] 顶着 —— 每个插件都「加载
+//! 不了」，有一个就拒一个请求（出错时拒绝是出厂的做法），而不是悄悄放过。测试拿一个
+//! 假的引擎接在这里。
 
 use std::sync::Arc;
 

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

@@ -3,7 +3,7 @@
 //! 插件是 JavaScript，**只在沙箱里跑**（`tw-plugin`，Wasmtime）。网关这一侧分三块：
 //!
 //! - [`engine`]：网关要从运行时那里拿到的东西 —— 把一份源码编成一个插件，读出它的
-//!   manifest。运行时还没接上时由一个替身顶着，所有插件都是「加载不了」；
+//!   manifest。真的运行时在 [`sandbox`]，测试可以换一个假的；
 //! - [`host`]：一个编好的插件能做什么（跑请求钩子、回答钩子），数据面调它；
 //! - [`set`]：跟着配置一起换的那一份 —— 每个配置了的插件此刻的样子（能跑、文件
 //!   变了、加载出错）、范围、出错时怎么办，以及跨重载存活的计数和日志。
@@ -35,6 +35,7 @@ pub mod load;
 pub mod pool;
 pub mod reply;
 pub mod request;
+pub mod sandbox;
 pub mod set;
 pub mod trial;
 pub mod view;
@@ -94,10 +95,10 @@ impl crate::AppState {
     }
 }
 
-/// 这个进程用的插件运行时。**沙箱还没接上**：在那之前每个插件都「加载不了」，
-/// 管得着的请求照它的 `on_error` 处置。
+/// 这个进程用的插件运行时：`tw-plugin` 的沙箱（见 [`sandbox`]）。**第一次编插件时
+/// 才真的起来**；起不来时每个插件都「加载不了」，管得着的请求照它的 `on_error` 处置。
 pub fn default_engine() -> std::sync::Arc<dyn Engine> {
-    std::sync::Arc::new(Unavailable::default())
+    std::sync::Arc::new(sandbox::Sandbox)
 }
 
 /// 出错却没说为什么。数据面总该给一句，这里只是不让通知空着

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

@@ -35,8 +35,9 @@ pub struct Pool {
     tx: OnceLock<Result<Mutex<mpsc::Sender<Job>>, String>>,
 }
 
-/// 一根线程的栈。沙箱里的调用要比默认的 2 MiB 深一些
-const STACK: usize = 8 * 1024 * 1024;
+/// 一根线程的栈。**跑沙箱的线程至少要 2 MiB**（wasm 自己最多用 1 MiB，外面还有宿主
+/// 那一侧的调用），这里明着给足，不靠平台的默认值（Windows 上只有 1 MiB）
+pub(crate) const STACK: usize = 8 * 1024 * 1024;
 
 impl Pool {
     /// `threads` 根线程，最多 `queue` 个任务在跑或者排着。

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

@@ -531,6 +531,11 @@ impl Chain {
                         changed = true;
                         self.stages[i].counts.dropped += 1;
                     }
+                    // 交回一个空数组也是去掉
+                    Ok(ToolCallOutcome::Replace(vals)) if vals.is_empty() => {
+                        changed = true;
+                        self.stages[i].counts.dropped += 1;
+                    }
                     Ok(ToolCallOutcome::Replace(vals)) => {
                         // 原样交回来的一个调用就是没改
                         if let [one] = vals.as_slice()
@@ -653,7 +658,8 @@ impl Drop for Chain {
     }
 }
 
-/// 交回来的调用和交出去的一样（id、名字、参数都没变）
+/// 交回来的调用和交出去的一样（id、名字、参数都没变）。参数按 JavaScript 的眼光比
+/// （[`tw_plugin::js_equal`]）：进出一趟 JS 的 `1.0` 回来是 `1`，那不算改
 fn same_call(v: &Value, given: &Value) -> bool {
     let o = match v.as_object() {
         Some(o) => o,
@@ -662,7 +668,10 @@ fn same_call(v: &Value, given: &Value) -> bool {
     o.keys()
         .all(|k| matches!(k.as_str(), "id" | "name" | "input"))
         && o.get("name") == given.get("name")
-        && o.get("input") == given.get("input")
+        && match (o.get("input"), given.get("input")) {
+            (Some(a), Some(b)) => tw_plugin::js_equal(a, b),
+            (a, b) => a == b,
+        }
         && o.get("id").is_none_or(|id| Some(id) == given.get("id"))
 }
 

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

@@ -328,25 +328,39 @@ async fn a_replaced_tool_call_is_written_whole_and_later_blocks_move_up() {
 #[tokio::test]
 async fn dropping_every_tool_call_ends_the_turn_instead_of_waiting_for_results() {
     let input = anthropic_stream();
-    let mut s = Stream::new(
-        chain_of(
-            Dialect::Anthropic,
-            vec![tools(|_| ToolCallOutcome::Drop)],
-            &json!({}),
-        )
-        .await,
-        Framing::Sse,
-    );
-    let (out, _) = run(&mut s, &input, 4096).await;
-    let blocks = anthropic_blocks(&out);
-    let idx: Vec<u64> = blocks.iter().map(|b| b.0).collect();
-    assert_eq!(idx, [0, 1, 2]);
-    assert_eq!(blocks[2], (2, "text".into(), "bye".into()));
-    let stop = frames(&out)
-        .into_iter()
-        .find(|(_, v)| v["type"] == "message_delta")
-        .unwrap();
-    assert_eq!(stop.1["delta"]["stop_reason"], "end_turn");
+    // 交回 null 和交回空数组是同一个意思
+    for drop in [
+        tools(|_| ToolCallOutcome::Drop),
+        tools(|_| ToolCallOutcome::Replace(Vec::new())),
+    ] {
+        let mut s = Stream::new(
+            chain_of(Dialect::Anthropic, vec![drop], &json!({})).await,
+            Framing::Sse,
+        );
+        let (out, _) = run(&mut s, &input, 4096).await;
+        let blocks = anthropic_blocks(&out);
+        let idx: Vec<u64> = blocks.iter().map(|b| b.0).collect();
+        assert_eq!(idx, [0, 1, 2]);
+        assert_eq!(blocks[2], (2, "text".into(), "bye".into()));
+        let stop = frames(&out)
+            .into_iter()
+            .find(|(_, v)| v["type"] == "message_delta")
+            .unwrap();
+        assert_eq!(stop.1["delta"]["stop_reason"], "end_turn");
+    }
+}
+
+/// 进出一趟 JavaScript 的调用：`2.0` 回来是 `2`，超过 2^53 的整数丢了精度 —— 这还是
+/// 原来那个调用，不能算改过
+#[test]
+fn a_call_that_went_through_javascript_unchanged_is_the_same_call() {
+    let given = json!({"id": "t1", "name": "Read",
+                       "input": {"limit": 2.0, "seed": 12345678901234567890u64}});
+    let back = json!({"id": "t1", "name": "Read",
+                      "input": {"seed": 12345678901234567000u64, "limit": 2}});
+    assert!(same_call(&back, &given));
+    let other = json!({"id": "t1", "name": "Read", "input": {"limit": 3, "seed": 1}});
+    assert!(!same_call(&other, &given));
 }
 
 #[tokio::test]

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

@@ -0,0 +1,219 @@
+//! 真的插件运行时：`tw-plugin` 的沙箱（QuickJS 跑在 Wasmtime 里）接到 [`Engine`] 这道
+//! 接缝上。
+//!
+//! **只是一层把类型对上的适配**：manifest、钩子的结局、错误、日志，一样一样换成网关的
+//! 那一份，语义一点不改 —— 怎么跑、上限多少都是 `tw-plugin` 的事，视图、权限、写回
+//! 都是网关的事。
+//!
+//! 运行时一个进程一份，**第一次编插件时才起**：起它要加载沙箱、起一个计时线程，没装
+//! 插件的进程（以及绝大多数测试）不该为它付这个钱。起不来（比如地址空间受限的机器）
+//! 就和没有运行时一样：每个插件都加载不了，原因照实说，管得着的请求按 `on_error` 处置。
+
+use std::sync::{Arc, OnceLock};
+
+use serde_json::Value;
+
+use crate::plugin::engine::{Engine, Hooks, LoadError, Manifest, SettingSpec};
+use crate::plugin::host::{
+    Invocation, PluginHost, ReplyHost, RequestOutcome, RunError, ToolCallOutcome,
+};
+use crate::plugin::set::{LogLine, Scope};
+
+/// 进程里那一份运行时。起不来时是起不来的原因
+fn runtime() -> Result<&'static tw_plugin::Runtime, LoadError> {
+    static RT: OnceLock<Result<tw_plugin::Runtime, String>> = OnceLock::new();
+    RT.get_or_init(|| {
+        tw_plugin::Runtime::new(tw_plugin::Limits::default()).map_err(|e| e.to_string())
+    })
+    .as_ref()
+    .map_err(|e| LoadError::Engine(e.clone()))
+}
+
+/// `tw-plugin` 的沙箱。生产上用的就是它（见 [`crate::plugin::default_engine`]）。
+#[derive(Debug, Default, Clone, Copy)]
+pub struct Sandbox;
+
+impl Engine for Sandbox {
+    fn load(&self, source: &[u8]) -> Result<Arc<dyn PluginHost>, LoadError> {
+        let rt = runtime()?;
+        // 编译要在沙箱里跑一遍模块顶层，**调用方的线程栈未必够**：插件是在换配置的那一路
+        // 上编的，那可能是主线程（Windows 上只有 1 MiB）。在一根栈给足了的线程上编 ——
+        // 编插件只在换配置、装插件时发生，多起一根线程不算什么
+        let plugin = std::thread::scope(|s| {
+            let compiling = std::thread::Builder::new()
+                .name("tw-plugin-load".into())
+                .stack_size(crate::plugin::pool::STACK)
+                .spawn_scoped(s, || rt.load(source).map_err(load_error))
+                .map_err(|e| {
+                    LoadError::Engine(format!("cannot start a thread to compile the plugin: {e}"))
+                })?;
+            compiling
+                .join()
+                .unwrap_or_else(|_| Err(LoadError::Engine("compiling the plugin crashed".into())))
+        })?;
+        let manifest = manifest(plugin.manifest());
+        Ok(Arc::new(Host { plugin, manifest }))
+    }
+}
+
+/// 编好的一个插件。克隆、跨线程共享都便宜（`tw_plugin::Plugin` 里是一个 `Arc`）
+struct Host {
+    plugin: tw_plugin::Plugin,
+    /// 换成网关那一份的 manifest。编的时候换一次，之后每次问都是它
+    manifest: Manifest,
+}
+
+impl PluginHost for Host {
+    fn manifest(&self) -> &Manifest {
+        &self.manifest
+    }
+
+    fn sha256(&self) -> [u8; 32] {
+        self.plugin.sha256()
+    }
+
+    fn on_request(&self, view: Value, ctx: Value) -> Invocation<RequestOutcome> {
+        invocation(self.plugin.on_request(view, ctx), |o| match o {
+            tw_plugin::RequestOutcome::Unchanged => RequestOutcome::Unchanged,
+            tw_plugin::RequestOutcome::Changed(v) => RequestOutcome::Changed(v),
+            tw_plugin::RequestOutcome::Rejected(r) => RequestOutcome::Rejected(r),
+        })
+    }
+
+    fn reply(&self, ctx: Value) -> Result<Box<dyn ReplyHost>, RunError> {
+        match self.plugin.reply(ctx) {
+            Ok(r) => Ok(Box::new(Reply(r))),
+            Err(e) => Err(run_error(e)),
+        }
+    }
+}
+
+/// 一个回答的实例
+struct Reply(tw_plugin::Reply);
+
+impl ReplyHost for Reply {
+    fn on_text(&mut self, text: &str) -> Invocation<Option<String>> {
+        invocation(self.0.on_text(text), |t| t)
+    }
+
+    fn on_text_end(&mut self) -> Invocation<Option<String>> {
+        invocation(self.0.on_text_end(), |t| t)
+    }
+
+    fn on_tool_call(&mut self, call: Value) -> Invocation<ToolCallOutcome> {
+        invocation(self.0.on_tool_call(call), |o| match o {
+            tw_plugin::ToolCallOutcome::Unchanged => ToolCallOutcome::Unchanged,
+            tw_plugin::ToolCallOutcome::Replace(calls) => ToolCallOutcome::Replace(calls),
+            tw_plugin::ToolCallOutcome::Drop => ToolCallOutcome::Drop,
+        })
+    }
+}
+
+// ── 换类型 ───────────────────────────────────────────────────────
+
+fn invocation<T, U>(inv: tw_plugin::Invocation<T>, f: impl FnOnce(T) -> U) -> Invocation<U> {
+    Invocation {
+        result: inv.result.map(f).map_err(run_error),
+        logs: inv.logs.into_iter().map(log_line).collect(),
+        cpu: inv.cpu,
+    }
+}
+
+fn log_line(l: tw_plugin::LogLine) -> LogLine {
+    LogLine {
+        level: match l.level {
+            tw_plugin::LogLevel::Log => tw_api::PluginLogLevel::Log,
+            tw_plugin::LogLevel::Info => tw_api::PluginLogLevel::Info,
+            tw_plugin::LogLevel::Warn => tw_api::PluginLogLevel::Warn,
+            tw_plugin::LogLevel::Error => tw_api::PluginLogLevel::Error,
+        },
+        text: l.text,
+    }
+}
+
+fn run_error(e: tw_plugin::RunError) -> RunError {
+    match e {
+        tw_plugin::RunError::CpuLimit => RunError::CpuLimit,
+        tw_plugin::RunError::MemoryLimit => RunError::MemoryLimit,
+        tw_plugin::RunError::OutputLimit => RunError::OutputLimit,
+        tw_plugin::RunError::Threw { message, stack } => RunError::Threw { message, stack },
+        tw_plugin::RunError::BadOutput(d) => RunError::BadOutput(d),
+        tw_plugin::RunError::Trap(d) => RunError::Trap(d),
+    }
+}
+
+fn load_error(e: tw_plugin::LoadError) -> LoadError {
+    match e {
+        tw_plugin::LoadError::TooLarge => LoadError::TooLarge,
+        tw_plugin::LoadError::Syntax {
+            message,
+            line,
+            column,
+        } => LoadError::Syntax {
+            message,
+            line,
+            column,
+        },
+        tw_plugin::LoadError::Manifest(d) => LoadError::Manifest(d),
+        tw_plugin::LoadError::UnsupportedApi(api) => LoadError::UnsupportedApi(api),
+        tw_plugin::LoadError::Engine(d) => LoadError::Engine(d),
+    }
+}
+
+fn permission(p: tw_plugin::Permission) -> tw_api::Permission {
+    match p {
+        tw_plugin::Permission::System => tw_api::Permission::System,
+        tw_plugin::Permission::Messages => tw_api::Permission::Messages,
+        tw_plugin::Permission::Tools => tw_api::Permission::Tools,
+        tw_plugin::Permission::Params => tw_api::Permission::Params,
+        tw_plugin::Permission::ReplyText => tw_api::Permission::ReplyText,
+        tw_plugin::Permission::ReplyToolCalls => tw_api::Permission::ReplyToolCalls,
+    }
+}
+
+fn manifest(m: &tw_plugin::Manifest) -> Manifest {
+    let granted: Vec<tw_api::Permission> = m.permissions.iter().copied().map(permission).collect();
+    Manifest {
+        name: m.name.clone(),
+        api: m.api,
+        description: m.description.clone(),
+        // 网关这边按 `Permission::ALL` 的顺序排
+        permissions: tw_api::Permission::ALL
+            .iter()
+            .copied()
+            .filter(|p| granted.contains(p))
+            .collect(),
+        scope: Scope {
+            clients: m.scope.clients.clone(),
+            models: m.scope.models.clone(),
+            upstreams: m.scope.upstreams.clone(),
+        },
+        reply_mode: match m.reply_mode {
+            tw_plugin::ReplyMode::Block => tw_api::ReplyMode::Block,
+            tw_plugin::ReplyMode::Stream => tw_api::ReplyMode::Stream,
+        },
+        settings: m
+            .settings
+            .iter()
+            .map(|s| SettingSpec {
+                key: s.key.clone(),
+                kind: match s.kind {
+                    tw_plugin::SettingKind::String => tw_api::SettingKind::String,
+                    tw_plugin::SettingKind::Number => tw_api::SettingKind::Number,
+                    tw_plugin::SettingKind::Boolean => tw_api::SettingKind::Boolean,
+                },
+                label: s.label.clone(),
+                default: s.default.clone(),
+            })
+            .collect(),
+        hooks: Hooks {
+            request: m.hooks.request,
+            reply_text: m.hooks.reply_text,
+            reply_text_end: m.hooks.reply_text_end,
+            tool_call: m.hooks.tool_call,
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests;