diff --git a/README.md b/README.md index 78a0407..a5f5d05 100644 --- a/README.md +++ b/README.md @@ -145,6 +145,7 @@ agentmail/ │ └── dsh-mail-bridge/ # DeepSeek Harness（Cordis） ├── web/ # 前端（React + Vite + Tailwind） └── deploy/ # systemd 单元 + 安装脚本 + └── remote-agent-demo.py # 最小跟主机 Agent（纯标准库，验证协议层能力） ``` ## 技术栈 diff --git a/deploy/check-shared-libs.sh b/deploy/check-shared-libs.sh index 7c1020a..0284d9e 100755 --- a/deploy/check-shared-libs.sh +++ b/deploy/check-shared-libs.sh @@ -7,14 +7,14 @@ set -euo pipefail A=plugins/opencode-mail-bridge B=plugins/dsh-mail-bridge fail=0 -for f in relay-dedup inbox-format session-snapshot workspace model-scope; do +for f in relay-dedup inbox-format session-snapshot workspace model-scope catchup; do if ! diff -q "$A/lib/$f.js" "$B/lib/$f.js" >/dev/null 2>&1; then echo "共用模块已分叉：lib/$f.js" >&2 diff "$A/lib/$f.js" "$B/lib/$f.js" | head -20 >&2 fail=1 fi done -for f in inbox-format session-snapshot workspace model-scope; do +for f in inbox-format session-snapshot workspace model-scope catchup; do if ! diff -q "$A/test/$f.test.mjs" "$B/test/$f.test.mjs" >/dev/null 2>&1; then echo "共用测试已分叉：test/$f.test.mjs" >&2 fail=1 diff --git a/deploy/remote-agent-demo.py b/deploy/remote-agent-demo.py new file mode 100755 index 0000000..de3880f --- /dev/null +++ b/deploy/remote-agent-demo.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +"""最小跨主机 Agent —— 只用 python 标准库，证明跨主机不需要新组件。 + +用途：在一台**没有装 AgentMail 任何代码**的机器上收发邮件。 +如果这个脚本能跑通，「跨主机 Agent」在协议层面就已经成立 —— +不需要注册中心，因为连接方向是单向的：Agent 主动连 Gateway，Gateway 从不外呼。 + +用法： + + export AGENTMAIL_GATEWAY_URL=https://mail.example.com/api/v1 + export AGENTMAIL_AGENT_KEY=<在 Gateway 上建的 Agent 密钥> + export AGENTMAIL_AGENT_NAME=remotebot + export AGENTMAIL_WORKSPACE=/tmp/remotebot-ws # 可选 + python3 remote-agent-demo.py [运行秒数] + +它做四件事：注册 → 心跳（带模型目录）→ SSE 长连 → 收到邮件就回一封。 +真正的插件还要做权限转发、会话命名回写、附件等，见 docs/PLUGIN-GUIDE.md。 +""" +import json +import os +import sys +import threading +import time +import urllib.error +import urllib.request + +GW = os.environ.get("AGENTMAIL_GATEWAY_URL", "http://127.0.0.1:8180/api/v1").rstrip("/") +KEY = os.environ.get("AGENTMAIL_AGENT_KEY", "") +NAME = os.environ.get("AGENTMAIL_AGENT_NAME", "remotebot") +WORKSPACE = os.environ.get("AGENTMAIL_WORKSPACE", "/tmp/%s-ws" % NAME) + +if not KEY: + sys.exit("需要 AGENTMAIL_AGENT_KEY（在 Gateway 的管理页或 admin API 上建）") + + +def api(path, body=None, method=None): + """打一次 Gateway。密钥走 Authorization 头，与插件完全一致。""" + req = urllib.request.Request( + GW + path, + data=json.dumps(body).encode() if body is not None else None, + headers={"Authorization": "Bearer " + KEY, "Content-Type": "application/json"}, + method=method or ("POST" if body is not None else "GET"), + ) + try: + with urllib.request.urlopen(req, timeout=20) as r: + return json.loads(r.read() or b"{}") + except urllib.error.HTTPError as e: + # Gateway 的错误体是 {"error": "..."}，把它读出来 —— + # 不读的话只剩一个 "HTTP Error 400"，看不出是哪个字段不对 + detail = e.read().decode(errors="replace")[:300] + raise SystemExit("%s %s -> HTTP %s %s" % (req.method, path, e.code, detail)) + + +def register(): + """注册。两个容易踩的地方： + + - 路由是 `/agent/register`（**单数**），复数会 404 + - `workspaces` 是对象数组 `[{name, path}]`，不是字符串数组； + 传字符串会得到 400 "请求体 JSON 解析失败" + """ + return api("/agent/register", { + "name": NAME, + "platform": "stdlib-demo", + "workspaces": [{"name": "demo", "path": WORKSPACE}], + }) + + +def heartbeat(): + """心跳。models 是这台机器上「有」的模型目录； + 响应回传管理员划定的范围，真正的插件据此决定按什么顺序尝试。""" + return api("/agent/heartbeat", { + "models": [ + {"provider": "stdlib", "model": "echo-1", "display_name": "Echo（演示）"}, + ], + }) + + +def reply_to_unread(event): + """收到 new_mail 就把未读的都回一封，然后标记已读。 + + 标记已读不能省：不标的话下一轮会把同一批邮件再捞一次。 + """ + box = api("/mail/inbox?status=unread&limit=5") + mails = box.get("mails", box if isinstance(box, list) else []) + for m in mails: + print(" 收到:", m.get("subject"), "| 工作目录:", event.get("to_workspace")) + api("/mail/send", { + "to": m.get("from_name"), + "subject": "Re: " + (m.get("subject") or ""), + "body": ( + "这封回信来自另一台主机上的一个纯标准库脚本，" + "它没有安装 AgentMail 的任何代码。\n\n" + "- 收到的工作目录: `%s`\n" + "- 原邮件 ID: `%s`\n\n" + "跨主机在协议层面已经成立：Agent 主动连 Gateway，" + "只需要一个公网 URL 加一把密钥。" + % (event.get("to_workspace"), m.get("mail_id")) + ), + "reply_to": m.get("mail_id"), + }) + api("/mail/read", {"mail_ids": [m.get("mail_id")]}) + print(" 已回信") + + +def stream(): + """SSE 长连。真正的插件还要处理断线重连与 Last-Event-ID 补投。""" + req = urllib.request.Request( + GW + "/events/stream", + headers={"Authorization": "Bearer " + KEY, "Accept": "text/event-stream"}, + ) + with urllib.request.urlopen(req, timeout=300) as r: + etype = None + for raw in r: + line = raw.decode(errors="replace").rstrip("\n") + if line.startswith("event: "): + etype = line[7:] + elif line.startswith("data: ") and etype: + data = json.loads(line[6:]) + print("SSE <-", etype, json.dumps(data, ensure_ascii=False)[:100]) + if etype == "new_mail": + try: + reply_to_unread(data) + except Exception as e: # noqa: BLE001 —— 一封失败不该断掉长连 + print(" 回信失败:", e) + etype = None + + +def main(): + os.makedirs(WORKSPACE, exist_ok=True) + print("Gateway:", GW, "| 身份:", NAME) + print("注册:", json.dumps(register(), ensure_ascii=False)[:120]) + + hb = heartbeat() + print("心跳: pending=%s models_synced=%s allowed=%s unrestricted=%s" % ( + hb.get("pending_mails"), + hb.get("models_synced"), + [f"{m['provider']}/{m['model']}" for m in hb.get("allowed_models", [])], + hb.get("models_unrestricted"), + )) + + threading.Thread(target=stream, daemon=True).start() + + # 启动时先补拉一次：SSE 只推连上之后的事件， + # 离线期间到的邮件只能从心跳的 pending_mails 得知。 + # 不补这一次的后果：重启前发的邮件永远不会被处理。 + if hb.get("pending_mails"): + print("启动补拉 %s 封未读" % hb["pending_mails"]) + try: + reply_to_unread({"to_workspace": WORKSPACE}) + except Exception as e: # noqa: BLE001 + print(" 补拉失败:", e) + + seconds = float(sys.argv[1]) if len(sys.argv) > 1 else 120 + deadline = time.time() + seconds + # 30 秒一次心跳：Gateway 靠 last_seen 判在线 + while time.time() < deadline: + time.sleep(min(30, max(1, deadline - time.time()))) + try: + heartbeat() + except Exception as e: # noqa: BLE001 + print("心跳失败:", e) + print("结束") + + +if __name__ == "__main__": + main() diff --git a/docs/API.md b/docs/API.md index b781022..3e59dd3 100644 --- a/docs/API.md +++ b/docs/API.md @@ -308,6 +308,15 @@ POST /attachments 上传附件 GET /attachments/{id} 下载附件 POST /sessions/{id}/sync 回写平台侧生成的会话标题/slug GET /agent/models/allowed 读当前生效的模型范围（通常不需要——心跳已回传） + +注册体的 `workspaces` 是**对象数组**，不是字符串数组： + +```json +{"name":"mybot","platform":"stdlib","workspaces":[{"name":"demo","path":"/tmp/ws"}]} +``` + +两个官方插件都传 `workspaces: []`（工作目录由每封邮件的地址 path 位决定， +见「心跳与平台会话快照」一节的 `to_workspace`）。 ``` 发信与转发扣**本任务（会话）的往返预算**。 @@ -484,6 +493,21 @@ curl -N {host}/api/v1/events/stream -H "Authorization: Bearer $TOKEN" ## 六、错误约定 +### 400 的信息指向具体字段 + +请求体解析失败时不再回一句笼统的 `Invalid JSON`，而是说出是哪个字段、 +期望什么、收到什么： + +```json +{"error": "字段 \"workspaces\" 类型不对：期望 object，收到 string"} +{"error": "JSON 语法错误（第 8 字节处）"} +{"error": "请求体为空"} +``` + +期望类型用 JSON 的说法（`object` / `string` / `number` / `boolean` / `... 数组`）， +不回显 Go 类型名——那是本侧的实现细节。 + + 失败响应统一为 `{"error": "中文可操作描述"}`，状态码： | 码 | 含义 | diff --git a/docs/PHASE7-REMAINING.md b/docs/PHASE7-REMAINING.md index 0683601..55c5492 100644 --- a/docs/PHASE7-REMAINING.md +++ b/docs/PHASE7-REMAINING.md @@ -2,12 +2,84 @@ 7.7 DSH 插件已完成（见 `docs/PLUGIN-GUIDE.md` 与 PLAN.md §7.7）。 -## 无法立即推进（缺基础设施） +## 7.8 跨主机 Agent —— 协议层面已支持 -### 7.8 跨主机 Agent 发现 -- Gateway + Registry 拆分为独立服务 -- etcd / Consul 服务注册与发现 -- Agent 跨主机路由 +**原计划**（Gateway + Registry 拆分、etcd/Consul 服务注册、跨主机路由）**不做**。 +它解决的是「Gateway 怎么找到 Agent」，而这个方向从一开始就不成立： + +**连接方向是单向的 —— Agent 主动连 Gateway，Gateway 从不外呼。** +因此「发现」不是 Gateway 的问题，是 Agent 的配置问题：它只需要知道一个 +公网 URL 加一把密钥。注册中心要解决的「被叫方在哪」在这个架构里不存在 +—— 被叫方自己会打进来。 + +同一个理由让平台会话同步走插件上报（见 PLUGIN-GUIDE §4）： +Gateway 不外呼，就不需要知道任何 Agent 的地址。 + +### 已验证可用（2026-09-02，从 192.168.2.106 打到公网） + +完整一轮往返跑通了：`admin` 在本机发信给 `remotebot@/tmp/remotebot-ws`， +`.106` 上的脚本收到并回信入库。 + +| 能力 | 结果 | +|---|---| +| 注册（`POST /agent/register`） | 通，`workspaces` 落库为 `[{"name":"demo","path":"/tmp/remotebot-ws"}]` | +| 心跳（`POST /agent/heartbeat`） | 通，`models_synced: 1`、回传 `allowed_models` 与 `models_unrestricted` | +| SSE 长连（`GET /events/stream` + Bearer） | 通，收到 `connected` | +| 收件箱 + 标记已读 | 通 | +| 发信（`POST /mail/send` 带 `reply_to`） | 通，`from_workspace` 正确 | +| Gateway 侧在线状态 | `status=online`，`last_seen` 随心跳推进 | + +验证用的是一个**只依赖 python 标准库的脚本**（`deploy/remote-agent-demo.py`）， +它没装 AgentMail 的任何代码。这就是「协议层面已支持」的含义： +跨主机不需要新组件，只需要三个环境变量。 + +### 写这个脚本时踩的两个坑（新平台接入会重复踩） + +**`workspaces` 是对象数组 `[{name, path}]`，不是字符串数组。** +传字符串原先只得到一句固定的 400 `Invalid JSON` —— 完全没指向是哪个字段， +只能靠翻服务端结构体才能发现。两个正式插件都传 `workspaces: []`， +所以这个坑一直没暴露过；第三方客户端没有「翻服务端源码」这个条件。 + +**已修**：新增 `handler.DecodeBody`，22 处 `Decode` + 固定文案的调用点全部换过去。 +现在同样的请求回： + +```json +{"error": "字段 \"workspaces\" 类型不对：期望 object，收到 string"} +``` + +刻意不回显 `encoding/json` 的原文——它带 Go 类型名（`models.Workspace`）， +那是本侧的实现细节，不该出现在公开 API 的响应里。 +`internal/handler/decode_test.go` 钉住这一点（含「不得泄漏 Go 类型名」的断言）。 + +**SSE 只推连上之后的事件，离线期间的邮件要靠心跳的 `pending_mails` 补拉。** +第一版脚本只挂了 SSE，于是启动前发的那封邮件永远不会被处理 —— +日志里 `pending=1` 明明写着有一封未读，却没人去拉。 + +**这个坑两个正式插件也有**（原以为它们做了补拉，查了才发现没有）。已修： +新增共用模块 `lib/catchup.js`，两插件在首个成功心跳后补投一次。 + +- 只在**首个**心跳后补，不是每轮：每轮都补会把「模型正在处理中、尚未标已读」 + 的邮件重复投递 +- 串行投递、一次最多 5 封：每封都要起一轮模型，并发放出去等于对上游打 N 个 + 并发请求，且最后那几封要等前面全部跑完 +- 与 SSE 共用 `deliveredMails` 去重：心跳与 SSE 建连之间有个窗口， + 那期间到的邮件既在 `pending_mails` 里也会被 SSE 推一次 +- 按时间**正序**补投（收件箱是倒序返回的）：同一会话里的多封邮件倒着塞进去， + 上下文顺序是乱的 +- `permission` 类邮件不补投：原来的工具调用早随进程一起没了， + 投过去模型没有可恢复的上下文 + +端到端验证（两平台各一次）：停插件 → 发邮件 → 启插件 → 日志出现 +「补投 1 封离线期间的邮件」→ 回信入库。随后在线状态再发一封确认只回一次。 + +### 剩下的确实是运维便利，不是能力缺失 + +- [ ] 一条命令为远端主机建密钥并打印那三个环境变量（现在要手工调 admin API） +- [ ] 密钥轮换（现在换密钥要重启远端 agent） +- [ ] Agent 列表显示来源主机（`agents.host_url` 列已存在但没人写， + 要写的话应当由心跳带上自报的地址 —— 仍然不是 Gateway 去探测） + +这三项都不阻塞跨主机使用，因此不再归入 Phase 7。 ## 可以立即推进的生产缺陷 diff --git a/docs/PLAN.md b/docs/PLAN.md index 5a97b66..e0b5285 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -1120,11 +1120,21 @@ execute: { 与第三方客户端），新增同序的 `candidates`；过滤时标题也参与匹配 —— 人记得的是「缓存选型」而不是 `brisk-harbor` 这种随机短名 -### 7.8 跨主机 Agent 发现 +### 7.8 跨主机 Agent（协议层面已支持，注册中心不做） -- [ ] Gateway + Registry 拆分为独立服务 -- [ ] etcd / Consul 服务注册 -- [ ] Agent 跨主机路由 +原计划的 Gateway + Registry 拆分与 etcd/Consul 注册**取消**。 +它要解决「Gateway 怎么找到 Agent」，而这个问题在本架构里不存在： +**连接方向是单向的 —— Agent 主动连 Gateway，Gateway 从不外呼。** +远端 Agent 只需要一个公网 URL 加一把密钥，被叫方自己会打进来。 + +- [x] Agent 从另一台主机经公网完成注册 / 心跳 / 收件箱 / SSE 长连 + （2026-09-02 用 `deploy/remote-agent-demo.py` 验证，纯标准库 60 行） +- [x] 心跳带模型目录与平台会话快照同样跨主机可用 +- [ ] （运维便利，不阻塞）一条命令为远端主机建密钥并打印环境变量 +- [ ] （运维便利，不阻塞）密钥轮换 +- [ ] （运维便利，不阻塞）`agents.host_url` 由心跳自报填充 + +细节见 `docs/PHASE7-REMAINING.md` 的「7.8 跨主机 Agent」一节。 ### 7.9 插件自动转发 + 会话往返预算（已完成） diff --git a/docs/PLUGIN-GUIDE.md b/docs/PLUGIN-GUIDE.md index 7212f76..9daa211 100644 --- a/docs/PLUGIN-GUIDE.md +++ b/docs/PLUGIN-GUIDE.md @@ -432,6 +432,23 @@ opencode 还有个额外问题：**插件是懒加载的**，进程起来了插 --- +### `workspaces` 是对象数组 + +注册体的 `workspaces` 要 `[{name, path}]`。传字符串数组会 400。 +两个官方插件都传 `[]`（工作目录由每封邮件的 `to_workspace` 决定）， +所以照抄它们不会踩；自己从 API 文档写起就会。 + +### 启动时要主动拉一次收件箱 + +SSE 只推连上之后的事件。插件重启前发来的邮件不会再推一次 —— +心跳响应的 `pending_mails` 是唯一线索。不补拉的后果： +那封邮件永远躺在收件箱里，发件人以为 Agent 收到了。 + +共用实现 `lib/catchup.js` 的 `selectCatchup(mails, seen, max)`。四条约束： +只在**首个**心跳后补（每轮都补会重复投递正在处理中的邮件）、串行且一次最多 5 封 +（每封都要起一轮模型）、与 SSE 共用一个 `deliveredMails` 集合去重 +（建连窗口期的邮件两条路都会到）、按时间**正序**投（收件箱是倒序返回的）。 + ## 八、新平台适配清单 ``` @@ -470,6 +487,8 @@ opencode 还有个额外问题：**插件是懒加载的**，进程起来了插 [ ] **等异步结论**再判成功／失败（失败不是同步抛的！） [ ] 全部失败 → renderFailureReport + relay:'summary' 发信 +[ ] 6.5 启动补拉：心跳 pending_mails > 0 时先处理存量未读 + [ ] 7. 命名与快照 [ ] alias/title 回写 POST /sessions/{id}/sync [ ] slug 去掉 . @ / 等寻址分隔符 @@ -496,6 +515,7 @@ opencode 还有个额外问题：**插件是懒加载的**，进程起来了插 | `session-snapshot.js` | 平台会话快照整理（含 subagent 过滤、slug 派生） | | `workspace.js` | 寻址 path 位 → 可用的 cwd | | `model-scope.js` | 模型目录整理 + 降级顺序 + 失败报告 | +| `catchup.js` | 离线期间积压邮件的补投选择 | | `message.js` | DSH 的消息构造与会话日志读取（DSH 专用） | `test/` 下对应的测试文件同样逐字节共用。 diff --git a/gateway/internal/handler/agents.go b/gateway/internal/handler/agents.go index a0a8d19..2531560 100644 --- a/gateway/internal/handler/agents.go +++ b/gateway/internal/handler/agents.go @@ -54,8 +54,7 @@ type heartbeatRequest struct { // 2. body 里带 secret —— 旧方式，兼容保留。 func RegisterAgent(w http.ResponseWriter, r *http.Request) { var req registerRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.Name == "" { diff --git a/gateway/internal/handler/auth.go b/gateway/internal/handler/auth.go index 8e74792..099c668 100644 --- a/gateway/internal/handler/auth.go +++ b/gateway/internal/handler/auth.go @@ -70,8 +70,7 @@ type setupRequest struct { // POST /api/v1/setup/admin —— 公开，但仅在系统无任何用户时可用 func SetupAdmin(w http.ResponseWriter, r *http.Request) { var req setupRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if len(req.Password) < 8 { @@ -110,8 +109,7 @@ func SetupAdmin(w http.ResponseWriter, r *http.Request) { // POST /api/v1/auth/login func Login(w http.ResponseWriter, r *http.Request) { var req loginRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } name := strings.ToLower(strings.TrimSpace(req.Username)) @@ -192,8 +190,7 @@ func ChangePassword(w http.ResponseWriter, r *http.Request) { } var req changePasswordRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if len(req.NewPassword) < 8 { @@ -240,8 +237,7 @@ type createUserRequest struct { // POST /api/v1/admin/users func AdminCreateUser(w http.ResponseWriter, r *http.Request) { var req createUserRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if len(req.Password) < 8 { @@ -280,8 +276,7 @@ func AdminUpdateUser(w http.ResponseWriter, r *http.Request) { return } var req updateUserRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } @@ -361,8 +356,7 @@ func AdminResetPassword(w http.ResponseWriter, r *http.Request) { return } var req resetPasswordRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if len(req.NewPassword) < 8 { diff --git a/gateway/internal/handler/contacts.go b/gateway/internal/handler/contacts.go index 3e2fb26..343ca2a 100644 --- a/gateway/internal/handler/contacts.go +++ b/gateway/internal/handler/contacts.go @@ -52,8 +52,7 @@ func ArchiveContact(w http.ResponseWriter, r *http.Request) { } var req archiveRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } diff --git a/gateway/internal/handler/decode_test.go b/gateway/internal/handler/decode_test.go new file mode 100644 index 0000000..5f804fe --- /dev/null +++ b/gateway/internal/handler/decode_test.go @@ -0,0 +1,138 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/agentmail/gateway/internal/models" +) + +// 400 的信息必须指向具体字段。 +// +// 起因：写跨主机验证脚本时把 workspaces 传成了字符串数组， +// 服务端回的是一句固定的 "Invalid JSON" —— 只能靠翻服务端结构体才发现是哪个字段。 +// 第三方客户端没有这个条件。 +func TestDecodeBodyErrorNamesTheField(t *testing.T) { + type body struct { + Name string `json:"name"` + Workspaces []models.Workspace `json:"workspaces"` + } + + cases := []struct { + name string + payload string + wantHas []string + wantMiss []string + }{ + { + name: "字段类型不对要说出字段名与期望类型", + payload: `{"name":"bot","workspaces":["/tmp/ws"]}`, + // 期望能看出：是 workspaces，要的是 object 数组，给的是 string + wantHas: []string{"workspaces", "object", "string"}, + // 不该把 Go 类型名漏出去 + wantMiss: []string{"models.Workspace", "[]models"}, + }, + { + name: "整个体的类型不对", + payload: `["not","an","object"]`, + wantHas: []string{"object"}, + }, + { + // 截断的 JSON 走的是 io.ErrUnexpectedEOF，不是 json.SyntaxError + name: "被截断的体要说明是截断", + payload: `{"name":`, + wantHas: []string{"语法", "结束"}, + }, + { + name: "非法字符要给出位置", + payload: `{"name":1x}`, + wantHas: []string{"语法", "字节"}, + }, + { + name: "空体单独说明", + payload: ``, + wantHas: []string{"为空"}, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := httptest.NewRequest(http.MethodPost, "/x", strings.NewReader(c.payload)) + w := httptest.NewRecorder() + + var v body + if DecodeBody(w, r, &v) { + t.Fatal("这个体应当解析失败") + } + if w.Code != http.StatusBadRequest { + t.Fatalf("状态码应为 400，实际 %d", w.Code) + } + + var resp map[string]string + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("响应不是 JSON: %v", err) + } + msg := resp["error"] + if msg == "" { + t.Fatal("error 字段为空") + } + for _, want := range c.wantHas { + if !strings.Contains(msg, want) { + t.Errorf("信息里应含 %q，实际 %q", want, msg) + } + } + for _, miss := range c.wantMiss { + if strings.Contains(msg, miss) { + t.Errorf("信息里不该含 Go 类型名 %q：%q", miss, msg) + } + } + }) + } +} + +// 合法请求体不该被拦，也不该写任何响应 —— +// 写了的话调用方接着写自己的响应就成了两次 WriteHeader。 +func TestDecodeBodyPassesValidPayload(t *testing.T) { + type body struct { + Name string `json:"name"` + Workspaces []models.Workspace `json:"workspaces"` + } + + payload := `{"name":"bot","workspaces":[{"name":"demo","path":"/tmp/ws"}]}` + r := httptest.NewRequest(http.MethodPost, "/x", strings.NewReader(payload)) + w := httptest.NewRecorder() + + var v body + if !DecodeBody(w, r, &v) { + t.Fatalf("合法体被拒：%s", w.Body.String()) + } + if w.Body.Len() != 0 { + t.Errorf("成功时不该写响应体，实际写了 %q", w.Body.String()) + } + if v.Name != "bot" || len(v.Workspaces) != 1 || v.Workspaces[0].Path != "/tmp/ws" { + t.Errorf("解析结果不对：%+v", v) + } +} + +// 空数组是合法的 —— 两个正式插件注册时都传 workspaces: []。 +func TestDecodeBodyAcceptsEmptyWorkspaces(t *testing.T) { + type body struct { + Name string `json:"name"` + Workspaces []models.Workspace `json:"workspaces"` + } + + r := httptest.NewRequest(http.MethodPost, "/x", + strings.NewReader(`{"name":"opencode","workspaces":[]}`)) + w := httptest.NewRecorder() + + var v body + if !DecodeBody(w, r, &v) { + t.Fatalf("空 workspaces 被拒：%s", w.Body.String()) + } + if len(v.Workspaces) != 0 { + t.Errorf("应为空数组，实际 %+v", v.Workspaces) + } +} diff --git a/gateway/internal/handler/forward.go b/gateway/internal/handler/forward.go index d09cd9e..020b8a0 100644 --- a/gateway/internal/handler/forward.go +++ b/gateway/internal/handler/forward.go @@ -75,8 +75,7 @@ func forwardSubject(custom, original string) string { // actor 是转发者名（Agent 名或用户名），fromWorkspace 仅 Agent 有。 func doForward(w http.ResponseWriter, r *http.Request, mailID uuid.UUID, actor, fromWorkspace string, isAgent bool) { var req forwardRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if strings.TrimSpace(req.To) == "" { @@ -247,8 +246,7 @@ func AdminSetQuota(w http.ResponseWriter, r *http.Request) { } var req setQuotaRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } n := req.DefaultRounds diff --git a/gateway/internal/handler/helpers.go b/gateway/internal/handler/helpers.go index 7f6b533..707d17f 100644 --- a/gateway/internal/handler/helpers.go +++ b/gateway/internal/handler/helpers.go @@ -3,7 +3,10 @@ package handler import ( "encoding/json" "errors" + "io" "net/http" + "reflect" + "strconv" "strings" "unicode/utf8" @@ -27,6 +30,76 @@ func Decode(r *http.Request, v interface{}) error { return json.NewDecoder(r.Body).Decode(v) } +// DecodeBody 解析请求体，失败时直接写 400 并返回 false。 +// +// 与直接用 Decode 的区别是错误信息**指向具体字段**。原先 22 处调用点 +// 一律回一句固定的 "Invalid JSON"，客户端只知道「有问题」却不知道哪里有问题 —— +// 实测踩过一次：`workspaces` 要的是 `[{name, path}]`，传字符串数组得到的 +// 就是那句固定文案，只能靠翻服务端结构体才发现。第三方客户端没有这个条件。 +func DecodeBody(w http.ResponseWriter, r *http.Request, v interface{}) bool { + if err := Decode(r, v); err != nil { + Error(w, http.StatusBadRequest, decodeErrMsg(err)) + return false + } + return true +} + +// decodeErrMsg 把 json 解码错误翻成一句能照着改的话。 +// +// 刻意不回显 json 包的原文：它带 Go 的类型名（如 models.Workspace）， +// 那是本侧的实现细节，对调用方没有意义，也不该出现在公开 API 的响应里。 +func decodeErrMsg(err error) string { + if errors.Is(err, io.EOF) { + return "请求体为空" + } + // 截断的 JSON 走的不是 SyntaxError 而是 ErrUnexpectedEOF —— + // 不单独处理的话会落到最后那句笼统的兜底文案里 + if errors.Is(err, io.ErrUnexpectedEOF) { + return "JSON 语法错误：请求体在解析完成前就结束了（可能被截断）" + } + + var typeErr *json.UnmarshalTypeError + if errors.As(err, &typeErr) { + if typeErr.Field != "" { + return "字段 \"" + typeErr.Field + "\" 类型不对：期望 " + + jsonKindName(typeErr.Type) + "，收到 " + typeErr.Value + } + return "请求体类型不对：期望 " + jsonKindName(typeErr.Type) + "，收到 " + typeErr.Value + } + + var syntaxErr *json.SyntaxError + if errors.As(err, &syntaxErr) { + return "JSON 语法错误（第 " + strconv.FormatInt(syntaxErr.Offset, 10) + " 字节处）" + } + + return "请求体不是合法 JSON" +} + +// jsonKindName 把 Go 类型说成 JSON 的说法。 +// 调用方写的是 JSON，用 []models.Workspace 去解释它要的是什么毫无帮助。 +func jsonKindName(t reflect.Type) string { + if t == nil { + return "未知类型" + } + switch t.Kind() { + case reflect.Slice, reflect.Array: + return jsonKindName(t.Elem()) + " 数组" + case reflect.Map, reflect.Struct: + return "object" + case reflect.String: + return "string" + case reflect.Bool: + return "boolean" + case reflect.Ptr: + return jsonKindName(t.Elem()) + default: + if k := t.Kind(); k >= reflect.Int && k <= reflect.Float64 { + return "number" + } + return t.Kind().String() + } +} + // httpError 携带 HTTP 状态码的错误 type httpError struct { status int diff --git a/gateway/internal/handler/keys.go b/gateway/internal/handler/keys.go index 07760f7..5ffdcf9 100644 --- a/gateway/internal/handler/keys.go +++ b/gateway/internal/handler/keys.go @@ -50,8 +50,7 @@ func CreateAgentKey(w http.ResponseWriter, r *http.Request) { } var req createKeyRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } @@ -102,8 +101,7 @@ func BindAgentKey(w http.ResponseWriter, r *http.Request) { return } var req bindKeyRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } name := strings.TrimSpace(req.AgentName) @@ -129,8 +127,7 @@ func CreateMyKey(w http.ResponseWriter, r *http.Request) { } var req createKeyRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } diff --git a/gateway/internal/handler/mail.go b/gateway/internal/handler/mail.go index fcc0847..81249b4 100644 --- a/gateway/internal/handler/mail.go +++ b/gateway/internal/handler/mail.go @@ -125,8 +125,7 @@ func SendMail(w http.ResponseWriter, r *http.Request) { } var req sendMailRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.To == "" || req.Subject == "" || req.Body == "" { @@ -453,8 +452,7 @@ func MarkInboxRead(w http.ResponseWriter, r *http.Request) { var req markReadRequest // 允许空 body：`POST /mail/read` 不带任何内容 = 全部标掉 if r.ContentLength > 0 { - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } } diff --git a/gateway/internal/handler/me.go b/gateway/internal/handler/me.go index e957259..b73ddb3 100644 --- a/gateway/internal/handler/me.go +++ b/gateway/internal/handler/me.go @@ -40,8 +40,7 @@ func MeSendMail(w http.ResponseWriter, r *http.Request) { } var req meSendMailRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.To == "" || req.Subject == "" || req.Body == "" { diff --git a/gateway/internal/handler/models_scope.go b/gateway/internal/handler/models_scope.go index b417c42..c1a1263 100644 --- a/gateway/internal/handler/models_scope.go +++ b/gateway/internal/handler/models_scope.go @@ -83,8 +83,7 @@ func AdminSetAgentModels(w http.ResponseWriter, r *http.Request) { var req struct { Models []repo.ModelRef `json:"models"` } - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if len(req.Models) > maxAllowedModels { diff --git a/gateway/internal/handler/permission.go b/gateway/internal/handler/permission.go index 199589f..99dde93 100644 --- a/gateway/internal/handler/permission.go +++ b/gateway/internal/handler/permission.go @@ -43,8 +43,7 @@ func RequestPermission(w http.ResponseWriter, r *http.Request) { } var req permissionRequestRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.Question == "" { @@ -165,8 +164,7 @@ func DecidePermission(w http.ResponseWriter, r *http.Request) { } var req permissionDecideRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.MailID == "" || req.Decision == "" { diff --git a/gateway/internal/handler/sessions.go b/gateway/internal/handler/sessions.go index 1eea27b..b055573 100644 --- a/gateway/internal/handler/sessions.go +++ b/gateway/internal/handler/sessions.go @@ -107,8 +107,7 @@ func UpdateSessionAlias(w http.ResponseWriter, r *http.Request) { } var req updateAliasRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } alias := strings.TrimSpace(req.Alias) @@ -176,8 +175,7 @@ func SyncSession(w http.ResponseWriter, r *http.Request) { } var req syncSessionRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } @@ -307,8 +305,7 @@ func UpdateSessionBudget(w http.ResponseWriter, r *http.Request) { return } var req sessionBudgetRequest - if err := Decode(r, &req); err != nil { - Error(w, http.StatusBadRequest, "Invalid JSON") + if !DecodeBody(w, r, &req) { return } if req.MaxRounds == nil && !req.Reset { diff --git a/plugins/dsh-mail-bridge/lib/catchup.d.ts b/plugins/dsh-mail-bridge/lib/catchup.d.ts new file mode 100644 index 0000000..61ef006 --- /dev/null +++ b/plugins/dsh-mail-bridge/lib/catchup.d.ts @@ -0,0 +1,20 @@ +export declare const MAX_CATCHUP: number; + +export interface CatchupEvent { + mail_id: string; + session_id: string; + from_name: string; + subject: string; + mail_type: string; + role: string; + to_workspace: string; + catchup: true; +} + +export declare function mailToEvent(mail: any): CatchupEvent; + +export declare function selectCatchup( + mails: any, + seen: Set | undefined, + max?: number, +): CatchupEvent[]; diff --git a/plugins/dsh-mail-bridge/lib/catchup.js b/plugins/dsh-mail-bridge/lib/catchup.js new file mode 100644 index 0000000..a0a4a6d --- /dev/null +++ b/plugins/dsh-mail-bridge/lib/catchup.js @@ -0,0 +1,74 @@ +/** + * 启动补拉：把插件离线期间到的邮件变成与 SSE 事件同形的投递任务。 + * + * 为什么需要它：**SSE 只推连上之后的事件**。插件重启前发来的邮件不会再推一次， + * 心跳响应的 `pending_mails` 是唯一线索。不补拉的后果是那封邮件永远躺在 + * 收件箱里，而发件人以为 Agent 收到了 —— 这比明确的失败更难排查。 + * + * 两个平台共用，必须逐字节相同（deploy/check-shared-libs.sh 校验）。 + */ + +/** + * 一次补拉最多处理几封。 + * + * 上限存在的理由：每封都要起一轮模型。攒了 80 封的时候一次性全放出去， + * 等于对上游打 80 个并发请求，且最后那几封要等前面全部跑完。 + * 超出的部分留在收件箱里，下次重启或人工触发时再处理。 + */ +export const MAX_CATCHUP = 5; + +/** + * 把收件箱里的一封邮件转成 SSE `new_mail` 那个形状。 + * + * 补拉与 SSE 走同一条投递路径（deliverMail），因此形状必须一致 —— + * 两条路径各写一遍投递逻辑的话，某一条上的修复会漏掉另一条。 + * + * @param {any} mail `/mail/inbox` 返回的一行 + * @returns {{mail_id: string, session_id: string, from_name: string, + * subject: string, mail_type: string, role: string, + * to_workspace: string, catchup: true}} + */ +export function mailToEvent(mail) { + return { + mail_id: mail?.mail_id || '', + session_id: mail?.session_id || '', + from_name: mail?.from_name || '', + subject: mail?.subject || '', + mail_type: mail?.mail_type || 'normal', + role: 'to', + to_workspace: mail?.to_workspace || '', + // 标记来源，投递侧可据此决定是否在提示词里说明「这是积压的邮件」 + catchup: true, + }; +} + +/** + * 从收件箱挑出该补投的邮件。 + * + * @param {any[]} mails `/mail/inbox?status=unread` 的结果 + * @param {Set} seen 已经通过 SSE 投过的 mail_id（避免重复投递） + * @param {number} [max] 上限，默认 MAX_CATCHUP + * @returns {any[]} 与 SSE 事件同形的投递任务，按时间正序（老的先处理） + */ +export function selectCatchup(mails, seen, max = MAX_CATCHUP) { + if (!Array.isArray(mails) || mails.length === 0) return []; + + const picked = []; + for (const m of mails) { + const id = m?.mail_id; + if (!id) continue; + // 心跳与 SSE 建连之间有个窗口：那期间到的邮件既在 pending_mails 里、 + // 也会被 SSE 推一次。不去重就会投两遍，模型回两封信。 + if (seen && seen.has(id)) continue; + // permission 类邮件不补投：它是给人看的询问，Agent 侧没有可恢复的上下文 + // （原来的工具调用早随进程一起没了），投过去只会让模型困惑。 + if (m?.mail_type && m.mail_type !== 'normal') continue; + picked.push(m); + } + + // 收件箱按时间倒序返回，补投要按正序 —— 先来的先处理， + // 否则同一会话里的多封邮件会被倒着塞进去，上下文顺序是乱的。 + picked.reverse(); + + return picked.slice(0, Math.max(0, max)).map(mailToEvent); +} diff --git a/plugins/dsh-mail-bridge/src/index.ts b/plugins/dsh-mail-bridge/src/index.ts index 4c22c62..593bf96 100644 --- a/plugins/dsh-mail-bridge/src/index.ts +++ b/plugins/dsh-mail-bridge/src/index.ts @@ -34,6 +34,7 @@ import { renderFailureReport, } from '../lib/model-scope.js'; import { resolveWorkspaceCwd, ensureCwd, mailSessionFallback } from '../lib/workspace.js'; +import { selectCatchup } from '../lib/catchup.js'; import { renderInbox, idsToMarkRead, @@ -267,6 +268,40 @@ export function apply(ctx: any, config: PluginConfig): void { } } + // 已经投过的 mail_id。心跳与 SSE 建连之间有个窗口：那期间到的邮件 + // 既在 pending_mails 里、也会被 SSE 推一次 —— 不去重就会投两遍。 + const deliveredMails = new Set(); + let caughtUp = false; + + /** + * 补投离线期间积压的未读邮件。 + * + * SSE 只推连上之后的事件，插件重启前发来的邮件不会再推一次。 + * 不补的话那封邮件永远躺在收件箱里，而发件人以为 Agent 收到了。 + */ + async function catchUp(pending: unknown): Promise { + if (!pending) return; + try { + const box = await client.get('/mail/inbox?status=unread&limit=20'); + const tasks = selectCatchup(box?.mails ?? box, deliveredMails); + if (tasks.length === 0) return; + console.error(`[dsh-mail-bridge] 补投 ${tasks.length} 封离线期间的邮件（共 ${pending} 封未读）`); + // 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求 + for (const ev of tasks) { + // 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封 + if (deliveredMails.has(ev.mail_id)) continue; + deliveredMails.add(ev.mail_id); + try { + await deliverMail(ev, 'mail'); + } catch (e: any) { + ctx.logger.error(`[dsh-mail-bridge] 补投 ${ev.mail_id} 失败: ${e?.message || e}`); + } + } + } catch (e: any) { + ctx.logger.error(`[dsh-mail-bridge] 补投失败: ${e?.message || e}`); + } + } + async function beat(): Promise { const body: Record = {}; const [entries, models] = await Promise.all([collectSessions(), collectModels()]); @@ -279,6 +314,12 @@ export function apply(ctx: any, config: PluginConfig): void { // 生效的模型范围随心跳响应回传：管理员在配置页改了范围后， // 插件最多一个周期（30 秒）就能看到新值，不需要重启。 if (Array.isArray(res?.allowed_models)) allowedModels = res.allowed_models; + // 只在首个成功的心跳后补投一次：之后的积压都由 SSE 覆盖， + // 每轮心跳都补的话会把「模型正在处理中、尚未标已读」的邮件重复投递。 + if (!caughtUp) { + caughtUp = true; + await catchUp(res?.pending_mails); + } } catch { // 心跳失败不报错：网络抖动很常见，下一轮会补上。 // 真的持续连不上时 Gateway 会把它判成离线，那才是可见的信号。 @@ -858,6 +899,7 @@ export function apply(ctx: any, config: PluginConfig): void { startSSE((type, data) => { switch (type) { case 'new_mail': + if (data?.mail_id) deliveredMails.add(data.mail_id); deliverMail(data, 'mail') .then(({ sessionID, reused }) => { console.error(`[dsh-mail-bridge] ${type} -> ${reused ? '续谈' : '新会话'} ${sessionID}`); diff --git a/plugins/dsh-mail-bridge/test/catchup.test.mjs b/plugins/dsh-mail-bridge/test/catchup.test.mjs new file mode 100644 index 0000000..23d38ab --- /dev/null +++ b/plugins/dsh-mail-bridge/test/catchup.test.mjs @@ -0,0 +1,81 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { MAX_CATCHUP, mailToEvent, selectCatchup } from '../lib/catchup.js'; + +const mail = (over = {}) => ({ + mail_id: 'm1', + session_id: 's1', + from_name: 'admin', + subject: '主题', + mail_type: 'normal', + to_workspace: '/tmp/ws', + ...over, +}); + +test('mailToEvent 产出与 SSE new_mail 同形的对象', () => { + const ev = mailToEvent(mail()); + // 投递侧读的就是这几个键，形状不一致会让补拉那条路径静默地少带信息 + for (const k of ['mail_id', 'session_id', 'from_name', 'subject', 'mail_type', 'to_workspace']) { + assert.ok(k in ev, `缺少 ${k}`); + } + assert.equal(ev.role, 'to'); + assert.equal(ev.catchup, true); +}); + +test('mailToEvent 对缺字段的行给出空串而非 undefined', () => { + const ev = mailToEvent({}); + assert.equal(ev.mail_id, ''); + assert.equal(ev.to_workspace, ''); + assert.equal(ev.mail_type, 'normal'); +}); + +test('已经通过 SSE 投过的不再补投', () => { + const mails = [mail({ mail_id: 'a' }), mail({ mail_id: 'b' })]; + const got = selectCatchup(mails, new Set(['a'])); + assert.deepEqual(got.map(e => e.mail_id), ['b']); +}); + +test('按时间正序补投（收件箱是倒序返回的）', () => { + // 收件箱：新的在前 + const mails = [mail({ mail_id: 'new' }), mail({ mail_id: 'mid' }), mail({ mail_id: 'old' })]; + const got = selectCatchup(mails, new Set()); + assert.deepEqual( + got.map(e => e.mail_id), + ['old', 'mid', 'new'], + '先来的邮件必须先处理，否则同一会话里的上下文顺序是乱的', + ); +}); + +test('permission 类邮件不补投', () => { + const mails = [mail({ mail_id: 'p', mail_type: 'permission' }), mail({ mail_id: 'n' })]; + const got = selectCatchup(mails, new Set()); + assert.deepEqual(got.map(e => e.mail_id), ['n']); +}); + +test('超过上限的部分留在收件箱里', () => { + const mails = Array.from({ length: MAX_CATCHUP + 4 }, (_, i) => mail({ mail_id: 'm' + i })); + const got = selectCatchup(mails, new Set()); + assert.equal(got.length, MAX_CATCHUP, '一次补拉不该把几十封邮件同时放出去'); +}); + +test('上限可显式压到 0（用于禁用补拉）', () => { + const got = selectCatchup([mail()], new Set(), 0); + assert.deepEqual(got, []); +}); + +test('空输入与非数组不炸', () => { + assert.deepEqual(selectCatchup([], new Set()), []); + assert.deepEqual(selectCatchup(undefined, new Set()), []); + assert.deepEqual(selectCatchup(null, new Set()), []); +}); + +test('没有 mail_id 的行跳过', () => { + const got = selectCatchup([mail({ mail_id: '' }), mail({ mail_id: 'ok' })], new Set()); + assert.deepEqual(got.map(e => e.mail_id), ['ok']); +}); + +test('seen 传 undefined 时不去重也不报错', () => { + const got = selectCatchup([mail({ mail_id: 'x' })], undefined); + assert.deepEqual(got.map(e => e.mail_id), ['x']); +}); diff --git a/plugins/opencode-mail-bridge/index.js b/plugins/opencode-mail-bridge/index.js index dbb606f..33ac039 100644 --- a/plugins/opencode-mail-bridge/index.js +++ b/plugins/opencode-mail-bridge/index.js @@ -7,6 +7,7 @@ import { join, dirname, basename } from "node:path"; // 都当成插件工厂，入口文件多导出一个东西就会 "Plugin export is not a function"。 import { snapshotOpencodeSessions } from "./lib/session-snapshot.js"; import { resolveWorkspaceCwd } from "./lib/workspace.js"; +import { selectCatchup } from "./lib/catchup.js"; import { snapshotOpencodeModels, modelAttemptOrder, @@ -845,6 +846,42 @@ export default async function mailBridge(input) { } } + // 已经投过的 mail_id。心跳与 SSE 建连之间有个窗口：那期间到的邮件 + // 既在 pending_mails 里、也会被 SSE 推一次 —— 不去重就会投两遍。 + const deliveredMails = new Set(); + + /** + * 补投离线期间积压的未读邮件。 + * + * SSE 只推连上之后的事件，插件重启前发来的邮件不会再推一次。 + * 不补的话那封邮件永远躺在收件箱里，而发件人以为 Agent 收到了。 + */ + async function catchUp(pending) { + if (!pending) return; + try { + const box = await apiGet("/mail/inbox?status=unread&limit=20"); + const tasks = selectCatchup(box?.mails ?? box, deliveredMails); + if (tasks.length === 0) return; + console.error(`[mail-bridge] 补投 ${tasks.length} 封离线期间的邮件（共 ${pending} 封未读）`); + // 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求 + for (const ev of tasks) { + // 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封 + // （selectCatchup 只在拉完那一刻去过重） + if (deliveredMails.has(ev.mail_id)) continue; + deliveredMails.add(ev.mail_id); + try { + await deliverMail(client, directory, ev, "mail"); + } catch (e) { + console.error(`[mail-bridge] 补投 ${ev.mail_id} 失败:`, e?.message || e); + } + } + } catch (e) { + console.error("[mail-bridge] 补投失败:", e?.message || e); + } + } + + let caughtUp = false; + const beat = async () => { const [platform_sessions, models] = await Promise.all([ reportSessions(), @@ -858,6 +895,12 @@ export default async function mailBridge(input) { // 生效的模型范围随心跳响应回传：管理员在配置页改了范围后， // 插件最多一个周期（30 秒）就能看到新值，不需要重启。 if (Array.isArray(res?.allowed_models)) allowedModels = res.allowed_models; + // 只在首个成功的心跳后补投一次：之后的积压都由 SSE 覆盖， + // 每轮心跳都补的话会把「模型正在处理中、尚未标已读」的邮件重复投递。 + if (!caughtUp) { + caughtUp = true; + await catchUp(res?.pending_mails); + } } catch { // 心跳失败不报错：网络抖动很常见，下一轮会补上。 // 真的持续连不上时 Gateway 会把它判成离线，那才是可见的信号。 @@ -878,6 +921,7 @@ export default async function mailBridge(input) { } if (type !== "new_mail") return; + if (data?.mail_id) deliveredMails.add(data.mail_id); deliverMail(client, directory, data, "mail") .then(({ sessionID, reused }) => { console.error(`[mail-bridge] ${type} -> ${reused ? "续谈" : "新会话"} ${sessionID}`); diff --git a/plugins/opencode-mail-bridge/lib/catchup.js b/plugins/opencode-mail-bridge/lib/catchup.js new file mode 100644 index 0000000..a0a4a6d --- /dev/null +++ b/plugins/opencode-mail-bridge/lib/catchup.js @@ -0,0 +1,74 @@ +/** + * 启动补拉：把插件离线期间到的邮件变成与 SSE 事件同形的投递任务。 + * + * 为什么需要它：**SSE 只推连上之后的事件**。插件重启前发来的邮件不会再推一次， + * 心跳响应的 `pending_mails` 是唯一线索。不补拉的后果是那封邮件永远躺在 + * 收件箱里，而发件人以为 Agent 收到了 —— 这比明确的失败更难排查。 + * + * 两个平台共用，必须逐字节相同（deploy/check-shared-libs.sh 校验）。 + */ + +/** + * 一次补拉最多处理几封。 + * + * 上限存在的理由：每封都要起一轮模型。攒了 80 封的时候一次性全放出去， + * 等于对上游打 80 个并发请求，且最后那几封要等前面全部跑完。 + * 超出的部分留在收件箱里，下次重启或人工触发时再处理。 + */ +export const MAX_CATCHUP = 5; + +/** + * 把收件箱里的一封邮件转成 SSE `new_mail` 那个形状。 + * + * 补拉与 SSE 走同一条投递路径（deliverMail），因此形状必须一致 —— + * 两条路径各写一遍投递逻辑的话，某一条上的修复会漏掉另一条。 + * + * @param {any} mail `/mail/inbox` 返回的一行 + * @returns {{mail_id: string, session_id: string, from_name: string, + * subject: string, mail_type: string, role: string, + * to_workspace: string, catchup: true}} + */ +export function mailToEvent(mail) { + return { + mail_id: mail?.mail_id || '', + session_id: mail?.session_id || '', + from_name: mail?.from_name || '', + subject: mail?.subject || '', + mail_type: mail?.mail_type || 'normal', + role: 'to', + to_workspace: mail?.to_workspace || '', + // 标记来源，投递侧可据此决定是否在提示词里说明「这是积压的邮件」 + catchup: true, + }; +} + +/** + * 从收件箱挑出该补投的邮件。 + * + * @param {any[]} mails `/mail/inbox?status=unread` 的结果 + * @param {Set} seen 已经通过 SSE 投过的 mail_id（避免重复投递） + * @param {number} [max] 上限，默认 MAX_CATCHUP + * @returns {any[]} 与 SSE 事件同形的投递任务，按时间正序（老的先处理） + */ +export function selectCatchup(mails, seen, max = MAX_CATCHUP) { + if (!Array.isArray(mails) || mails.length === 0) return []; + + const picked = []; + for (const m of mails) { + const id = m?.mail_id; + if (!id) continue; + // 心跳与 SSE 建连之间有个窗口：那期间到的邮件既在 pending_mails 里、 + // 也会被 SSE 推一次。不去重就会投两遍，模型回两封信。 + if (seen && seen.has(id)) continue; + // permission 类邮件不补投：它是给人看的询问，Agent 侧没有可恢复的上下文 + // （原来的工具调用早随进程一起没了），投过去只会让模型困惑。 + if (m?.mail_type && m.mail_type !== 'normal') continue; + picked.push(m); + } + + // 收件箱按时间倒序返回，补投要按正序 —— 先来的先处理， + // 否则同一会话里的多封邮件会被倒着塞进去，上下文顺序是乱的。 + picked.reverse(); + + return picked.slice(0, Math.max(0, max)).map(mailToEvent); +} diff --git a/plugins/opencode-mail-bridge/test/catchup.test.mjs b/plugins/opencode-mail-bridge/test/catchup.test.mjs new file mode 100644 index 0000000..23d38ab --- /dev/null +++ b/plugins/opencode-mail-bridge/test/catchup.test.mjs @@ -0,0 +1,81 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { MAX_CATCHUP, mailToEvent, selectCatchup } from '../lib/catchup.js'; + +const mail = (over = {}) => ({ + mail_id: 'm1', + session_id: 's1', + from_name: 'admin', + subject: '主题', + mail_type: 'normal', + to_workspace: '/tmp/ws', + ...over, +}); + +test('mailToEvent 产出与 SSE new_mail 同形的对象', () => { + const ev = mailToEvent(mail()); + // 投递侧读的就是这几个键，形状不一致会让补拉那条路径静默地少带信息 + for (const k of ['mail_id', 'session_id', 'from_name', 'subject', 'mail_type', 'to_workspace']) { + assert.ok(k in ev, `缺少 ${k}`); + } + assert.equal(ev.role, 'to'); + assert.equal(ev.catchup, true); +}); + +test('mailToEvent 对缺字段的行给出空串而非 undefined', () => { + const ev = mailToEvent({}); + assert.equal(ev.mail_id, ''); + assert.equal(ev.to_workspace, ''); + assert.equal(ev.mail_type, 'normal'); +}); + +test('已经通过 SSE 投过的不再补投', () => { + const mails = [mail({ mail_id: 'a' }), mail({ mail_id: 'b' })]; + const got = selectCatchup(mails, new Set(['a'])); + assert.deepEqual(got.map(e => e.mail_id), ['b']); +}); + +test('按时间正序补投（收件箱是倒序返回的）', () => { + // 收件箱：新的在前 + const mails = [mail({ mail_id: 'new' }), mail({ mail_id: 'mid' }), mail({ mail_id: 'old' })]; + const got = selectCatchup(mails, new Set()); + assert.deepEqual( + got.map(e => e.mail_id), + ['old', 'mid', 'new'], + '先来的邮件必须先处理，否则同一会话里的上下文顺序是乱的', + ); +}); + +test('permission 类邮件不补投', () => { + const mails = [mail({ mail_id: 'p', mail_type: 'permission' }), mail({ mail_id: 'n' })]; + const got = selectCatchup(mails, new Set()); + assert.deepEqual(got.map(e => e.mail_id), ['n']); +}); + +test('超过上限的部分留在收件箱里', () => { + const mails = Array.from({ length: MAX_CATCHUP + 4 }, (_, i) => mail({ mail_id: 'm' + i })); + const got = selectCatchup(mails, new Set()); + assert.equal(got.length, MAX_CATCHUP, '一次补拉不该把几十封邮件同时放出去'); +}); + +test('上限可显式压到 0（用于禁用补拉）', () => { + const got = selectCatchup([mail()], new Set(), 0); + assert.deepEqual(got, []); +}); + +test('空输入与非数组不炸', () => { + assert.deepEqual(selectCatchup([], new Set()), []); + assert.deepEqual(selectCatchup(undefined, new Set()), []); + assert.deepEqual(selectCatchup(null, new Set()), []); +}); + +test('没有 mail_id 的行跳过', () => { + const got = selectCatchup([mail({ mail_id: '' }), mail({ mail_id: 'ok' })], new Set()); + assert.deepEqual(got.map(e => e.mail_id), ['ok']); +}); + +test('seen 传 undefined 时不去重也不报错', () => { + const got = selectCatchup([mail({ mail_id: 'x' })], undefined); + assert.deepEqual(got.map(e => e.mail_id), ['x']); +});