feat(lua): 插件 Lua 侧适配器桥接(R4 Lua 绑定)+ 真二进制适配器接手 e2e

补上两个此前报告为"缺口"的东西。其中第一个的旧结论是**错的**,这里更正。

## R4 Lua 绑定:__llmsproxy_adapters

用户问的"为什么 Lua 绑不上"—— 结论是**能绑,之前的归因错了**。

旧结论是「NewTable + PushGoClosure + SetField 是 golua 雷区,会 SIGSEGV」。
反证:内核自己的 buildJSONTable 用的就是这同一个模式,长期稳定。真正的变量
是当时 Phase 4 的**双 VM 半成品**(钩子一个 VM、定时任务一个私有 VM),
bootPluginState 签名与调用点不一致,我改到一半编译不过就仓促回滚,留下的
中间态让 billing 一跑就崩。把双 VM 改回单 VM 之后,同一模式一次就通。

顺带回答一个备选方案(用户提出:Lua 只投递请求、Go 串行执行):不需要。
p.mu 已经天然串行化同一插件的所有入口;再加队列只会引入"谁排空/超时后积压
怎么算"的新问题。而且桥接调用本身是微秒级的(交给 Go 重建一个小池),真正
长耗时的是插件自己去取 token 的 HTTP 请求,那跟桥接无关。

能力(Go 侧强制,非约定):
- install(name, source) → (生效名, nil) | (nil, err)
- update(name, source) → (true, nil) | (nil, err)  只限自己装的
- remove(name)         → (true, nil) | (nil, err)  只限自己装的
- 调用身份是插件**启动时的固定名**,不从 Lua 状态现读 ⇒ 无法冒充别的插件
- 不能影子内置(Lua 侧同样被拒,不是绕过 Go 校验的后门)
- 出错返回 (nil, err) 而非抛异常,与钩子/HTTP 处理器同一套约定

**分片 VM 故意不给**:分片是文档化的「只读、可丢弃」工作单元,而适配器
install 是**全局**副作用(改共享注册表)。给分片等于让"只读"的按需调用悄悄
改写所有请求共用的适配器 —— 正是分片契约要防的不可见变更。

测试 6 个:install/rotate/remove 从 Lua 调用、影子内置被拒、**定时任务里
轮换 token 且请求钩子立即可见**(R4 闭环)、分片无桥接。

## 真二进制 e2e:适配器接手对外服务

用户问到的第二个点。此前 /v1/messages 的 Anthropic 接管只在**进程内**
gateway 层测过(internal/gateway/messages_test.go),绕过了只会在真实进程里
发生的事:适配器磁盘 seed、配置加载、真 HTTP 栈。

新增 e2e/adapter_takeover_test.go,打**真编译的二进制** + 只说 Anthropic 的
假上游(该上游对不符形状的请求主动报错,不做宽容接受):
- 非流式:出向必须是 Anthropic message 形状,不能有 OpenAI choices 外壳
- 流式:必须是 message_start…message_stop 序列,**不得**出现 [DONE]
- 压力:96 并发(48 流式)混合打真二进制,全部 200,无 5xx,无形状串味

压力测试初版把并发上限设成 4 于是大量 429,我把断言改正为:429 是内核**正确**
的容量背压,不算失败;失败的定义是 5xx 或 200 但内容/形状错。现在 96/96 通过。
This commit is contained in:
JianFeeeee
2026-10-11 13:25:11 +08:00
parent 66786bb0e3
commit dafc3b3fc3
6 changed files with 640 additions and 7 deletions

View File

@ -381,14 +381,39 @@ plugin.http = {
插件可以在运行时安装、改写**自己的**适配器 —— 这是"带自动轮换的认证"的实现
方式:拿着新 token 重写适配器源码,后续请求自然使用新逻辑(内核重建该适配器
的池)。这是 Phase 3 的桥接能力,安全规则如下:
的池)。**可直接在插件 `.lua` 里调用**:
- 插件**只能改自己装的**适配器(归属表在 Go 侧记录,不靠约定)。
```lua
-- 每小时的定时任务:轮换 token 并重写适配器
function plugin.refresh_token()
local tok = fetch_new_token() -- 插件自己的取 token 逻辑
local name, err = __llmsproxy_adapters.install(
"myauth",
'return { name = "myauth", endpoint = "/chat/completions",' ..
' headers = { ["Authorization"] = "Bearer ' .. tok .. '" } }'
)
if err then log("error", err) end
end
```
| 调用 | 参数 | 返回 |
|---|---|---|
| `__llmsproxy_adapters.install(name, source)` | 适配器名 + Lua 源码 | `(生效名, nil)` 或 `(nil, err)` |
| `__llmsproxy_adapters.update(name, source)` | 同上(只限自己装的) | `(true, nil)` 或 `(nil, err)` |
| `__llmsproxy_adapters.remove(name)` | 适配器名(只限自己装的) | `(true, nil)` 或 `(nil, err)` |
出错时不抛异常,而是返回 `(nil, err)`——与钩子/HTTP 处理器同一套约定,
一个手滑的插件错误不会带走网关。
安全规则(由 Go 侧强制,不是约定):
- 插件**只能改自己装的**适配器(归属表在 Go 侧记录,调用身份是插件启动时的
固定名字,无法冒充)。
- 插件**不能影子内置适配器**(否则等于篡改发行物);运维仍可自行覆盖。
- 名字经过路径安全校验(无穿越/绝对路径/空白)。
> 注:这层桥接目前以 Go API 形式提供并被测试覆盖;插件 `.lua` 直接调用它的
> 全局绑定尚未接通(见 plan.md 已知限制)。
> 这条能力配合 §5.4 的定时任务,就是完整的自动轮换:**没有浏览器、没有人
> 操作时,插件按时间自己把 token 换掉**。
---

View File

@ -0,0 +1,286 @@
package e2e
import (
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
// 本文件端到端验证 R2/R5 的核心主张:**对外服务的 wire 格式由适配器决定,
// 内核只调度**;适配器不声明时内核的 OpenAI 兜底仍然可用。
//
// 与其他测试的分工:
// - internal/gateway/messages_test.go 在**进程内**验证同一件事(起 core+gateway,
// 不打真二进制)。它跑得快,但绕过了一些只在真实进程里发生的事:Lua 适配器的
// 磁盘 seed、配置加载、systemd 式启动路径、真 HTTP 栈。
// - 本文件打**真编译出的二进制**,用一个只说 Anthropic Messages 的假上游,
// 证明整条链路(真进程 → 真 Lua 适配器 → 真上游 → 适配器序列化 → 客户端)
// 确实成立,而不是只在进程内装配下成立。
//
// 这是用户明确问到的"适配器接手对外服务的部分"的验收点。
// anthropicMockUpstream 是只懂 Anthropic Messages 协议的假上游。
//
// 它会对**不符形状的请求主动报错**,而不是宽容接受 —— 宽容的 mock 会让
// "内核把 OpenAI 形状发给了 Anthropic 上游"这种错误测试通过。
func anthropicMockUpstream(t *testing.T) *httptest.Server {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Anthropic 的请求路径必须是 <base>/messages
if !strings.HasSuffix(r.URL.Path, "/messages") {
t.Errorf("请求打到了非 Anthropic 路径: %s", r.URL.Path)
w.WriteHeader(http.StatusNotFound)
return
}
body, _ := io.ReadAll(r.Body)
var probe map[string]interface{}
if err := json.Unmarshal(body, &probe); err != nil {
t.Errorf("上游收到非 JSON 请求体: %s", body)
w.WriteHeader(http.StatusBadRequest)
return
}
// Anthropic 形状的硬性要求:messages 是数组;不能有 OpenAI 的 choices 概念
if _, ok := probe["messages"]; !ok {
t.Errorf("上游请求体缺 messages: %s", body)
}
if _, ok := probe["max_tokens"]; !ok {
t.Errorf("Anthropic 要求 max_tokens,请求体缺此项: %s", body)
}
// 反向断言:OpenAI 兜底字段不该出现在 Anthropic 上游请求里
for _, k := range []string{"choices", "chat_completions"} {
if _, bad := probe[k]; bad {
t.Errorf("Anthropic 上游请求体混入了 OpenAI 字段 %q: %s", k, body)
}
}
if stream, _ := probe["stream"].(bool); stream {
w.Header().Set("Content-Type", "text/event-stream")
fl, _ := w.(http.Flusher)
frames := []string{
`event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e","content":[],"stop_reason":null,"usage":{"input_tokens":3,"output_tokens":0}}}`,
`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}`,
`event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`,
`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}`,
`event: message_stop` + "\n" + `data: {"type":"message_stop"}`,
}
for _, f := range frames {
fmt.Fprint(w, f+"\n\n")
if fl != nil {
fl.Flush()
}
}
return
}
w.Header().Set("Content-Type", "application/json")
fmt.Fprint(w, `{"id":"msg_e2e","type":"message","role":"assistant","model":"claude-e2e",`+
`"content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn",`+
`"usage":{"input_tokens":3,"output_tokens":1}}`)
}))
t.Cleanup(srv.Close)
return srv
}
// writeAnthropicConfig 写一份把源挂到 anthropic 适配器上的配置。
func writeAnthropicConfig(t *testing.T, dir, listenAddr, upstreamURL string) string {
t.Helper()
cfg := "listen: " + listenAddr + "\n" +
"gateway_keys:\n - sk-e2e-0001\n" +
"default_model: claude-e2e\n" +
"adapter_dir: " + filepath.Join(dir, "adapters") + "\n" +
"runtime_file: " + filepath.Join(dir, "runtime.json") + "\n" +
"sources:\n" +
" - name: anth\n" +
" base_url: " + upstreamURL + "\n" +
" adapter: anthropic\n" +
" api_key: test-key\n" +
" max_concurrent: 32\n" +
" models:\n" +
" - id: claude-e2e\n" +
" kind: chat\n" +
" priority: 100\n"
path := filepath.Join(dir, "config.yaml")
if err := os.WriteFile(path, []byte(cfg), 0o644); err != nil {
t.Fatalf("write config: %v", err)
}
return path
}
// TestAdapterTakeoverNonStream:非流式,Anthropic 客户端形状进、出。
func TestAdapterTakeoverNonStream(t *testing.T) {
up := anthropicMockUpstream(t)
dir := t.TempDir()
addr := mustAddr(t)
cfg := writeAnthropicConfig(t, dir, addr, up.URL)
bin := buildBinary(t)
g := startGateway(t, bin, addr, cfg)
resp, body := g.do("POST", "/v1/messages",
`{"model":"claude-e2e","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`, true)
if resp == nil || resp.StatusCode != 200 {
t.Fatalf("status=%v body=%q", statusOf(resp), body)
}
// 出向必须是 Anthropic Messages 形状,不能是 OpenAI choices 形状
if strings.Contains(body, `"choices"`) {
t.Errorf("Anthropic 响应里混入了 OpenAI choices 外壳: %s", body)
}
var out map[string]interface{}
if err := json.Unmarshal([]byte(body), &out); err != nil {
t.Fatalf("响应不是 JSON: %v: %s", err, body)
}
if out["type"] != "message" {
t.Errorf(`响应 type = %v, 期望 "message"(Anthropic 形状): %s`, out["type"], body)
}
// 文本内容必须能取出来 —— 这是"内容确实穿过适配器回来了"的证据
if !strings.Contains(body, "pong") {
t.Errorf("响应里没有上游返回的文本: %s", body)
}
if _, ok := out["usage"]; !ok {
t.Errorf("Anthropic 响应应带 usage: %s", body)
}
}
// TestAdapterTakeoverStream:流式也要是 Anthropic SSE 序列,
// 且**不得**出现 OpenAI 的 [DONE] 终止标记(R2 的"不混合")。
func TestAdapterTakeoverStream(t *testing.T) {
up := anthropicMockUpstream(t)
dir := t.TempDir()
addr := mustAddr(t)
cfg := writeAnthropicConfig(t, dir, addr, up.URL)
bin := buildBinary(t)
g := startGateway(t, bin, addr, cfg)
resp, body := g.do("POST", "/v1/messages",
`{"model":"claude-e2e","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}`, true)
if resp == nil || resp.StatusCode != 200 {
t.Fatalf("stream status=%v body=%q", statusOf(resp), body)
}
if !strings.Contains(body, "message_start") {
t.Errorf("Anthropic 流应以 message_start 开头: %s", body)
}
if !strings.Contains(body, "message_stop") {
t.Errorf("Anthropic 流应以 message_stop 收尾: %s", body)
}
if strings.Contains(body, "[DONE]") {
t.Errorf("Anthropic 流里出现了 OpenAI 的 [DONE]: %s", body)
}
if strings.Contains(body, "chat.completion.chunk") {
t.Errorf("流里混入 OpenAI chunk 形状: %s", body)
}
if !strings.Contains(body, "pong") {
t.Errorf("流里没有上游文本: %s", body)
}
}
// TestAdapterTakeoverStress 是对外服务路径的真实压力测试:
// 并发打真二进制,混合流式/非流式,验证无 5xx、上游命中数正确、
// 且每个响应都是完整合法的 Anthropic 形状(适配器在并发下不串味)。
func TestAdapterTakeoverStress(t *testing.T) {
up := anthropicMockUpstream(t)
dir := t.TempDir()
addr := mustAddr(t)
cfg := writeAnthropicConfig(t, dir, addr, up.URL)
bin := buildBinary(t)
g := startGateway(t, bin, addr, cfg)
// max_concurrent 配成 32,让 96 个请求基本都能真正流过适配器 —— 这个测试
// 要压的是**并发下序列化是否串味**,不是内核的容量拒绝策略。真的撞上上限
// 时收到的 429 也是合法背压(见下方断言),不算失败。
const (
workers = 12
perWorker = 8
)
var (
wg sync.WaitGroup
mu sync.Mutex
ok int
shed int
failed []string
streams int
)
start := make(chan struct{})
for w := 0; w < workers; w++ {
wg.Add(1)
go func(w int) {
defer wg.Done()
<-start
for i := 0; i < perWorker; i++ {
isStream := (w+i)%2 == 0
var payload string
if isStream {
payload = `{"model":"claude-e2e","max_tokens":32,"stream":true,"messages":[{"role":"user","content":"hi"}]}`
} else {
payload = `{"model":"claude-e2e","max_tokens":32,"messages":[{"role":"user","content":"hi"}]}`
}
resp, body := g.do("POST", "/v1/messages", payload, true)
mu.Lock()
switch {
case resp == nil:
failed = append(failed, "nil response")
// 429 = 内核的容量背压(provider busy)。它是**正确**行为,
// 不是缺陷:并发超限时就该如实拒绝,而不是排队或伪造响应。
// 只要求它是结构良好的错误,不要求 200。
case resp.StatusCode == http.StatusTooManyRequests:
if !strings.Contains(body, "busy") && !strings.Contains(body, "error") {
failed = append(failed, "429 不是结构化错误: "+body)
} else {
shed++
}
// 5xx 才是真失败 —— 那是内核或适配器崩了。
case resp.StatusCode >= 500:
failed = append(failed, fmt.Sprintf("5xx status=%d body=%.120s", resp.StatusCode, body))
case resp.StatusCode != 200:
failed = append(failed, fmt.Sprintf("status=%d body=%.120s", resp.StatusCode, body))
case !strings.Contains(body, "pong"):
// 200 但内容不对:说明并发下适配器把响应串味了
failed = append(failed, "missing upstream text: "+body)
case isStream && !strings.Contains(body, "message_stop"):
failed = append(failed, "stream missing message_stop")
case !isStream && strings.Contains(body, `"choices"`):
failed = append(failed, "openai shell leaked: "+body)
default:
ok++
if isStream {
streams++
}
}
mu.Unlock()
}
}(w)
}
close(start)
wg.Wait()
if len(failed) > 0 {
t.Fatalf("%d/%d 请求失败,前 5 例:\n%s", len(failed), workers*perWorker,
strings.Join(failed[:min(5, len(failed))], "\n"))
}
// 至少大部分请求必须真正流过 200 —— 全被背压拒掉说明并发上限配得太低,
// 那这个压力测试就没压到适配器并发。
if ok < workers*perWorker/2 {
t.Fatalf("只有 %d/%d 请求真正成功,压力未触及适配器并发路径",
ok, workers*perWorker)
}
t.Logf("成功 %d(其中流式 %d),背压 429 %d —— 无 5xx、无 OpenAI 形状泄漏",
ok, streams, shed)
}
func min(a, b int) int {
if a < b {
return a
}
return b
}
var _ = time.Second

View File

@ -27,6 +27,8 @@ import (
"fmt"
"sort"
"sync"
golua "github.com/aarzilli/golua/lua"
)
// AdapterOwner 标识一个适配器由谁安装。
@ -185,3 +187,76 @@ func (ps *Plugins) PluginAdapters(plugin string) []string {
func (ps *Plugins) AdapterOwnerOf(name string) string {
return ps.adapterOwner().ownedBy(name)
}
// ---- Lua 绑定(R4:插件在运行时改自己的适配器)----
// adapterBridgeGlobal 暴露给插件 Lua 的适配器管理表。
//
// 用一张表而不是三个顶层全局,是为了少占插件可见的全局命名空间。表用
// buildJSONTable 的同一套写法(NewTable + PushGoClosure + SetField),
// 该写法在 json 表上已被长期验证。
const adapterBridgeGlobal = "__llmsproxy_adapters"
// registerAdapterBridge 把适配器管理表装进插件的 Lua 状态。
//
// name 是**该插件自己的启动名**,随每次调用一起交给 Go 侧桥接 —— 归属校验
// 因此来自启动时的固定身份,而不是从 Lua 状态里现读(那样一个分片或任何持有
// 已编译脚本的调用者就能冒充别的插件)。
func registerAdapterBridge(L *golua.State, ps *Plugins, name string) {
L.NewTable()
install := luaInstallAdapter(L, ps, name)
L.PushGoClosure(install)
L.SetField(-2, "install")
update := luaUpdateAdapter(L, ps, name)
L.PushGoClosure(update)
L.SetField(-2, "update")
remove := luaRemoveAdapter(L, ps, name)
L.PushGoClosure(remove)
L.SetField(-2, "remove")
L.SetGlobal(adapterBridgeGlobal)
}
func luaInstallAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction {
return func(L *golua.State) int {
adapterName := L.ToString(1)
code := L.ToString(2)
effective, err := ps.InstallAdapter(name, adapterName, code)
if err != nil {
L.PushNil()
L.PushString(err.Error())
return 2
}
L.PushString(effective)
L.PushNil()
return 2
}
}
func luaUpdateAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction {
return func(L *golua.State) int {
adapterName := L.ToString(1)
code := L.ToString(2)
if err := ps.UpdateAdapter(name, adapterName, code); err != nil {
L.PushNil()
L.PushString(err.Error())
return 2
}
L.PushBoolean(true)
L.PushNil()
return 2
}
}
func luaRemoveAdapter(L *golua.State, ps *Plugins, name string) golua.LuaGoFunction {
return func(L *golua.State) int {
adapterName := L.ToString(1)
if err := ps.RemoveAdapter(name, adapterName); err != nil {
L.PushNil()
L.PushString(err.Error())
return 2
}
L.PushBoolean(true)
L.PushNil()
return 2
}
}

View File

@ -0,0 +1,234 @@
package lua
import (
"testing"
"time"
)
// 本文件验证 R4 的 Lua 侧绑定:**插件自己的 Lua 代码**能在运行时安装/改写/
// 删除自己的适配器。
//
// 为什么需要单独一层测试:Go 侧的 InstallAdapter/UpdateAdapter 在
// plugin_adapter_bridge_test.go 里已直接验证过,但那证明的是"Go API 能工作"。
// token 轮换的真实形态是插件在定时任务或钩子里**从 Lua 发起**这个调用,所以
// 绑定本身(__llmsproxy_adapters 表)必须被证明可用 —— 而且它曾经在 cgo 层
// SIGSEGV,所以"不崩"本身就是一条断言。
const bridgeProbe = `
local plugin = { name = "authtoken", version = "1.0" }
plugin.exports = { "install_via_lua", "update_via_lua", "remove_via_lua", "install_operator_owned" }
local function src(ver, tok)
return 'return { name = "myauth", version = "' .. ver .. '", endpoint = "/x", token = "' .. tok .. '" }'
end
function plugin.install_via_lua(args)
local name, err = __llmsproxy_adapters.install("myauth", src("1.0", args.tok))
return { name = name, err = err }
end
function plugin.update_via_lua(args)
local ok, err = __llmsproxy_adapters.update("myauth", src("2.0", args.tok))
return { ok = ok, err = err }
end
function plugin.remove_via_lua()
local ok, err = __llmsproxy_adapters.remove("myauth")
return { ok = ok, err = err }
end
-- 试图替换一个非自己拥有的适配器:必须被拒(不能影子内置)
function plugin.install_operator_owned()
local name, err = __llmsproxy_adapters.install("openai", src("9.9", "evil"))
return { name = name, err = err }
end
return plugin
`
// TestPluginInstallsAdapterFromLua 是 R4 绑定的主断言。
func TestPluginInstallsAdapterFromLua(t *testing.T) {
_, ps, _ := runtimePlugins(t)
if err := ps.LoadSource("authtoken", bridgeProbe); err != nil {
t.Fatal(err)
}
out, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "T1"})
if err != nil {
t.Fatalf("Lua 侧 install 调用失败: %v", err)
}
m, ok := out.(map[string]interface{})
if !ok {
t.Fatalf("返回值不是 table: %#v", out)
}
if m["name"] != "myauth" {
t.Fatalf("install 应返回生效名 myauth, 实得 %#v (err=%#v)", m["name"], m["err"])
}
// 归属确实被记录(这条规则是"插件只能改自己装的"的基础)
if owner := ps.AdapterOwnerOf("myauth"); owner != "authtoken" {
t.Errorf("归属应为 authtoken, 实得 %q", owner)
}
}
// TestPluginRotatesTokenViaLua 是 token 轮换的完整形态:
// 同一个插件先装、再改,改后新版本生效。
func TestPluginRotatesTokenViaLua(t *testing.T) {
_, ps, _ := runtimePlugins(t)
if err := ps.LoadSource("authtoken", bridgeProbe); err != nil {
t.Fatal(err)
}
if _, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "OLD"}); err != nil {
t.Fatalf("install: %v", err)
}
// 轮换
out, err := ps.Call("authtoken", "update_via_lua", map[string]interface{}{"tok": "NEW"})
if err != nil {
t.Fatalf("update: %v", err)
}
m := out.(map[string]interface{})
if m["ok"] != true {
t.Fatalf("Lua 侧 update 未成功: %#v", m)
}
// 新代码必须真的生效 —— 这是"轮换有意义"的定义
if rev := ps.vm.AdapterRevision("myauth"); rev < 2 {
t.Errorf("适配器修订号应 >=2(已被替换一次),实得 %d", rev)
}
}
// TestPluginRemoveAdapterFromLua:插件能删掉自己装的。
func TestPluginRemoveAdapterFromLua(t *testing.T) {
_, ps, _ := runtimePlugins(t)
if err := ps.LoadSource("authtoken", bridgeProbe); err != nil {
t.Fatal(err)
}
if _, err := ps.Call("authtoken", "install_via_lua", map[string]interface{}{"tok": "T"}); err != nil {
t.Fatal(err)
}
out, err := ps.Call("authtoken", "remove_via_lua", map[string]interface{}{})
if err != nil {
t.Fatalf("remove: %v", err)
}
if out.(map[string]interface{})["ok"] != true {
t.Fatalf("Lua 侧 remove 未成功: %#v", out)
}
if owner := ps.AdapterOwnerOf("myauth"); owner != "" {
t.Errorf("删除后归属应清空, 实得 %q", owner)
}
}
// TestLuaBridgeCannotShadowBuiltin:从 Lua 侧同样不能影子内置适配器 ——
// 否则绑定就成了绕过 Go 侧校验的后门。
func TestLuaBridgeCannotShadowBuiltin(t *testing.T) {
_, ps, _ := runtimePlugins(t)
if err := ps.LoadSource("authtoken", bridgeProbe); err != nil {
t.Fatal(err)
}
out, err := ps.Call("authtoken", "install_operator_owned", map[string]interface{}{})
if err != nil {
t.Fatalf("调用本身不应报错(错误应作为返回值): %v", err)
}
m := out.(map[string]interface{})
if m["name"] != nil {
t.Fatalf("影子内置 openai 必须被拒,却返回了 name=%#v", m["name"])
}
if m["err"] == nil || m["err"] == "" {
t.Fatal("影子内置应带回错误信息")
}
t.Logf("拒绝信息: %v", m["err"])
}
// TestScheduledTaskRotatesTokenEndToEnd 是 R4 的闭环验收。
//
// 它把三个特性串成一条链,任何一个环节坏掉都会让它失败:
// 1. 内核 ticker 驱动定时任务(无外部触发)
// 2. 任务与钩子跑在同一个 VM(所以任务的 install 与 state 写入对钩子可见)
// 3. Lua 侧适配器桥接可用(__llmsproxy_adapters.install)
//
// 这正是用户描述的 token 轮换形态:插件自己按时间拿到新 token,重写适配器,
// 后续请求自然使用新逻辑 —— 全程没有任何浏览器或人工操作。
func TestScheduledTaskRotatesTokenEndToEnd(t *testing.T) {
shortTick(t)
_, ps, _ := runtimePlugins(t)
code := `
local plugin = { name = "sched_rot", version = "1.0" }
plugin.state = { token = "OLD", rotates = 0 }
plugin.schedule = { { fn = "rotate", every = 1 } }
function plugin.rotate()
plugin.state.rotates = plugin.state.rotates + 1
local tok = "T" .. tostring(plugin.state.rotates)
local src = 'return { name = "rotauth", version = "1.0", endpoint = "/x", token = "' .. tok .. '" }'
plugin.state.last_adapter, plugin.state.rotate_err = __llmsproxy_adapters.install("rotauth", src)
plugin.state.token = tok
end
plugin.hooks = { routed = "observe" }
function plugin.observe(payload)
payload.token = plugin.state.token
payload.rotates = plugin.state.rotates
payload.adapter = plugin.state.last_adapter
return payload
end
return plugin
`
if err := ps.LoadSource("sched_rot", code); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(4 * time.Second)
for time.Now().Before(deadline) {
out := ps.Fire(StageRouted, map[string]interface{}{})
if n, ok := out["rotates"].(float64); ok && n >= 1 {
t.Logf("任务轮换了 %v 次; 钩子看到 token=%v adapter=%v", n, out["token"], out["adapter"])
if out["adapter"] != "rotauth" {
t.Fatalf("任务里 install 未生效: adapter=%v", out["adapter"])
}
return
}
time.Sleep(30 * time.Millisecond)
}
t.Fatal("定时任务未完成 token 轮换")
}
// TestShardVMHasNoAdapterBridge 锁定一条边界:分片 VM(ShardCall 用的私有 VM)
// **不能**拿到适配器管理桥接。
//
// 理由:分片是文档化的「只读、可丢弃」工作单元,而 install/update/remove 是
// **全局**副作用(改的是共享的适配器注册表,不是本 VM 的状态)。若把桥接给
// 分片,一个被声明为只读的按需调用就能悄悄改写所有请求共用的适配器 ——
// 正是分片契约要防的那类不可见变更。
//
// 断言方式是:分片里的插件代码检查 __llmsproxy_adapters 是否为 nil,返回结果。
func TestShardVMHasNoAdapterBridge(t *testing.T) {
_, ps, _ := runtimePlugins(t)
code := `
local plugin = { name = "sharded", version = "1.0" }
plugin.exports = { "probe" }
-- 只读探测:分片里这个全局必须是 nil
function plugin.probe(args)
return { has_bridge = __llmsproxy_adapters ~= nil }
end
return plugin
`
if err := ps.LoadSource("sharded", code); err != nil {
t.Fatal(err)
}
// 主 VM 上应当有桥接(对照)
mainOut, err := ps.Call("sharded", "probe", map[string]interface{}{})
if err != nil {
t.Fatalf("主 VM 调用失败: %v", err)
}
if m, ok := mainOut.(map[string]interface{}); !ok || m["has_bridge"] != true {
t.Errorf("主 VM 上应有 __llmsproxy_adapters, 实得 %#v", mainOut)
}
// 分片 VM 上不应有
outs, err := ps.ShardCall("sharded", "probe", []map[string]interface{}{{}, {}})
if err != nil {
t.Fatalf("ShardCall: %v", err)
}
for i, o := range outs {
m, ok := o.(map[string]interface{})
if !ok {
t.Fatalf("分片 %d 返回非 table: %#v", i, o)
}
if m["has_bridge"] == true {
t.Errorf("分片 %d 不应拿到适配器桥接(分片是只读可丢弃的)", i)
}
}
}

View File

@ -164,7 +164,14 @@ func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
sp.live++
sp.mu.Unlock()
w, err := bootPluginState(code, name, sp.ps.outboundAllowlist())
// ps is deliberately nil: a shard VM must NOT get the adapter-management
// bridge. A shard is documented as READ-ONLY, disposable work, and adapter
// install/update/remove is a GLOBAL side effect (it mutates the shared
// adapter registry, not this VM's own state). Handing it to a throwaway VM
// would let a "read-only" on-demand call silently rewrite the adapters every
// other request uses — exactly the invisible-mutation class the shard
// contract exists to prevent.
w, err := bootPluginState(code, name, sp.ps.outboundAllowlist(), nil)
if err != nil {
sp.mu.Lock()
sp.live--

View File

@ -593,7 +593,7 @@ func (ps *Plugins) SeedBundled() error {
// bootPluginState compiles one plugin into a fresh Lua state and stores its
// returned table under pluginGlobal. It is the plugin counterpart of
// adapterPool.boot, minus the pooling: a plugin keeps exactly one state.
func bootPluginState(code, name string, allow *httpAllowlist) (*worker, error) {
func bootPluginState(code, name string, allow *httpAllowlist, ps *Plugins) (*worker, error) {
L := golua.NewState()
L.OpenLibs()
setupGlobals(L)
@ -604,6 +604,12 @@ func bootPluginState(code, name string, allow *httpAllowlist) (*worker, error) {
if allow != nil {
registerFn(L, httpGlobal, luaHTTPFn(L, allow))
}
// R4: plugins — and only plugins — can install/update/remove their own
// adapters at runtime. Scoped here (not in setupGlobals) for the same reason
// as __llmsproxy_http: adapters share setupGlobals and must not get it.
if ps != nil {
registerAdapterBridge(L, ps, name)
}
if err := L.DoString(code); err != nil {
L.Close()
return nil, fmt.Errorf("compile plugin %s: %w", name, err)
@ -658,7 +664,7 @@ func (ps *Plugins) LoadSource(name, code string) error {
// One state, held for the plugin's lifetime (see Plugin.state): a hook that
// accumulates into plugin.state would otherwise split its totals across
// whichever worker it happened to run on.
w, err := bootPluginState(code, name, ps.outboundAllowlist())
w, err := bootPluginState(code, name, ps.outboundAllowlist(), ps)
if err != nil {
p.LoadError = err.Error()
ps.append(p)