Sakurasan
979973ddde
流式请求在首个内容事件到达前由网关缓冲（借鉴 magpie 的 holdWriter）：上游的错误 事件、空流、断连、协议形态不符（要流给了整包）都在放行前换渠道消化，客户端无感知。 扣住期间照常发 SSE 注释心跳；缓冲超过 1MB 直接放行退化为普通转发，判定失效不吞正文。 - 新增 proxy/hold.go：streamHold（缓冲/放行/心跳/首 token 计时与放行互斥）、 sseEventHasContent（chat/messages/responses 三种协议的内容判定）、 sseUpstreamError（200+流式形式的错误事件识别）、classifyUpstreamError + quotaSignals（余额/配额特征识别） - 非流式改为先读完整响应再写头：读失败（网络错误/超时）发生在写出之前，可换渠道 重试；此前先写 200 头再读，超时只能给客户端留一个 200 空响应 - 协议形态校验：要流却给了整包（didNotStream）按失败换渠道，反复如此的渠道会被 健康计数剔除 - 记账：每次尝试独立 trySink，只有成功走到 EOF 才并入请求级 sink——失败渠道的 message_start 用量不计入账单 - 冷却按原因分类（channel.FailKind）：余额/配额类第一次失败即冷却 30 分钟（429 不 一定是限流，智谱 429「余额不足」也在此列）；其余按连续进冷却次数指数退避 30s→60s→120s，封顶 10 分钟；短于退避值的 Retry-After 不再缩短冷却 - 新增 OT_PROXY_FIRST_TOKEN_WAIT（默认 0=不限）：放行前等内容的最长时间，超时即 取消本次尝试换下一个渠道 - 渠道列表接口返回 cooldown_reason，后台可区分「限流」与「余额耗尽」 - 测试：hold 缓冲/放行/心跳、内容与错误事件判定、失败分类与指数退避、容量热生效 - docs/magpie-plugin-mechanism.md：参考实现的插件机制调研
274 lines
19 KiB
Markdown
274 lines
19 KiB
Markdown
# magpie 插件机制深度调研
|
||
|
||
> 调研对象：[yetone/magpie](https://github.com/yetone/magpie)（本地 LLM 网关，Go，~855K 行）。
|
||
> 本文聚焦其插件机制本身，目标是把可移植的设计提炼出来供其他项目参考。所有行号基于 2026-10-30 的 main 分支。
|
||
|
||
## 0. 总览：三层并存的「插件」
|
||
|
||
magpie 里「插件」一词覆盖三个完全不同的机制，看懂分层是看懂代码的前提：
|
||
|
||
| 层 | 位置 | 形态 | 运行环境 | 典型用途 |
|
||
| --- | --- | --- | --- | --- |
|
||
| 内置插件（built-in） | `plugins/<id>` 包 + `internal/pluginrt` | 编译进二进制的 Go 包，用户可开关 | 网关主进程 | routing、usage、redact、sessions 等功能开关 |
|
||
| JS 网关中间件 | `internal/middleware` | 用户安装的 JS 模块 | 进程内 moejs 引擎（Go 写的 JS 引擎） | 改写请求/响应、路由策略、用量观察 |
|
||
| Provider 插件 | `internal/plugin`（`host.js`） | OpenCode v1 兼容的 npm/git/本地包 | Bun 子进程，stdin/stdout JSON-RPC | 供应商接入：登录、模型列表、代发请求 |
|
||
|
||
三层的调用频率完全不同，这决定了它们的技术选型（见 §4.1）。
|
||
|
||
---
|
||
|
||
## 1. 内置插件系统（pluginrt）
|
||
|
||
### 1.1 设计动机
|
||
|
||
包注释（`internal/pluginrt/pluginrt.go:1-10`）说得直白：magpie 的功能「不是所有人都需要」（Lxxyx on X: "功能太多了"），所以把非核心功能拆成可开关的包。关键约束写在注释里并**由测试强制**：
|
||
|
||
- 运行时是 core，**永不 import 任何插件**；
|
||
- 插件之间**互相不 import**，只能通过彼此 Provide 的服务通信。
|
||
|
||
这两条不是 review 纪律，而是 `TestPluginsImportOnlyCore`（`internal/pluginrt/pluginrt_test.go:94`）：用 AST 扫描全部 `.go` 文件的 import，违反即编译期红。架构边界测试化，比文档和口头约定可靠得多。
|
||
|
||
### 1.2 Manifest：插件的全部声明
|
||
|
||
`plugins/<id>` 包在 `init()` 里调 `pluginrt.Register(Manifest)`（`pluginrt.go:128`，重复 ID 直接 panic）。Manifest 字段（`pluginrt.go:35-72`）：
|
||
|
||
```go
|
||
type Manifest struct {
|
||
ID string
|
||
API string // 宿主 API 版本区间，caret 语义 "^1"
|
||
Requires, Provides []string // 服务名：插件间唯一的依赖/通信表达
|
||
Essential bool // essentials 预设是否保留
|
||
Name, Sub string // 展示名（窗口负责翻译）
|
||
One, Many string // "1 个分组在用" / "{n} 个分组在用"
|
||
InUse func() (int, error) // 运行时清点"有多少东西在用它"
|
||
Apply func(*Ctx) error // 注册钩子；nil 表示纯 UI 插件
|
||
Parts []Part // 前端 UI 插槽：nav.page / settings.section / part
|
||
Defaults map[string]any
|
||
Settings any // 设置结构体 → 反射生成 JSON Schema → 合并进 Defaults
|
||
Client []string // 窗口侧资源（脚本/i18n/css）
|
||
}
|
||
```
|
||
|
||
三个值得单独说的设计：
|
||
|
||
**InUse 运行时清点**（`pluginrt.go:48-51`）。插件能否被关掉不靠用户记忆，而是运行时数「还有多少用户创建的东西依赖它」（分组、预热任务……）。InUse > 0 时无视 plugins.yml 强制开启；**读取出错也算在用**——"unknown is never unused"，宁可多开不少开，避免用户的东西失去宿主却还在运行。
|
||
|
||
**Settings → JSON Schema**（`pluginrt.go:137-150` + `schema.go`）。Register 时反射设置结构体生成 schema 并把零值合并进 Defaults；窗口用 schema 画表单，`SetConfig` 只接受 schema 允许的字段——用户手改的非法值被拒收但保留在文件里，不会静默丢失。
|
||
|
||
**API 版本 caret 区间**（`pluginrt.go:391-402`）。`Compatible("^1.2")` = 主版本相同且 minor ≤ 宿主。不匹配的插件不 apply，状态记为 Failed 并说明原因。
|
||
|
||
### 1.3 生命周期：Reconcile 不动点状态机
|
||
|
||
唯一的入口是 `Runtime.Reconcile(on, config)`（`pluginrt.go:211-278`），每次配置变化都整体收敛一遍：
|
||
|
||
1. **先关**：迭代求出「不再 wanted」∪「Requires 的服务被某个要关插件 Provides」的闭包（`needsGoing`，:281-292），然后**按 apply 的逆序** dispose——保证服务在它的使用者之后才被撤回。每个 dispose 闭包单独 recover（`undo`，:342-354），一个插件清理时 panic 不影响其他插件。
|
||
2. **后开**：循环「能应用就应用」直到不动点——Requires 全部满足才 apply（core 提供 `settings/providers/gateway/window` 四个基础服务，:31），不满足则挂 Pending 并记录在等谁。这个不动点循环天然实现了**按服务依赖图的拓扑序 apply、逆拓扑序 dispose**，不需要声明式 priority。
|
||
3. 状态集：`off / pending / active / failed / disposed`（:83-89），GUI 可见每个插件为什么没起来（`Why` 字段）。
|
||
|
||
Apply 的 panic 被 recover 成该插件 FAILED（:308-330），**它注册了一半的东西也全部 undo**——不存在「半应用」状态。
|
||
|
||
### 1.4 Ctx：注册即登记撤销
|
||
|
||
`Apply` 拿到的 `Ctx`（`pluginrt.go:93-119`）只有四个能力，全部自带撤销：
|
||
|
||
- `OnDispose(f)`：登记清理函数，关插件时逆序执行；
|
||
- `Contribute(slot, key)`：往窗口放一块 UI；
|
||
- `Provide(name, v)`：提供 named service，撤销时从 services map 删除；
|
||
- `Service(name)`：取别的插件 Provide 的服务。
|
||
|
||
配套的核心模式是「**注册函数返回 remove 闭包**」：每个扩展点的 `Use/Serve/Register` 都返回 `remove func()`，插件把它交给 `Ctx.OnDispose`，一行完成即插即拔。remove 闭包统一用 `sync.Once` 包裹防重复（`gwmw.go:141-154` 的泛型 `add[T]` 是模板实现）。
|
||
|
||
### 1.5 配置系统
|
||
|
||
**plugins.yml 三层合并**（`pluginrt/config.go:126-175` `Compose`）：内置 Defaults → preset（essentials/all）→ 用户行。每字段记录来源 `Source`（default/preset/user/in use），`magpie plugin dump` 能解释「这个值从哪来」——排查配置不生效的利器。
|
||
|
||
**热更新三件套**（`config.go:363-395` + `gateway.go:649-650`）：
|
||
|
||
1. `filememo`：按 (mtime, size, 原始字节) 记忆解析结果，本进程写文件后经 channel 立即失效；
|
||
2. fsnotify 看其他进程的写入；
|
||
3. 轮询兜底；
|
||
4. 再加一条「**请求路径按需拉起**」：请求到达时发现「需要 masker/limiter/中间件但没有」，节流（≤1 次/秒）触发 `pluginrt.Refresh` 立刻重新收敛（`gateway.go:631-644`）。效果是配置改了不用重启，稳态开销近乎为零。
|
||
|
||
**settings 读取的健壮性**（`internal/settings/settings.go` + `internal/filememo`）：解析结果按原始字节缓存 + `Shared()` 免拷贝只读实例；损坏文件回退 lastgood（最后一次完好的代）；「读不了」与「没有文件」语义分开（`LoadChecked`）；写入走全量校验 + temp+rename 原子替换 + `.bak`。
|
||
|
||
---
|
||
|
||
## 2. 网关扩展点（core 侧）
|
||
|
||
core 为插件预留了五个「点」，每个是一个独立小包，统一模式：**接口 + 注册返回 remove 闭包 + 堆叠时后注册者生效 + 显式的失败策略**。
|
||
|
||
### 2.1 中间件点（gwmw）——同一包内两种失败策略
|
||
|
||
`internal/gwmw/gwmw.go` 包注释（1-16 行）开门见山：这个包有两个位置，**各有各的失败策略**：
|
||
|
||
- **Agent 中间件**（fail-open）：挂在客户端请求进网关路由之前。接口（:45-59）：
|
||
|
||
```go
|
||
type Agent interface { Begin(Info) Run }
|
||
type Run interface {
|
||
Request(body []byte) ([]byte, *Reject) // 改写或拒绝
|
||
Wrap(w http.ResponseWriter) (http.ResponseWriter, func()) // 包响应写手
|
||
End()
|
||
}
|
||
```
|
||
|
||
中间件挂了/超时了 → 请求原样放行。可选接口 `Tried(body, To) []byte`（:72-74）让网关**每次换上游重试前重新变形 body**，每次 try 从 onRequest 后的 body 出发、互不污染。
|
||
- **Masker**（fail-closed）：挂在发往供应商之前。`Masking(wanted)` 在「设置要求遮罩但没有 masker 在服务」时返回 `ErrUnmasked` 拒发（:269-277），错误信息直接告诉用户去哪里打开插件——**错误写到用户能行动的层面**。`Safely`（:281-288）把 masker 的 panic 也转成错误，绝不把未遮罩的请求发出去。
|
||
|
||
另有一个精巧的「拉起」机制：`Served()` 发现没有 masker 时节流地回调 refresh（最多 1 秒一次、不重入，`throttle` :194-212），于是「设置打开了遮罩」会在下一个请求时自动把 redact 插件拉起来。
|
||
|
||
### 2.2 路由策略点（routes）——只排序，不决策
|
||
|
||
`internal/routes/routes.go`：`Strategy` 只做一件事——给出候选池的**排列**（`Order func(pool string, seats []Seat) []int`）。resting、亲和、故障转移全部留在网关核心，策略碰不到。`Apply`（routes.go 尾部）对策略输出做校验：panic、超时、或者不是合法排列，就 fail-open 回退到列表顺序并打日志。`Seat`（:29-72）携带策略可能需要的全部事实：使用率、各窗口重置时间、衰减 token 数、权重、是否在冷却、实测 TTFT/TPS、钱包余额……**策略是无状态函数，事实由核心喂给它**。
|
||
|
||
### 2.3 预算点（gwbudget）——fail-closed 的限流
|
||
|
||
`Limiter.Reserve(who, body, model, now) (release, Refusal)`（`gwbudget.go:33-35`）。key 配了 limit 但没有 limiter 在服务时返回 `ErrUnlimited` 拒绝（:106-124）——计费限额和遮罩一样 fail-closed：**该拦的没拦住，比误拦严重**。
|
||
|
||
### 2.4 观察点（usage.Observer）——按需声明开销
|
||
|
||
`internal/usage/observe.go:19-31`：
|
||
|
||
```go
|
||
type Observer interface {
|
||
Serve() (stop func()) // 常驻 goroutine 的生命周期
|
||
Bodies() Bodies // None / 256KiB / Whole
|
||
Begin(Request) Watch // Attempt(每此尝试) + End(整请求)
|
||
Ledger(Record) // 账本每一行
|
||
}
|
||
```
|
||
|
||
精髓在 `Bodies()`：**观察者自己声明要多少数据**，网关据此决定要不要为它缓冲完整流（`ObservedBodies` :146-152 取最大值）；没有观察者注册时，网关连整包缓冲都不做，「不做任何观察工作，连回复的 body 都不留」——零观察者零开销。
|
||
|
||
### 2.5 工具点（gwtool）
|
||
|
||
极简的按名开关：`Gate()` 网关启动时全局关闸、`Serve(name)`/`Served(name)` 按名放行（`gwtool.go:34-72`）。search 插件供 `web_search`、tools 插件供 `plugin_tools`，就这么点东西——**能用一个 gate 表达的就不用接口**。
|
||
|
||
---
|
||
|
||
## 3. JS 中间件引擎（moejs，进程内）
|
||
|
||
### 3.1 为什么不用 Bun：按调用频率选引擎
|
||
|
||
包注释（`internal/middleware/middleware.go:1-12`）给了决策依据：一个流式回复有数千个事件，**进程内 moejs 跑一个 hook 约 1µs**，Bun 子进程一个来回要数十 µs——每事件都走子进程是不可接受的。于是：
|
||
|
||
- **高频钩子**（每请求、每 SSE 事件）→ 进程内 moejs；
|
||
- **低频异步钩子**（可 await I/O、观察完整往返）→ Bun 子进程（见 §4）。
|
||
|
||
这是「插件热路径留在进程内、重 I/O 放子进程」的教科书分层。
|
||
|
||
### 3.2 运行时管理
|
||
|
||
- 每个中间件**编译一次**成模块；每个请求从 `sync.Pool` 取独立 runtime（`middleware.go:44-71`），请求的多个 hook 共享 `ctx.state`；
|
||
- 超时硬限：请求级 hook 250ms、事件级 50ms、加载 1s（:35-39），超时用 `runtime.Interrupt` 打断，输入保持原样；
|
||
- 热加载用 stamp（plugins.json mtime + 各中间件文件 stat），**每请求最多重查一次/秒**（`current()` :95-116），热路径几乎零开销。
|
||
|
||
### 3.3 四个 hook 与两个扩展导出
|
||
|
||
| Hook | 签名 | 语义 |
|
||
| --- | --- | --- |
|
||
| `onRequest` | `(body, ctx) → body \| undefined` | 改写入站请求；`ctx.reject()` 可拒；undefined 原样放行 |
|
||
| `onProvider` | `(body, ctx) → body` | 每次换上游重试前调用，`ctx.provider` 告诉你是谁；返回 null 报错（已路由不能丢） |
|
||
| `onEvent` | `(data, ctx)` | SSE 逐事件；`export const events` 可白名单事件名 |
|
||
| `onResponse` | `(body, ctx)` | 整段 JSON 响应（缓冲后调用） |
|
||
|
||
两个额外导出（`points.go:1-10`，都跑在模块自己的单线程 runtime 里，互相可见）：
|
||
|
||
- **`strategies`**：导出路由策略函数 `(pool, seats, options) → 顺序`，注册成 routes.Strategy。在请求路径上跑，20ms 超时、必须返回座位下标的排列，否则 fail-open 回退原顺序；
|
||
- **`onUsage`**：账本行观察者，**跑在离线队列**——容量 1024，满了丢弃并计数，绝不阻塞请求路径（`Ledger` 的 select/default，points.go:376-398）；row 是白名单化的 `usageRow`，**没有 body、prompt、错误文本、会话 id**——第三方代码拿不到敏感数据。
|
||
|
||
防第三方代码拖垮网关的完整清单：超时 Interrupt、日志限速 60 行/分钟（`middleware.go:383-404`）、onUsage 满队丢弃并计数、钩子 context 白名单化（无鉴权头）、策略必须返回合法排列。
|
||
|
||
---
|
||
|
||
## 4. Provider 插件宿主（Bun 子进程）
|
||
|
||
### 4.1 线协议（`internal/plugin/host.js:1-20`）
|
||
|
||
stdin/stdout，一行一个 JSON：
|
||
|
||
```
|
||
magpie → host: {id, method, params} host 应答 {id, result} 或 {id, error}
|
||
```
|
||
|
||
fetch 类请求的流式应答分三段：`{id, event:"head", status, headers}` → 多个 `{id, event:"chunk", data(base64)}` → `{id, result:null}`。`{method:"abort"}` 取消单个请求。host 主动上报：`log`/`toast`（插件想展示的）、`auth`（插件保存或刷新了登录态）。
|
||
|
||
两个值得抄的细节：
|
||
|
||
- **credit 窗口背压**（`host.js:22-38` 一带）：每个 fetch 有字节窗口，读端消费后用 `{method:"credit", id, n}` 归还额度——**慢读端会暂停上游，而不是让宿主无界缓冲**。子进程 IPC 背压的标准解法。
|
||
- **多账号槽位**：第一个账号存 provider id 下，其余存 `id#<slot>`；请求内的一切（getAuth、auth.set、loader）通过 `AsyncLocalStorage` 只见自己的账号作用域。
|
||
|
||
### 4.2 入口声明与异步请求钩子
|
||
|
||
Provider 插件走 OpenCode v1 约定；网关异步钩子另有两个入口（`internal/plugin/gateway.go:15-45`）：package.json 的 `magpie.hooks` 字段，或 `*.hooks.{js,mjs,ts,mts}` 文件（Bun 直接跑 TS）。与中间件的 `*.gateway.js` **入口名刻意不重叠**，一个文件永远不会被当两种东西加载。
|
||
|
||
钩子协议是 begin/finish/close 三段（`gateway.go:174-260`）：Begin 返回改写后的 body 和 Run 句柄，Finish 异步通知回复，Close 释放；总预算 11 秒；**传给插件的 context 刻意剥离鉴权头**（:82-92）。
|
||
|
||
### 4.3 供应链
|
||
|
||
Bun 本体按平台下载，SHA-256 钉死在代码里（对照 Bun 官方 SHASUMS256.txt 校验）；npm 没有的包从社区仓库 `dist` 分支拉，sha512 与代码同源——信任模型是「谁有权限推 dist 分支」而非哈希本身，这个判断写在 LESSONS.md 里。插件更新有 `min` 版本机制，内置代码里记录着「要求插件 ≥ 某版」。
|
||
|
||
---
|
||
|
||
## 5. 设计原则提炼（可移植清单）
|
||
|
||
1. **架构边界测试化**。core 不依赖插件、插件互不依赖，写成 AST 扫描测试。一条测试顶一百条 review 约定。
|
||
2. **注册表 + 服务依赖 + 逆序 dispose**。依赖用服务名表达（`Requires/Provides`），Reconcile 不动点自动拓扑排序，不需要 priority 字段；关停永远逆序，清理函数各自 recover。
|
||
3. **remove 闭包 + sync.Once** 是唯一的注册/注销协议。所有扩展点长一个样，学一次就会用。
|
||
4. **每个扩展点显式声明 fail-open 还是 fail-closed**，并把错误信息写到用户能行动的层面。判断标准：功能性增强（改写、路由、观察）fail-open；安全/计费/隐私（遮罩、限额）fail-closed。
|
||
5. **按调用频率选执行引擎**。热路径（每事件 µs 级）留进程内，重 I/O 放子进程，两者入口命名隔离。
|
||
6. **给第三方代码上全套缰绳**：超时 Interrupt、日志限速、队列满丢弃计数、context 白名单、输出结构校验。宁可功能降级，不让插件拖垮宿主。
|
||
7. **观察者按需声明数据需求**（Bodies 分级），零注册零开销。
|
||
8. **InUse 运行时清点**决定插件可否关闭，错误也算在用。
|
||
9. **配置四层合并 + 来源追踪 + lastgood 兜底 + temp+rename 原子写**；热更新 = 文件记忆缓存 + fsnotify + 轮询兜底 + 请求路径按需拉起。
|
||
10. **策略无状态化**：策略只给排列，事实（冷却、用量、延迟）由核心以 `Seat` 喂入，输出非法就 fail-open。
|
||
|
||
## 6. 最小移植骨架
|
||
|
||
如果要在别的 Go 项目里复刻这套机制，最小核心约 600 行：
|
||
|
||
```go
|
||
// 1. 运行时（~200 行）
|
||
type Manifest struct {
|
||
ID string
|
||
Requires, Provides []string
|
||
Apply func(*Ctx) error
|
||
}
|
||
type Ctx struct { /* id + disposes + services */ }
|
||
func (c *Ctx) OnDispose(f func()) { c.disposes = append(c.disposes, f) }
|
||
func (rt *Runtime) Reconcile(on map[string]bool) {
|
||
// 先关：wanted=false ∪ Requires 命中正在关闭者 Provides 的闭包，逆序 dispose
|
||
// 后开：不动点循环，Requires 满足才 apply，apply panic → undo + Failed
|
||
}
|
||
|
||
// 2. 扩展点（每个 ~30 行）
|
||
func Use(h Hook) (remove func()) { // 泛型 add[T]
|
||
// append 到注册表；返回 sync.Once 包裹的 remove
|
||
}
|
||
|
||
// 3. 边界测试（~50 行）
|
||
// AST 扫描：core 包不得 import plugins/*；plugins/* 互不 import
|
||
```
|
||
|
||
再按需加：配置三层合并（yaml + Source 追踪）、一个 fail-open 的请求中间件点、一个 fail-closed 的计费点。Bun/JS 宿主只在确实需要第三方生态时再引入。
|
||
|
||
## 7. 关键文件索引
|
||
|
||
| 文件 | 内容 |
|
||
| --- | --- |
|
||
| `internal/pluginrt/pluginrt.go` | 注册表、Manifest、Reconcile 状态机、API 兼容 |
|
||
| `internal/pluginrt/config.go` | plugins.yml 三层合并、热更新、原子写 |
|
||
| `internal/pluginrt/schema.go` | Settings → JSON Schema |
|
||
| `internal/pluginrt/pluginrt_test.go` | `TestPluginsImportOnlyCore`（AST 边界测试） |
|
||
| `internal/gwmw/gwmw.go` | 中间件点：Agent（fail-open）/ Masker（fail-closed） |
|
||
| `internal/routes/routes.go` | 路由策略点：Seat 与 Strategy |
|
||
| `internal/gwbudget/gwbudget.go` | 预算点（fail-closed） |
|
||
| `internal/usage/observe.go` | 观察点：Bodies 分级、Observer 生命周期 |
|
||
| `internal/middleware/middleware.go` | moejs 引擎、编译缓存、热加载 stamp |
|
||
| `internal/middleware/run.go` | 每请求 runtime、hook 派发 |
|
||
| `internal/middleware/points.go` | strategies / onUsage 扩展导出、离线队列 |
|
||
| `internal/plugin/host.js` | Bun 宿主：JSON 行协议、credit 背压、账号作用域 |
|
||
| `internal/plugin/gateway.go` | 异步请求钩子入口与三段协议 |
|
||
| `docs/subsystems/plugin-runtime.md` | 官方设计文档（P3/P4 扩展点清单） |
|
||
| `docs/subsystems/gateway-middleware.md` | 中间件点官方文档 |
|