diff --git a/server/internal/mcp/e2e_test.go b/server/internal/mcp/e2e_test.go new file mode 100644 index 0000000..e891860 --- /dev/null +++ b/server/internal/mcp/e2e_test.go @@ -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) + } +} diff --git a/server/internal/mcp/rename_proposal_test.go b/server/internal/mcp/rename_proposal_test.go index 14b814c..dbaeb8a 100644 --- a/server/internal/mcp/rename_proposal_test.go +++ b/server/internal/mcp/rename_proposal_test.go @@ -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) diff --git a/server/internal/mcp/stream.go b/server/internal/mcp/stream.go index 22550b4..f010864 100644 --- a/server/internal/mcp/stream.go +++ b/server/internal/mcp/stream.go @@ -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 方言推给接入方。