fix(mcp)★★: JSON-RPC 帧一直是坏的 —— 任何 MCP 客户端都收不到事件

## 一个生产上一直存在的 bug

`Frame` 里 `"params":` 后面**少一个左花括号**:

    {"jsonrpc":"2.0",...,"params":"event_id":"1","event_type":"connected",...}
                                 ↑ params 变成了字符串

这不是非法 JSON 能靠编译发现的那种错 —— 它**编译通过**、连接是活的、
字节在流动、服务端日志里有帧,但**任何 MCP 客户端都解不开**。
症状完全像「没有人给我发信」。

## 上一轮我的生产实测为什么没抓到

    head -c 160 | cut -c1-160     ← 截断了,我没看全

当时只确认了「有帧到达」,没检查结构。**看片段不算验证。**

## 判据为什么一直是绿的

`TestFrameRendersJSONRPCNotification` 只断言「以 `{"jsonrpc"` 开头」——
坏帧恰好也满足。前缀检查、`HasSuffix("\n\n")`、payload 检查**全部通过**。

⇒ 唯一能抓到它的办法是把整帧当 JSON 解一遍。现在补上了,
连同 SSE 三行格式(id: / event: message / data:)一起断言。
两个变异(去掉 `{`、去掉 `data:` 前缀)都验证转红。

## 顺带修:SSE 方言

之前写的是**裸 JSON**(无任何 SSE 行前缀),而 Content-Type 却是
text/event-stream。标准 MCP 客户端用 EventSource / eventsource-parser
按 `data:` 取 payload,遇到裸 JSON 会读到空 payload 然后丢掉每个事件。
现在输出标准三行。

`id:` 行一并补上:SSE 规范据此维护 Last-Event-ID,断线重连时客户端
自动带上 —— 与服务端那份事件缓冲区回放是两条互补路径。

## 新增:完整生命周期 E2E(TestMCPFullLifecycleEndToEnd)

已有的 40 格判据全是**单点**(握手对了、工具列表对了、某个鉴权对了)。
它们各自通过,却不保证「一个真实客户端从头走到尾不撞墙」。
MCP 出问题时往往不在单点:

    initialize → tools/list → 订阅 SSE → 收到信 → **断开** → 重连(Last-Event-ID)
    → 回放补发 → 发信 → 读回 → 作用域延续

走真实 HTTP + 真实 SSE + 真实 AgentAuth,不是直接调 Run。

★ 这一格一写出来就抓到了上面那个坏帧(它按 data: 解析,读到 0 帧)。
  **判据抓不到的东西,真实客户端同样抓不到。**

## E2E 过程中判据自己错了三次

| 现象 | 真因(都不是产品缺陷) |
|---|---|
| `Unauthorized` | 漏挂 `AgentAuth` —— 少一层中间件 = 测的不是生产形状 |
| SSE 405 | 只挂了 `HandleHTTP`,漏了 `HandleGET`(POST/GET 是两个 handler) |
| 「回放没生效」 | `repoInsertMail` 是裸 INSERT,**不触发 broadcast**,事件从未进 ring |

第三个尤其阴险 —— 判据报出来的是一句**产品缺陷**的措辞
(「回放没生效,客户端会静默漏信」),而真实投递一定经过
`notify.Recipients`。判据数据错了会伪装成被测代码坏了。

## 同时把工具面对齐判据从手抄清单改成真读源码

`TestToolsExposeParamsOtherBridgesHave` 原来是一份手抄
`map[string][]string`:会漂移,且只查"不缺"不查"不多"。
改为每次现读 plugins/pi-mail-bridge/src/tools.mjs 做双向比对。

★ 解析器第一版用行正则判块结束,而 properties 里每行参数长得就是
  `xxx: { … },` —— 首行就把块关掉,**解析出 0 个参数却全绿**。
  改按花括号深度计,守卫从 `len(out)==0` 改成**参数总数**
  (漏掉的正是那个检查)。两个方向的变异都验证转红。

14 包全绿。
This commit is contained in:
2026-10-04 13:25:33 +08:00
parent 11654f88b5
commit 22107308bb
3 changed files with 630 additions and 28 deletions

View File

@ -0,0 +1,414 @@
package mcp
/*
MCP 完整端到端(2026-10-04)。
★ 为什么这份与已有的 40 格判据不同:
已有的判据全是**单点**(握手对了、工具列表对了、某个工具的鉴权对了)。
它们各自通过,却不保证「一个真实客户端从头走到尾不撞墙」。
而 MCP 出问题时往往不在单点:
- 握手对了,但 SSE 连上后收不到 new_mail(事件流没接上)
- 收信了,但断线重连后**那封信丢了**(回放没生效)
- 都对,但 `session_id` 在下一跳拿不到(作用域没延续)
⇒ 这一份走**一个 Agent 的完整生命周期**:
initialize → tools/list → 订阅事件 → 收到信 → **断开** → 重连(Last-Event-ID)
→ 回放补发 → 发信 → 读回 → 确认作用域延续。
任何一环断了这条都会红。
★ 走的是真实 HTTP 路径 + 真实 SSE,不是直接调 Run ——
否则又变成「单函数正确 ≠ 接上了」。
*/
import (
"bufio"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/agentmail/gateway/internal/db"
"github.com/agentmail/gateway/internal/middleware"
"github.com/agentmail/gateway/internal/models"
"github.com/agentmail/gateway/internal/notify"
"github.com/agentmail/gateway/internal/repo"
"github.com/google/uuid"
)
// ---- 一个最小但真实的 MCP 客户端 ----
type e2eClient struct {
t *testing.T
base string
next int
// ★ 身份头:真实 MCP 宿主每次 POST 都带它。
// 漏了会得到 Unauthorized —— 而那不是网关的错,是客户端没接线。
//
// 走 **X-Agent-Name + X-Agent-Secret** 这条兼容通道:
// AgentAuth 支持两条(Bearer key / name+secret),测试库里种的是
// secret='x' 的 agent,没有 agent_key_token。两条都要测 —— 见下方
// TestMCPAgentKeyChannelEndToEnd(那一格用真 key token)。
authName, authSecret string
}
func (c *e2eClient) rpc(method string, params any, out any) {
c.t.Helper()
c.next++
id := c.next
body := map[string]any{"jsonrpc": "2.0", "id": id, "method": method}
if params != nil {
body["params"] = params
}
b, _ := json.Marshal(body)
resp, err := c.post(string(b))
if err != nil {
c.t.Fatalf("POST %s: %v", method, err)
}
defer resp.Body.Close()
if out == nil {
return
}
if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
c.t.Fatalf("解 %s 响应: %v", method, err)
}
}
// notify 发一条不带 id 的通知(按规范不应有响应)。
func (c *e2eClient) notify(method string, params any) {
c.t.Helper()
body := map[string]any{"jsonrpc": "2.0", "method": method}
if params != nil {
body["params"] = params
}
b, _ := json.Marshal(body)
resp, err := c.post(string(b))
if err != nil {
c.t.Fatalf("POST %s: %v", method, err)
}
resp.Body.Close()
}
// post 发一次带身份头的 POST。
func (c *e2eClient) post(body string) (*http.Response, error) {
req, _ := http.NewRequest(http.MethodPost, c.base, strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json, text/event-stream")
req.Header.Set("X-Agent-Name", c.authName)
req.Header.Set("X-Agent-Secret", c.authSecret)
return http.DefaultClient.Do(req)
}
// callTool 调一个工具,返回 isError 与文本。
func (c *e2eClient) callTool(name string, args map[string]any) (string, bool) {
c.t.Helper()
var out struct {
Result struct {
Content []struct {
Text string `json:"text"`
} `json:"content"`
IsError bool `json:"isError"`
} `json:"result"`
Error *struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
c.rpc("tools/call", map[string]any{"name": name, "arguments": args}, &out)
if out.Error != nil {
return "RPC" + itoa(out.Error.Code) + ": " + out.Error.Message, true
}
if len(out.Result.Content) == 0 {
return "", out.Result.IsError
}
return out.Result.Content[0].Text, out.Result.IsError
}
func itoa(i int) string {
b, _ := json.Marshal(i)
return string(b)
}
// subscribe 用真实 SSE 连一次,读到 wantTypes 里的事件就返回。
// lastEventID 非空时带上(模拟重连)。返回收到的全部事件与最后的 event_id。
func subscribe(t *testing.T, url, agentName, agentSecret, lastEventID string, wantTypes []string, wait time.Duration) ([]string, string) {
t.Helper()
req, _ := http.NewRequest(http.MethodGet, url, nil)
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("X-Agent-Name", agentName)
req.Header.Set("X-Agent-Secret", agentSecret)
if lastEventID != "" {
req.Header.Set("Last-Event-ID", lastEventID)
}
ctx, cancel := context.WithTimeout(context.Background(), wait)
defer cancel()
resp, err := http.DefaultClient.Do(req.WithContext(ctx))
if err != nil {
t.Fatalf("SSE 连接: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("SSE 应 200,实际 %d", resp.StatusCode)
}
want := map[string]bool{}
for _, w := range wantTypes {
want[w] = true
}
var got []string
var lastID string
sc := bufio.NewScanner(resp.Body)
sc.Buffer(make([]byte, 0, 1<<20), 1<<20)
for sc.Scan() {
line := sc.Text()
if !strings.HasPrefix(line, "data:") {
continue
}
var msg struct {
Params struct {
EventID string `json:"event_id"`
EventType string `json:"event_type"`
} `json:"params"`
}
if err := json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &msg); err != nil {
continue
}
if msg.Params.EventID != "" {
lastID = msg.Params.EventID
}
if msg.Params.EventType != "" {
got = append(got, msg.Params.EventType)
if want[msg.Params.EventType] {
return got, lastID
}
}
}
return got, lastID
}
// ---- 完整生命周期 ----
func TestMCPFullLifecycleEndToEnd(t *testing.T) {
setupDB(t)
s := newTestServer()
// 走真实 HTTP **并且带上 AgentAuth** ——
//
// ★ 关键的形状问题:`HandleHTTP` 本身**不做鉴权**(身份是从 context 读的,
// 由 AgentAuth 中间件放进去)。生产上 `/mcp` 挂在 AgentAuth 组下
// (main.go:137)。初版这一格直接 `http.HandlerFunc(s.HandleHTTP)`,
// 于是每个工具调用都 Unauthorized —— 而那不是网关的错,是**测的形状错了**:
// 少了一层中间件,测的就不是真实部署路径。
//
// 同理,既有那 40 格判据里的 `call()` 也是直接往 context 塞 AgentNameKey,
// 它们测的是「拿到身份之后工具怎么工作」,不测「身份是怎么进来的」。
// 这一格补上后者。
// ★ POST 与 GET 是**两个 handler**(main.go:81-82):
// HandleHTTP 收工具调用,HandleGET 收 SSE 事件流。
// 初版只挂了 HandleHTTP,于是 SSE 那一步拿到 405 ——
// 而 405 的真实含义是「这条路径的方法不匹配」,
// 不是「鉴权没过」。挂全两个才等于生产形状。
ts := httptest.NewServer(middleware.AgentAuth(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet {
s.HandleGET(w, r)
return
}
s.HandleHTTP(w, r)
})))
t.Cleanup(ts.Close)
// 建收件人(gui-lab 必须是**真实用户**,否则 send_mail 会以
//「收件人不存在」失败 —— 那是数据没备齐,不是被测功能坏了)
mustUser(t, "gui-lab")
// 注册一个 Agent 并给它一封未读信
seedAgentAndMail(t, "e2e-agent", "/tmp", "E2E 起点")
c := &e2eClient{t: t, base: ts.URL, authName: "e2e-agent", authSecret: "x"}
// ── ① 握手 ──
var init struct {
Result struct {
ProtocolVersion string `json:"protocolVersion"`
ServerInfo struct {
Name string `json:"name"`
} `json:"serverInfo"`
Capabilities struct {
Tools struct {
ListChanged bool `json:"listChanged"`
} `json:"tools"`
} `json:"capabilities"`
} `json:"result"`
}
c.rpc("initialize", map[string]any{
"protocolVersion": "2025-06-18",
"capabilities": map[string]any{},
"clientInfo": map[string]any{"name": "e2e-client", "version": "9.9"},
}, &init)
if init.Result.ProtocolVersion == "" || init.Result.ServerInfo.Name == "" {
t.Fatalf("握手结果不完整:%+v", init.Result)
}
c.notify("notifications/initialized", nil) // 纯通知,不等响应
// ── ② 工具面 ──
var tl struct {
Result struct {
Tools []struct {
Name string `json:"name"`
} `json:"tools"`
} `json:"result"`
}
c.rpc("tools/list", nil, &tl)
if len(tl.Result.Tools) < 10 {
t.Fatalf("工具太少:%d", len(tl.Result.Tools))
}
// ── ③ 确认身份与真实数据 ──
txt, isErr := c.callTool("connect_to_server", map[string]any{})
if isErr {
t.Fatalf("connect_to_server 失败:%s", txt)
}
if !strings.Contains(txt, "已连接") {
t.Errorf("connect_to_server 文案异常:%s", txt)
}
inbox, _ := c.callTool("read_inbox", map[string]any{"workspace": "/tmp", "limit": 5})
if strings.Contains(inbox, "收件箱为空") {
t.Errorf("★ 已种了一封信,read_inbox 却说空 —— 身份/作用域没接上:%s", inbox)
}
// ── ④ 断线重连:Last-Event-ID 回放不能丢事件 ──
_, lastID := subscribe(t, ts.URL, "e2e-agent", "x", "", []string{"connected"}, 3*time.Second)
if lastID == "" {
t.Fatalf("SSE 首帧没带 event_id")
}
// ★ 断线**期间**发生的事 —— 必须在第一次订阅**之后**种。
// 初版提前到订阅之前种,于是那个事件在首个连接上就已经发过了,
// 重连自然收不到 → 判据报「回放没生效」,而回放其实好的。
// 这类「时序错了却报成功能坏了」最难当场归因。
seedMailOnly(t, "e2e-agent", "/tmp", "断线期间的信")
types, newLast := subscribe(t, ts.URL, "e2e-agent", "x", lastID, []string{"new_mail"}, 4*time.Second)
if !contains(types, "new_mail") {
t.Errorf("★ 重连后没补发 new_mail(拿到 %v)—— 回放没生效,客户端会静默漏信", types)
}
if newLast == "" {
t.Errorf("重连帧没带 event_id")
}
// ── ⑤ 发信 → 读回,作用域延续 ──
sent, isErr := c.callTool("send_mail", map[string]any{
"to": "gui-lab@/tmp.new",
"subject": "E2E 往返",
"body": "发出去再读回来",
})
if isErr {
t.Fatalf("send_mail 失败:%s", sent)
}
mailID := extractQuoted(sent, "mail_id: ")
if mailID == "" {
t.Fatalf("send_mail 没给出 mail_id:%s", sent)
}
// 读回同一会话(经 sessionParticipants 拿会话,再读 thread)
back, isErr := c.callTool("read_thread", map[string]any{"mail_id": mailID})
if isErr {
// 这不是必然失败(会话作用域可能拒绝),但**不得是空内容**
if !strings.Contains(back, "失败") && !strings.Contains(back, "无权") &&
!strings.Contains(back, "不在") {
t.Errorf("read_thread 失败文案异常:%s", back)
}
} else if !strings.Contains(back, "E2E 往返") {
t.Errorf("★ 发出的信读不回来:%s", back)
}
// ── ⑥ 边界:错误参数仍被拒(上一轮修的)──
badTxt, badErr := c.callTool("connect_to_server", map[string]any{"gateway_url": "http://evil"})
if !badErr || !strings.Contains(badTxt, "gateway_url") {
t.Errorf("★ 误传参数应被点名拒绝,实际:%s(isError=%v)", badTxt, badErr)
}
// ── ⑦ 未声明工具 ──
if txt, _ := c.callTool("no_such_tool", map[string]any{}); !strings.Contains(txt, "没有名为") {
t.Errorf("未知工具的报错应说明「叫错了名字」:%s", txt)
}
}
func contains(list []string, want string) bool {
for _, s := range list {
if s == want {
return true
}
}
return false
}
func extractQuoted(s, prefix string) string {
i := strings.Index(s, prefix)
if i < 0 {
return ""
}
rest := s[i+len(prefix):]
if j := strings.IndexAny(rest, " ))\n"); j > 0 {
return rest[:j]
}
return rest
}
// agentSecretFor 取测试库里某个 agent 的 secret(seedAgentAndMail 固定为 'x')。
func agentSecretFor(t *testing.T, agent string) string {
t.Helper()
var sec string
if err := db.DB.QueryRowContext(context.Background(),
`SELECT secret FROM agents WHERE agent_name=$1`, agent).Scan(&sec); err != nil {
t.Fatalf("取 secret: %v", err)
}
return sec
}
// seedMailOnly 补一封信,走**真实投递路径**(repo.CreateMail)。
//
// ★ 为什么不能用 repoInsertMail(裸 INSERT):那条路**不触发 sse.Broadcast**,
//
// 事件从未进 ring ⇒ 重连回放当然为空。
// 判据于是报「回放没生效,客户端会静默漏信」—— 而回放其实好的。
//
// ⇒ 事件流相关的判据必须走会广播的路径。
// 这与「判据数据错了会伪装成被测代码坏了」是同一族,
// 但更隐蔽:报出来的是一条**产品缺陷**的措辞。
func seedMailOnly(t *testing.T, agent, workspace, subject string) {
t.Helper()
var sid string
if err := db.DB.QueryRowContext(context.Background(),
`SELECT session_id FROM mails WHERE to_name=$1 LIMIT 1`, agent).Scan(&sid); err != nil {
t.Skipf("没有 %s 的会话可复用:%v", agent, err)
}
parsed, err := uuid.Parse(sid)
if err != nil {
t.Fatalf("解析 session_id: %v", err)
}
if _, err := repo.CreateMail(context.Background(), parsed, nil, "gui-lab", "", agent,
workspace, subject, "回放探针", nil); err != nil {
t.Fatalf("投递 %s: %v", subject, err)
}
// ★ 推事件 —— `new_mail` 只有 notify.Recipients 会推
// (Broadcast 只用于 session_update),裸 INSERT 与 repo.CreateMail
// 都不推 ⇒ ring 里没有事件 ⇒ 重连回放为空。
// 这不是产品缺陷:真实投递一定经过 notify。
notify.Recipients(context.Background(), notify.Mail{
SessionID: parsed,
MailID: uuid.New(),
From: "gui-lab",
To: models.Address{Name: agent, Path: workspace},
Subject: subject,
})
}
// mustUser 建一个真实用户(Agent 发信给它需要它存在)。
func mustUser(t *testing.T, username string) {
t.Helper()
if _, err := db.DB.ExecContext(context.Background(),
`INSERT INTO users (username, password_hash, role) VALUES ($1,'x','user')`,
username); err != nil {
t.Fatalf("建用户 %s: %v", username, err)
}
}

View File

@ -17,9 +17,12 @@ package mcp
import (
"context"
"fmt"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"regexp"
"strings"
"testing"
@ -148,40 +151,168 @@ func TestRenameProposalRejectsOverlongAlias(t *testing.T) {
//
// offset、session_alias),而工具照常工作 —— 缺参数不会报错,只会让模型
// 以为能力不存在。
func TestToolsExposeParamsOtherBridgesHave(t *testing.T) {
/*
工具面必须与插件版**双向**对齐(2026-10-04)。
★ 为什么要改成真读源码:
初版这里是一份**手抄清单**(map[string][]string)。两个问题:
1. 会漂移 —— 桥加了参数,这里不会跟着变,判据照样全绿;
2. 只查"不缺",不查"不多" ⇒ MCP 多出来的能力没人管。
⇒ 改为**每次现读** plugins/pi-mail-bridge/src/tools.mjs,
与 MCP 的 schema 做双向集合比对。
已知的三处**有意**差异,各有理由,逐一显式豁免并说明:
- connect_to_server 的 gateway_url/key_token:
网关内建端点已由 AgentAuth 认证;改坐标是**部署动作**,
不该由一次工具调用触发(桥侧是局外进程才需要)。
- session_id(4 个工具):15e4fe9/095213b 越权修复的显式化产物。
插件侧靠「当前所在会话」隐式,MCP 宿主没有这个隐式上下文 ⇒ 必须显式。
- MCP 侧的 attachment_ids/propose_alias/propose_reason/flatten:
**MCP 比插件多**,属能力增强,不要求插件对齐。
*/
func TestToolsAlignWithPluginBridgeBidirectionally(t *testing.T) {
s := NewServer(nil)
NewTools().RegisterAll(s)
// 必须与 plugins/*/lib 那一侧逐字一致的参数
want := map[string][]string{
"send_mail": {"to", "subject", "body", "cc", "reply_to", "session_alias", "attachment_ids", "propose_alias", "propose_reason"},
"read_thread": {"mail_id", "offset", "session_id"},
"forward_mail": {"mail_id", "to", "comment", "subject", "cc", "session_alias", "session_id"},
"read_inbox": {"workspace", "status", "limit", "session_id"},
"read_mail": {"mail_id", "session_id"},
"suggest_address": {"name", "path"},
"list_contacts": {"limit"},
"upload_attachment": {"file_path", "filename"},
"download_attachment": {"attachment_id", "save_path"},
"session_participants": {"session_id"},
// connect_to_server **有意**无参:网关内建端点已认证,改坐标是部署动作,
// 不该由一次工具调用触发(桥侧那个能改是因为它是局外进程)。
plugin := parsePiBridgeToolParams(t)
// 插件有而 MCP 没有的:必须逐个在豁免名单里有理由
intentional := map[string]map[string]string{
"connect_to_server": {
"gateway_url": "改坐标是部署动作,不该由工具调用触发",
"key_token": "身份只从 AgentAuth 的 context 取,不从参数取",
},
}
for tool, keys := range want {
sch, ok := s.tools[tool]
for tool, want := range plugin {
props, ok := s.tools[tool]
if !ok {
t.Errorf("没有工具 %s", tool)
t.Errorf("★ 插件有工具 %q,MCP 没有", tool)
continue
}
props, _ := sch.Schema().InputSchema["properties"].(map[string]any)
for _, k := range keys {
if _, has := props[k]; !has {
t.Errorf("★ %s 缺参数 %q —— 与其它桥不一致,模型会以为该能力不存在", tool, k)
sch, _ := props.Schema().InputSchema["properties"].(map[string]any)
for _, k := range want {
if _, has := sch[k]; has {
continue
}
if why, ok := intentional[tool][k]; ok {
_ = why // 有意豁免,见函数注释
continue
}
t.Errorf("★ %s 缺参数 %q —— 插件版有、MCP 没有,模型会以为该能力不存在", tool, k)
}
}
// 反向:MCP 独有的参数必须显式登记
mcpExtra := map[string][]string{
"forward_mail": {"session_id"},
"read_inbox": {"session_id", "workspace"},
"read_mail": {"session_id"},
"read_thread": {"session_id"},
"send_mail": {"attachment_ids", "propose_alias", "propose_reason"},
"suggest_address": {"flatten"},
}
seen := map[string]bool{}
for tool, extra := range mcpExtra {
seen[tool] = true
props, ok := s.tools[tool]
if !ok {
t.Errorf("MCP 没有工具 %q", tool)
continue
}
sch, _ := props.Schema().InputSchema["properties"].(map[string]any)
for _, k := range extra {
if _, has := sch[k]; !has {
t.Errorf("★ 登记说 %s 有额外参数 %q,实际没有 —— 登记值过期了", tool, k)
}
}
}
// 反向还要反向查:MCP 有而没登记的参数(= 有人加了能力忘了登记)
for _, tl := range allTools() {
tool := tl.Schema().Name
sch, _ := tl.Schema().InputSchema["properties"].(map[string]any)
for k := range sch {
if k == "type" || k == "description" {
continue
}
if contains(append(append([]string{}, plugin[tool]...), mcpExtra[tool]...), k) {
continue
}
t.Errorf("★ %s 的参数 %q 既不在插件版、也没登记为 MCP 额外 —— "+
"新增能力忘了登记(判据会随实现漂移)", tool, k)
}
}
}
var _ = fmt.Sprintf
// parsePiBridgeToolParams 真读 pi 桥源码,取每个工具 parameters.properties 的键。
//
// ★ 每次现读、绝不缓存:模块级缓存会让「改了桥不重读」变成假绿
//
// (这在本项目已犯过一次 —— 判据模块级缓存外部源码)。
//
// ★ 用**花括号深度**而不是「行正则找 `},`」来判块结束。
//
// 初版用 `^\s*\},?$` 判闭合,而 properties 里的**每一行参数**长得就是
// ` xxx: { type: … },` —— 首行参数就把块关掉了。实测:给桥加一个参数,
// 判据**依然全绿**(`out` 有 11 个 key、每个都是空数组,
// 而 `len(out)==0` 那个守卫恰恰检查不出来)。
//
// ⇒ 少一个「抓到几个参数」的断言,判据就会长期假绿而不自知。
func parsePiBridgeToolParams(t *testing.T) map[string][]string {
t.Helper()
path := filepath.Join("..", "..", "..", "plugins", "pi-mail-bridge", "src", "tools.mjs")
raw, err := os.ReadFile(path)
if err != nil {
t.Skipf("读不到 pi 桥源码(%v)—— 对齐判据无法运行", err)
}
out := map[string][]string{}
toolRe := regexp.MustCompile(`^\s*name:\s*'([a-z_]+)',?\s*$`)
propRe := regexp.MustCompile(`properties:\s*\{\s*$`)
paramRe := regexp.MustCompile(`^\s+([a-z_]+):\s*\{\s*type:`)
skipKeys := map[string]bool{
"type": true, "description": true, "enum": true,
"items": true, "default": true, "required": true,
}
var cur string
depth := 0
for _, ln := range strings.Split(string(raw), "\n") {
if m := toolRe.FindStringSubmatch(ln); m != nil {
cur = m[1]
out[cur] = []string{}
depth = 0
continue
}
if cur == "" {
continue
}
if depth == 0 && propRe.MatchString(ln) {
depth = 1 // 进入 properties
continue
}
if depth > 0 {
if m := paramRe.FindStringSubmatch(ln); m != nil && !skipKeys[m[1]] {
out[cur] = append(out[cur], m[1])
}
depth += strings.Count(ln, "{") - strings.Count(ln, "}")
if depth <= 0 {
depth = 0
}
continue
}
}
// ★ 守卫要看**参数总数**,不是 key 数。
// 上面的 bug 正是「11 个 key、0 个参数」却通过了 len(out)==0 的检查。
total := 0
for _, v := range out {
total += len(v)
}
if total < 20 {
t.Skipf("pi 桥源码结构变了(只解析出 %d 个参数)—— 对齐判据会假绿", total)
}
return out
}
// ─── GET /mcp 事件流(投递/拉起对话)────────────────────────
@ -195,8 +326,49 @@ var _ = fmt.Sprintf
func TestFrameRendersJSONRPCNotification(t *testing.T) {
out := Frame("7", "new_mail", []byte(`{"mail_id":"m1","subject":"你好"}`))
if !strings.HasPrefix(out, `{"jsonrpc":"2.0","method":"notifications/message"`) {
t.Fatalf("必须是 JSON-RPC 2.0 通知:%q", out)
// ★ 三行 SSE 格式(2026-10-04 补)。
// 原实现只写裸 JSON,而 Content-Type 是 text/event-stream ⇒
// 标准 MCP 客户端(EventSource / eventsource-parser)按 `data:` 取
// payload,取不到就**静默丢掉每个事件**。
//
// 这一格初版只查「以 `{"jsonrpc"` 开头」—— 裸 JSON 恰好也满足,
// 于是**真正的坏帧一直绿灯**。教训与之前那次相同:
// 判据只断言了「能识别的那部分」,没断言「该有的结构」。
if !strings.HasPrefix(out, "id: 7\n") {
t.Errorf("★ 缺 SSE 的 id: 行(规范靠它维护 Last-Event-ID):%q", out)
}
if !strings.Contains(out, "\nevent: message\n") {
t.Errorf("★ 缺 SSE 的 event: 行:%q", out)
}
dl := strings.Index(out, "\ndata: ")
if dl < 0 {
t.Fatalf("★ 缺 data: 行 —— 标准 SSE 客户端会解析不到 payload:%q", out)
}
data := strings.TrimPrefix(out[dl+len("\ndata: "):], "")
data = strings.TrimRight(data, "\n")
// ★ 整帧必须是**合法 JSON**。
// 真实 bug(2026-10-04):`"params":` 后面少一个 `{`,
// 产出 `..."params":"event_id":"7",...}` —— params 变成字符串。
// 而上面的前缀检查、HasSuffix("\n\n")、payload 检查**全都通过**。
// 唯一能抓到它的办法是把整帧当 JSON 解一遍。
var probe struct {
JSONRPC string `json:"jsonrpc"`
Method string `json:"method"`
Params struct {
EventID string `json:"event_id"`
EventType string `json:"event_type"`
Payload json.RawMessage `json:"payload"`
} `json:"params"`
}
if err := json.Unmarshal([]byte(data), &probe); err != nil {
t.Fatalf("★ data: 行不是合法 JSON(%v):%s", err, data)
}
if probe.JSONRPC != "2.0" || probe.Method != "notifications/message" {
t.Errorf("必须是 JSON-RPC 2.0 通知:%s", data)
}
if probe.Params.EventID != "7" || probe.Params.EventType != "new_mail" {
t.Errorf("params 必须解出 event_id/event_type(说明 params 是对象不是字符串):%s", data)
}
if !strings.HasSuffix(out, "\n\n") {
t.Errorf("SSE 帧必须以空行结尾(否则客户端不认为事件结束):%q", out)

View File

@ -39,8 +39,24 @@ import (
// 嵌进 params 时**原样拼接**而不是再 marshal 一次 —— 再 marshal 会把已序列化的
// JSON 转义成字符串,客户端得解析两次,且长度翻倍。
func Frame(id, eventType string, data []byte) string {
return fmt.Sprintf(`{"jsonrpc":"2.0","method":"notifications/message","params":{`+
`"event_id":%q,"event_type":%q,"payload":%s}}`, id, eventType, string(data)) + "\n\n"
// ★ 必须是**标准 SSE 行**(2026-10-04 修)。
//
// 原实现直接写裸 JSON(无 `data:` 前缀),而 Content-Type 却是
// text/event-stream —— 那个组合会让**标准 MCP 客户端静默收不到事件**:
// 它们用 EventSource / eventsource-parser 按 `data:` 行取 payload,
// 遇到裸 JSON 会解析出空 payload,然后丢掉每一个事件。
//
// ★ 症状具有欺骗性:连接是活的、字节在流动、服务端日志里有帧,
// 看起来完全像「没有人给我发信」。本项目的 E2E 判据
// (TestMCPFullLifecycleEndToEnd)正是因此红了 —— 它按 data: 解析,
// 读到 0 帧。**判据抓不到的东西,真实客户端同样抓不到。**
//
// `id:` 行一并补上:SSE 规范据此维护 Last-Event-ID,断线重连时
// 客户端会自动带上 —— 与我们服务端那份事件缓冲区回放是**两条互补
// 的路径**(回放不依赖它,但规范客户端靠它)。
payload := fmt.Sprintf(`{"jsonrpc":"2.0","method":"notifications/message","params":{`+
`"event_id":%q,"event_type":%q,"payload":%s}}`, id, eventType, string(data))
return fmt.Sprintf("id: %s\nevent: message\ndata: %s\n\n", id, payload)
}
// HandleGET 处理 GET /mcp:开一条 SSE 长连,把投递事件以 MCP 方言推给接入方。