From 560c46276848b6c43a2d27c4c134f57a44386d74 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Fri, 2 Oct 2026 15:37:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20GET=20/api/v1/mcp=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20=E6=8A=95=E9=80=92=E4=BE=A7=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E6=B5=81=EF=BC=88=E8=AE=A9=E6=8E=A5=E5=85=A5=E6=96=B9=E8=A2=AB?= =?UTF-8?q?=E5=8A=A8=E6=94=B6=E4=BF=A1=EF=BC=8C=E4=B8=8D=E7=94=A8=E8=BD=AE?= =?UTF-8?q?=E8=AF=A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 这半边解决什么 工具面（POST）只解决「接入方**问**」。这一条解决「服务端**说**」： 邮件投递时把 new_mail / session_update 推给接入方，让它**拉起对话** —— 与各桥靠 /api/v1/events/stream 收信是同一件事，只是方言不同： 桥: id: 7\nevent: new_mail\ndata: {…}\n\n MCP: {"jsonrpc":"2.0","method":"notifications/message","params":{…}} ## 为什么复用 sse.Manager 而不是另起一套 Manager 里那些东西**都是踩过坑才对的**：writeMu 串行化（2026-09-28 -race 实测 http.ResponseWriter 并发写会把 JSON 劈成半截，800 帧只切出 459 个完整）、 Last-Event-ID 回放（宁可重复也不丢失）、心跳（反代按空闲 30-58s 掐连接）、 环形缓冲上限、断线清理。复制一份等于把那些坑再踩一遍， 而两边的修复从此各走各的。 代价是 `sse.Client` 多了一个可选 `Frame` 钩子： **nil = AgentMail 原格式，各桥与 WebUI 行为一字未变**（默认值即历史行为）。 ## ★ 回放是第三条写路径，漏了就只在断线时现形 `Send` / `SendWithID` / `replay` 是三条写 Res 的路径。原先**三条都把格式写死**， 只改前两条的话：MCP 客户端**平时**一切正常，只有带 `Last-Event-ID` 重连时 才会收到一批自己解不开的帧 —— 同一个连接上两种方言。 判据 `TestCustomFrameAppliesToReplayToo` 专门钉这条，并带反向对照 （nil 帧必须回落 AgentMail 格式）。 `Frame` 必须在**注册时**传入（`AddClientWithFrame`），不能事后设 —— 回放发生在「先写响应、再注册」的前半段，事后设只影响之后推来的事件。 原先 `AddClient` 保留为薄封装，各桥与 WebUI 调用点一字未改。 ## 判据（6 格） Frame 是 JSON-RPC 2.0 通知 + 帧完整性（单事件、\n\n 结尾） payload 原样嵌入（不是 JSON 字符串）—— 再 marshal 会让客户端解析两次 event_id / event_type 必带（前者是 Last-Event-ID 续传的依据） Accept 判定（含 q 值、大小写） 匿名 GET → 401（不能变成静默的匿名订阅） 缺 Accept → 406（接错的客户端会静默收不到东西） ## 顺带修：TestAdvanceRecurrenceLunar 的时区缺陷（★ 今天第三次假红） 全量测试红了，查下来是**我今天早些时候改判据时引入的**，与本次改动无关。 农历换算必须按**本地公历日**算（`AdvanceRecurrence` 里那句 `eventTime.In(time.Local)` 就是这条规则）。库里读回的 EventTime 是 **UTC** （DSN 用 `_timezone=UTC`），UTC 比本地晚 8 小时，跨零点时农历日差一天： start (Local) = 2026-10-04 农历日 24 after (UTC) = 2026-11-01 16:00 农历日 23 ← 断言没换算时区（错） after.In(Local) = 2026-11-02 00:00 农历日 24 ← 正确 服务端代码一直是对的，是判据没照做。失败信息里现在打印时区， 免得下次要重新推导一遍。变异验证：去掉 `.In(time.Local)` → 红 1 ✓ （这条判据是农历的第三次假红了：3459605「断言要求不存在的农历日」、 今天早些「起点写死日期 + advanceToFuture 跳过过期月份」、现在「没换算时区」—— 三次都是判据自己写错，代码三次都对。它依赖 Local 时区与「今天」， 天生脆弱，值得记着。） ## 验证 go test ./... 14 包全绿 go test ./internal/sse/ 含新判据绿 go test ./internal/mcp/ 6 格新判据 + 原 19 格全绿 --- server/cmd/server/main.go | 5 + server/internal/mcp/rename_proposal_test.go | 105 ++++++++++++++++++++ server/internal/mcp/server.go | 5 +- server/internal/mcp/stream.go | 102 +++++++++++++++++++ server/internal/repo/calendar_test.go | 22 +++- server/internal/sse/manager.go | 73 +++++++++++++- server/internal/sse/manager_test.go | 58 ++++++++++- 7 files changed, 357 insertions(+), 13 deletions(-) create mode 100644 server/internal/mcp/stream.go diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 1a3d233..b37cf09 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -79,6 +79,7 @@ func main() { mcpServer := mcp.NewServer(nil) mcp.NewTools().RegisterAll(mcpServer) mcpHandler := mcpServer.HandleHTTP + mcpStreamHandler := mcpServer.HandleGET log.Printf("MCP 端点就绪：POST /api/v1/mcp（%d 个工具）", mcpServer.ToolCount()) r := chi.NewRouter() @@ -132,7 +133,11 @@ func main() { // // 这样接入端只需填一个 URL（Streamable HTTP 的 POST），不用装插件、 // 不用起进程、不用配环境变量。 + // 工具面：POST（JSON-RPC 请求-响应） r.Post("/mcp", mcpHandler) + // 事件流：GET + SSE（服务端→客户端，投递时推 new_mail） + // 这半边让接入方**不需要自己轮询**就能收到信 —— 与各桥收信同一件事。 + r.Get("/mcp", mcpStreamHandler) r.Post("/agent/heartbeat", handler.HeartbeatAgent) r.Post("/mail/send", handler.SendMail) diff --git a/server/internal/mcp/rename_proposal_test.go b/server/internal/mcp/rename_proposal_test.go index 5033a8a..14b814c 100644 --- a/server/internal/mcp/rename_proposal_test.go +++ b/server/internal/mcp/rename_proposal_test.go @@ -16,9 +16,14 @@ package mcp */ import ( + "context" "fmt" + "net/http" + "net/http/httptest" "strings" "testing" + + "github.com/agentmail/gateway/internal/middleware" ) // 服务端的正则，在这里 import 而不是复制 —— 复制一份就等于放弃了「镜像」的意义。 @@ -177,3 +182,103 @@ func TestToolsExposeParamsOtherBridgesHave(t *testing.T) { } var _ = fmt.Sprintf + +// ─── GET /mcp 事件流（投递/拉起对话）──────────────────────── + +/* +★ 方言必须是 JSON-RPC 通知。 + +这一格存在的理由：Frame 决定了接入方能不能解析。写错了不会报错 —— +连接是活的、字节在流动，只是客户端**解不开**而默默丢弃每一个事件。 +那种失败看起来完全像「没有人给我发信」。 +*/ +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) + } + if !strings.HasSuffix(out, "\n\n") { + t.Errorf("SSE 帧必须以空行结尾（否则客户端不认为事件结束）：%q", out) + } + // payload 必须**原样嵌入**而不是被转义成字符串 —— + // 否则客户端要解析两次，长度还翻倍。 + if !strings.Contains(out, `"payload":{"mail_id":"m1","subject":"你好"}`) { + t.Errorf("payload 应原样嵌入（不是 JSON 字符串）：%q", out) + } + // 反向对照：转义版本（错误实现）不该出现 + if strings.Contains(out, `"payload":"{`) { + t.Errorf("★ payload 被转义成字符串了：%q", out) + } +} + +func TestFrameCarriesEventTypeAndID(t *testing.T) { + out := Frame("42", "session_update", []byte(`{}`)) + if !strings.Contains(out, `"event_id":"42"`) { + t.Errorf("必须带 event_id（客户端据此做 Last-Event-ID 续传）：%q", out) + } + if !strings.Contains(out, `"event_type":"session_update"`) { + t.Errorf("必须带 event_type（否则客户端分不清新邮件与会话更新）：%q", out) + } +} + +// 单行 + 空行分隔：SSE 的帧完整性靠它，多一个换行会把一条拆成两条。 +func TestFrameIsSingleEvent(t *testing.T) { + out := Frame("1", "new_mail", []byte(`{}`)) + if strings.Count(out, "\n\n") != 1 { + t.Errorf("一个事件只能有一个空行分隔：%q", out) + } + if strings.Contains(strings.TrimSuffix(out, "\n\n"), "\n\n") { + t.Errorf("事件体内不得出现空行：%q", out) + } +} + +// Accept 判定：必须真的要求 text/event-stream。 +func TestAcceptsEventStream(t *testing.T) { + cases := []struct { + accept string + want bool + }{ + {"text/event-stream", true}, + {"text/event-stream, application/json", true}, + {"text/event-stream;q=0.9", true}, + {"TEXT/EVENT-STREAM", true}, + {"application/json", false}, + {"", false}, + {"text/plain", false}, + } + for _, c := range cases { + r := httptest.NewRequest(http.MethodGet, "/api/v1/mcp", nil) + if c.accept != "" { + r.Header.Set("Accept", c.accept) + } + if got := acceptsEventStream(r); got != c.want { + t.Errorf("Accept=%q 期望 %v 收到 %v", c.accept, c.want, got) + } + } +} + +// 未认证的 GET 必须拒绝（不能变成一处静默的匿名订阅）。 +func TestGetRequiresAgentIdentity(t *testing.T) { + s := newTestServer() + rec := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/api/v1/mcp", nil) + r.Header.Set("Accept", "text/event-stream") + // 故意不注入身份 + s.HandleGET(rec, r) + if rec.Code != http.StatusUnauthorized { + t.Errorf("匿名订阅必须 401，收到 %d", rec.Code) + } +} + +// 不带 Accept 的 GET 必须 406（接错的客户端会静默收不到东西）。 +func TestGetRequiresEventStreamAccept(t *testing.T) { + s := newTestServer() + rec := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/api/v1/mcp", nil) + r = r.WithContext(context.WithValue(r.Context(), middleware.AgentNameKey, "probe")) + s.HandleGET(rec, r) + if rec.Code != http.StatusNotAcceptable { + t.Errorf("缺 Accept 应 406，收到 %d", rec.Code) + } +} diff --git a/server/internal/mcp/server.go b/server/internal/mcp/server.go index 80f9737..4df81a8 100644 --- a/server/internal/mcp/server.go +++ b/server/internal/mcp/server.go @@ -208,8 +208,9 @@ func (s *Server) ToolCount() int { return len(s.tools) } // 所以 MCP 不能成为绕过既有鉴权与收窄的后门。 func (s *Server) HandleHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { - w.Header().Set("Allow", "POST") - writeJSON(w, http.StatusMethodNotAllowed, failure(nil, rpcInvalidRequest, "MCP 端点只接受 POST")) + w.Header().Set("Allow", "GET, POST") + writeJSON(w, http.StatusMethodNotAllowed, failure(nil, rpcInvalidRequest, + "POST 是工具面；事件流用 GET /api/v1/mcp（需 Accept: text/event-stream）")) return } // 限制请求体：工具调用都是小 JSON，附件走独立的 /attachments 端点。 diff --git a/server/internal/mcp/stream.go b/server/internal/mcp/stream.go new file mode 100644 index 0000000..22550b4 --- /dev/null +++ b/server/internal/mcp/stream.go @@ -0,0 +1,102 @@ +package mcp + +// GET /mcp —— 服务端→客户端的事件流（Streamable HTTP 的 SSE 通道）。 +// +// # 这半边解决什么 +// +// 工具面（POST /mcp）只解决「接入方**问**」。这一条解决「服务端**说**」： +// 邮件投递时把 new_mail / session_update 推给接入方，让它**拉起对话** —— +// 与各桥靠 /api/v1/events/stream 收信是同一件事，只是方言不同： +// +// 桥: id: 7\nevent: new_mail\ndata: {…}\n\n +// MCP: {"jsonrpc":"2.0","method":"notifications/message","params":{…}} +// +// # 为什么复用 sse.Manager 而不是另起一套 +// +// Manager 里那些东西**都是踩过坑才对的**：writeMu 串行化（2026-09-28 -race +// 实测 http.ResponseWriter 并发写会把 JSON 劈成半截，800 帧只切出 459 个完整）、 +// Last-Event-ID 回放（宁可重复也不丢失）、心跳（反代按空闲 30-58s 掐连接）、 +// 环形缓冲上限、断线清理。复制一份等于把那些坑再踩一遍， +// 而两边的修复从此各走各的。 +// +// 代价是 sse.Client 多了一个 Frame 钩子（见 sse/manager.go）。 +// 默认 nil = AgentMail 原格式，各桥与 WebUI 行为一字未变。 + +import ( + "fmt" + "net/http" + "strings" + + "github.com/agentmail/gateway/internal/middleware" + "github.com/agentmail/gateway/internal/sse" +) + +// Frame 把一条 AgentMail 事件渲染成 MCP 方言（JSON-RPC 通知）。 +// +// 独立成函数是为了能被单测直接调用、逐字节断言。 +// +// data 已是 JSON 字节（SendWithID marshal 过；回放时本来就是原始字节）， +// 嵌进 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" +} + +// HandleGET 处理 GET /mcp：开一条 SSE 长连，把投递事件以 MCP 方言推给接入方。 +// +// 认证在**外层**（与 POST 同一条 AgentAuth 路由）：身份已在 context 里， +// 未带凭证的请求根本到不了这里。 +func (s *Server) HandleGET(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.Header().Set("Allow", "GET, POST") + http.Error(w, "MCP GET 通道只接受 GET（工具面走 POST）", http.StatusMethodNotAllowed) + return + } + + agent := middleware.GetAgentName(r) + if agent == "" { + // AgentAuth 已拦过；能到这里说明路由挂错了位置。 + // 报出来而不是默默接受 —— 匿名订阅会变成一处静默的越权读。 + http.Error(w, "MCP 事件流需要 Agent 凭证", http.StatusUnauthorized) + return + } + + // 必须声明接受 SSE：否则客户端拿到一堆自己解析不了的文本流。 + // 这条不是形式主义 —— 规范里 POST 与 GET 的语义不同，接错的客户端 + // 会静默地什么都不收到（连接开着，但内容它不是当 SSE 读的）。 + if !acceptsEventStream(r) { + http.Error(w, "MCP 事件流要求 Accept: text/event-stream", http.StatusNotAcceptable) + return + } + + // ★ 同一把 Manager、同一个缓冲区、同一套回放 —— 与各桥唯一的差别是方言。 + // + // Frame 在**注册时**传入（而不是注册后设）：回放发生在"先写响应、再注册" + // 的前半段，事后设会让重连那一批走默认格式 —— 同一个连接两种方言， + // 且只在真实断线时现形。见 sse.AddClientWithFrame 的注释。 + client := sse.Default.AddClientWithFrame(w, r, agent, "", Frame) + if client == nil { + http.Error(w, "SSE not supported", http.StatusInternalServerError) + return + } + + // 连接还活着就阻塞在这里；断开（客户端关、反代掐、服务关停）即返回。 + <-r.Context().Done() + sse.Default.RemoveClient(client.ID) +} + +// acceptsEventStream 判 Accept 头里有没有 text/event-stream。 +// +// 手写而不是用 mime.ParseMediaType + 循环：Accept 是**列表**且带 q 值， +// 完整解析要处理 `text/event-stream;q=0.9, application/json;q=0.5`。 +// 这里只关心「有没有声明」，不关心优先级 —— 多解析的那部分没有消费者。 +func acceptsEventStream(r *http.Request) bool { + for _, part := range strings.Split(r.Header.Get("Accept"), ",") { + media := strings.TrimSpace(strings.SplitN(part, ";", 2)[0]) + if strings.EqualFold(media, "text/event-stream") { + return true + } + } + return false +} diff --git a/server/internal/repo/calendar_test.go b/server/internal/repo/calendar_test.go index 502242b..ef59e37 100644 --- a/server/internal/repo/calendar_test.go +++ b/server/internal/repo/calendar_test.go @@ -663,9 +663,27 @@ func TestAdvanceRecurrenceLunar(t *testing.T) { t.Errorf("event_time 未前进：%v", after.EventTime) } // 农历日保持（起点是明天，那个日就是起点的农历日） + // ★ 农历换算必须按**本地公历日**算（服务端 `AdvanceRecurrence` 里那句 + // `eventTime.In(time.Local)` 就是这条规则）。 + // + // 2026-10-02 实测踩到：库里读回的 EventTime 是 **UTC** + // （SQLite DSN 用 _timezone=UTC 保证时间比较统一）。直接拿它喂 + // `lunar.FromSolar` 会按 UTC 的日期换算 —— 而 UTC 比 HKT 晚 8 小时， + // 本地 11-02 00:00 在 UTC 里还是 11-01，农历日因此差一天： + // + // start (Local) = 2026-10-04 农历日 24 + // after (UTC) = 2026-11-01 16:00 农历日 23 ← 直接换算得到这个（错） + // after.In(Local) = 2026-11-02 00:00 农历日 24 ← 正确答案 + // + // 也就是说：**断言本身没换算时区**，与代码无关。这条不算代码缺陷。 wantLunarDay := lunar.FromSolar(start).Day - if d := lunar.FromSolar(after.EventTime); d.Day != wantLunarDay { - t.Errorf("农历日变成 %d，期望 %d（%s）", d.Day, wantLunarDay, d.String()) + gotLunar := lunar.FromSolar(after.EventTime.In(time.Local)) + if gotLunar.Day != wantLunarDay { + t.Errorf("农历日变成 %d，期望 %d（%s）\n"+ + "注意：after.EventTime 是 %s 时区的 %s，必须先 .In(time.Local) 再换算 —— "+ + "UTC 晚 8 小时，跨零点时会差一天", + gotLunar.Day, wantLunarDay, gotLunar.String(), + after.EventTime.Location(), after.EventTime.Format("2006-01-02 15:04")) } // 公历间隔应在一个农历月内 gap := int(after.EventTime.Sub(start).Hours() / 24) diff --git a/server/internal/sse/manager.go b/server/internal/sse/manager.go index f00bb95..56f0c8e 100644 --- a/server/internal/sse/manager.go +++ b/server/internal/sse/manager.go @@ -52,13 +52,22 @@ func (r *eventRing) push(evt StoredEvent) { // afterID 为空时：缓冲区未满不回放（首次连接无历史）；满了也不回放 // （首次连接的 EventSource 不传 Last-Event-ID）。 // afterID 非空时：找到该 ID 的位置，从下一条开始回放。 -func (r *eventRing) replay(afterID string, flush http.Flusher, res http.ResponseWriter) { +// +// ★ frame 必传（2026-10-02）：回放是**第三条**写 Res 的路径，它原本也把格式 +// 写死成 AgentMail 的 SSE 形状。MCP 连接断线重连时走的就是这条路 —— +// 不传 frame 会让它收到一堆自己的客户端解不开的帧（同一个连接上两种方言）。 +// 这种错只在「MCP 客户端带 Last-Event-ID 重连」时现形，平时完全看不见。 +func (r *eventRing) replay(afterID string, flush http.Flusher, res http.ResponseWriter, + frame func(id, eventType string, data []byte) string) { r.mu.Lock() defer r.mu.Unlock() if afterID == "" { return // 首次连接，不回放 } + if frame == nil { + frame = defaultFrame + } start := -1 total := r.cap @@ -90,7 +99,7 @@ func (r *eventRing) replay(afterID string, flush http.Flusher, res http.Response if evt.ID == "" { continue } - fmt.Fprintf(res, "id: %s\nevent: %s\ndata: %s\n\n", evt.ID, evt.EventType, evt.Data) + fmt.Fprint(res, frame(evt.ID, evt.EventType, evt.Data)) } flush.Flush() } @@ -127,6 +136,38 @@ type Client struct { // SendToRecipient / Broadcast / replay），要保证"同一个 client 的所有写互斥"， // 责任只能落在 client 自己身上 —— 那是唯一能覆盖**全部**写者的位置。 writeMu sync.Mutex + + // Frame 可选：把一条事件渲染成**要写进流里的字节**。 + // + // 为什么需要它（2026-10-02，MCP 端点要把同一批事件说成别的协议）： + // 本管理器原本把帧格式**写死**成 AgentMail 的 SSE 形状。而 MCP + // （Streamable HTTP）要求服务端→客户端的方向用 **JSON-RPC 通知**经 SSE 下发 + // —— 同一根管子、两种方言。 + // + // 把格式抽成一个函数而不是复制一份 Manager：心跳、Last-Event-ID 回放、 + // 断线清理、`writeMu` 串行化、容量上限、事件环缓冲 —— 这些是**踩过坑才对的** + // （见 writeMu 的注释）。复制一份等于把那些坑再踩一遍， + // 而两边的修复会各走各的。 + // + // nil = 用默认的 AgentMail 格式（各桥与 WebUI 走这条，行为一字未变）。 + Frame func(id, eventType string, data []byte) string +} + +// defaultFrame 是 AgentMail 自己的 SSE 帧格式（带 id）。 +func defaultFrame(id, eventType string, data []byte) string { + return fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n\n", id, eventType, data) +} + +// fill 是**唯一**把事件写进 Res 的地方（心跳除外的两条推送路径共用）。 +// +// 抽出来的意义：让「默认格式」与「自定义格式」在代码上对称 —— 将来改默认格式时， +// 两边的差异一眼可见；而漏掉其中一条路径就会让某种连接收到半截方言。 +func (c *Client) fill(id, eventType string, data []byte) { + if c.Frame != nil { + fmt.Fprint(c.Res, c.Frame(id, eventType, data)) + return + } + fmt.Fprint(c.Res, defaultFrame(id, eventType, data)) } // Manager 管理所有 SSE 客户端连接 @@ -182,7 +223,24 @@ func (m *Manager) getOrCreateRing(key string) *eventRing { } // AddClient 注册一个新 SSE 客户端（agentName 与 userName 二者恰其一） +// AddClient 注册一个 SSE 客户端，用**默认** AgentMail 帧格式。 +// +// 各桥与 WebUI 走这条（行为与历史完全一致）。需要别的方言的调用方 +// （MCP 端点把同一批事件说成 JSON-RPC 通知）用 AddClientWithFrame。 func (m *Manager) AddClient(res http.ResponseWriter, r *http.Request, agentName, userName string) *Client { + return m.AddClientWithFrame(res, r, agentName, userName, nil) +} + +// AddClientWithFrame 注册一个 SSE 客户端，用调用方给的帧格式。 +// +// ★ frame 必须在**构造时**传入，不能注册后再设。 +// +// 原因：Last-Event-ID 回放发生在“先写响应、再注册”的**前半段**（见下面那段 +// 关于持 writeMu 的注释）。事后挂 Frame 只能影响之后推来的事件， +// 而重连回放那一批仍会走默认格式 —— 同一个连接上出现两种方言， +// 且**只在真实断线重连时现形**。frame=nil 表示默认格式。 +func (m *Manager) AddClientWithFrame(res http.ResponseWriter, r *http.Request, + agentName, userName string, frame func(id, eventType string, data []byte) string) *Client { flusher, ok := res.(http.Flusher) if !ok { return nil @@ -196,6 +254,7 @@ func (m *Manager) AddClient(res http.ResponseWriter, r *http.Request, agentName, Res: res, Flusher: flusher, done: make(chan struct{}), + Frame: frame, } // 设置 SSE 响应头 @@ -231,7 +290,7 @@ func (m *Manager) AddClient(res http.ResponseWriter, r *http.Request, agentName, key := m.bufferKey(userName, agentName) if ring := m.getOrCreateRing(key); ring != nil && lastID != "" { client.writeMu.Lock() - ring.replay(lastID, flusher, res) + ring.replay(lastID, flusher, res, client.Frame) client.writeMu.Unlock() } @@ -383,7 +442,11 @@ func (c *Client) Send(eventType string, data interface{}) { c.writeMu.Lock() defer c.writeMu.Unlock() - fmt.Fprintf(c.Res, "event: %s\ndata: %s\n\n", eventType, jsonData) + if c.Frame != nil { + fmt.Fprint(c.Res, c.Frame("", eventType, jsonData)) + } else { + fmt.Fprintf(c.Res, "event: %s\ndata: %s\n\n", eventType, jsonData) + } c.Flusher.Flush() } @@ -398,7 +461,7 @@ func (c *Client) SendWithID(id, eventType string, data interface{}) { c.writeMu.Lock() defer c.writeMu.Unlock() - fmt.Fprintf(c.Res, "id: %s\nevent: %s\ndata: %s\n\n", id, eventType, jsonData) + c.fill(id, eventType, jsonData) c.Flusher.Flush() } diff --git a/server/internal/sse/manager_test.go b/server/internal/sse/manager_test.go index 973d0ac..cfa698d 100644 --- a/server/internal/sse/manager_test.go +++ b/server/internal/sse/manager_test.go @@ -2,6 +2,7 @@ package sse import ( "net/http/httptest" + "strconv" "strings" "testing" "time" @@ -22,14 +23,14 @@ func TestEventRingPushReplay(t *testing.T) { // 空 afterID → 首次连接，不回放（缓冲区未满） rec := httptest.NewRecorder() - ring.replay("", rec, rec) + ring.replay("", rec, rec, nil) if rec.Body.Len() > 0 { t.Error("首次连接不应回放事件，实际:", rec.Body.String()) } // 有 afterID → 从下一条开始回放 rec2 := httptest.NewRecorder() - ring.replay("1", rec2, rec2) + ring.replay("1", rec2, rec2, nil) body := rec2.Body.String() if !strings.Contains(body, "id: 2") { t.Error("afterID=1 应该回放 id:2，实际:", body) @@ -43,7 +44,7 @@ func TestEventRingPushReplay(t *testing.T) { // 不存在的 afterID → 从头回放全部 rec3 := httptest.NewRecorder() - ring.replay("999", rec3, rec3) + ring.replay("999", rec3, rec3, nil) body3 := rec3.Body.String() if !strings.Contains(body3, "id: 1") { t.Error("不存在的 afterID 应从头回放，实际:", body3) @@ -69,7 +70,7 @@ func TestEventRingOverflow(t *testing.T) { // afterID=2 已被覆盖 → 找不到位置，从头回放全部 rec := httptest.NewRecorder() - ring.replay("2", rec, rec) + ring.replay("2", rec, rec, nil) body := rec.Body.String() if !strings.Contains(body, "id: 3") || !strings.Contains(body, "id: 5") { t.Error("缓冲区溢出后应能回放可用范围，实际:", body) @@ -111,3 +112,52 @@ func TestHeartbeatIntervalIsWellUnderProxyIdleTimeout(t *testing.T) { "心跳必须明显小于该超时（当前上限取 20s）", heartbeatInterval) } } + +/* +★ 2026-10-02：Frame 钩子必须在**两条**写路径上都生效。 + +回放是第三条写 Res 的路径（另两条是 Send / SendWithID）。它原先也把格式写死成 +AgentMail 的 SSE 形状 —— MCP 端点要靠它把同一批事件说成 JSON-RPC 方言。 + +这一格存在的理由：只改 SendWithID 而漏改 replay，**平时完全看不出来** —— +只有在「MCP 客户端带着 Last-Event-ID 重连」时才会收到一堆自己解不开的帧。 +那种错要等真实断线才现形，所以只能靠判据钉。 +*/ +func TestCustomFrameAppliesToReplayToo(t *testing.T) { + ring := newEventRing(8) + for i := 1; i <= 3; i++ { + ring.push(StoredEvent{ + ID: strconv.Itoa(i), + EventType: "new_mail", + Data: []byte(`{"n":1}`), + Timestamp: time.Now(), + }) + } + + // 自定义方言：一眼可辨，且**不带** AgentMail 的 `event:` 行 + jsonrpcFrame := func(id, eventType string, data []byte) string { + return `{"jsonrpc":"2.0","method":"notifications/message","params":{` + + `"id":"` + id + `","type":"` + eventType + `","payload":` + string(data) + `}}` + "\n\n" + } + + rec := httptest.NewRecorder() + ring.replay("1", rec, rec, jsonrpcFrame) + body := rec.Body.String() + + if !strings.Contains(body, `"jsonrpc":"2.0"`) { + t.Errorf("★ 回放必须用自定义 frame（否则重连的 MCP 客户端收到异种方言）：%q", body) + } + if strings.Contains(body, "event: new_mail") { + t.Errorf("★ 回放仍写成 AgentMail 的 SSE 形状 —— Frame 钩子漏接在 replay 路径上：%q", body) + } + if !strings.Contains(body, `"id":"2"`) { + t.Errorf("回放内容不对（应从 afterID=1 之后开始）：%q", body) + } + + // 反向对照：nil ⇒ 默认格式一字未变（各桥与 WebUI 走这条） + rec2 := httptest.NewRecorder() + ring.replay("1", rec2, rec2, nil) + if !strings.Contains(rec2.Body.String(), "event: new_mail") { + t.Errorf("nil frame 必须回落默认 AgentMail 格式：%q", rec2.Body.String()) + } +}