修 platform_session_id 无差别下发导致抄送方邮件静默消失 + homeagent 补投漏去重
## platform_session_id 只发给归属方（Gateway） `notify.Recipients` 原来对所有参与方推同一个 `platform_session_id`， 而那是**会话级**的一个值。生产实测：会话 16845133 接管了 pi 的平台会话 `01a05a5e-…`，那封邮件抄送了 dsh@/home/program/agentmail.new。DSH 收到 同一个 id，在 ~/.dsh/sessions/ 里查不到（那是 /root/.pi/agent/sessions/ 下的文件），于是走进「平台侧会话已删」那道防线抛错。 那道防线本身是对的（N-8：不能退回新建，否则人在界面上看不到这封邮件带来 的对话），它拦下的却是「别人的会话」。异常被 ctx.logger.error 吞掉，而 DSH 的 logger 不进 journalctl —— 邮件静默消失，日志里一个字都没有。 - 新增 `repo.PlatformSessionFor` 一并返回归属 Agent：以镜像 `agent_platform_sessions.agent_name` 为准，镜像整表替换后退回 `sessions.from_agent`（AdoptPlatformSession 写在那里） - `PlatformIDOf` 变薄封装，保留原签名 - `notify.Recipients` 加 `platformFor(forName)`：归属方以外一律空串； 归属抽不到时（owner 空）也不下发 —— 宁可退回当普通会话处理， 也不让一个抽不到归属的 id 把邮件弄丢 - 归属与收件角色无关：归属方在抄送位上同样拿到 ## homeagent catchUp 漏 deliveredMails 去重 `go p.catchUp(…)` 与 `go p.sseLoop()` 是两个并发 goroutine，重启时窗口 重叠：SSE 推一次 + 补投拉一次 = 同一封邮件注入两遍。homeagent 的回信正文 印证了这一点（「之前的对话时序中已经收到并确认过多次了」）。另三个插件的 catchUp 都有这层双查，只有这里漏了。 去重放在循环内逐封查而不是拉完一批再筛：InjectInputSync 一封要跑几十秒， 那期间 SSE 完全可能已经投过后面那几封。 ## DSH 接管失败改用 console.error DSH 的 ctx.logger 不进 journalctl，投递失败是「发件人等不到回信」的唯一 线索。接管失败点与 SSE 分发的 catch 都改走 console.error，并带上 mail_id 与发件人。 ## 前端 ccAddress 移除（收尾上一轮未提交的改动） cc_list 里的 `.new` 是**原始意图**，不该被替换成主收件人的别名：每个抄送 方的 `.new` 是独立的 —— pi@/x.new 给 pi 开一条、dsh@/x.new 给 dsh 开另一 条，各有自己的别名。数据库存的就是原文。删掉 ccAddress，MailView / ThreadView 直接显示 c.raw。 ## 测试 - `internal/notify/notify_test.go` +3 例：挂真实 SSE 客户端读帧，验 归属方拿到 / 抄送方为空 / 归属方在抄送位也拿到 / 普通会话全空。 负向对照跑过：platformFor 无条件返回时两条用例失败 - `internal/repo/platform_owner_test.go` +3 例：镜像取归属、普通会话、 镜像被清后退回 from_agent - 修好 web/test/components/replyTarget.test.tsx（上一轮遗留的语法损坏）， 三条 .new 用例改成断言原样保留 - gateway 7 包全绿；web 176 例 + 主题 26；dsh 219 / pi 250 / opencode 201 ## 生产验证 - 抄送验证：jianf → pi（接管会话）cc dsh。DSH 正常建会话并回信「收到」， pi 走接管续谈 —— 两封回信都落在同一条线索上（此前 DSH 那封不存在） - homeagent 去重：连发两轮，其中一轮在邮件未处理完时重启 homeagent 造出 SSE/catchUp 并发窗口，两轮都只产生一封 Re: - homeagent SSE：换新 plugin.bin 后连续 89 分钟零断连（此前 2 小时 102 次 deadline exceeded 自激振荡）
This commit is contained in:
@ -74,9 +74,15 @@ func Recipients(ctx context.Context, m Mail) {
|
||||
// 反复建新会话。
|
||||
alias := repo.SessionAliasOf(ctx, m.SessionID)
|
||||
|
||||
// 这条会话是否接管了一条平台侧已存在的会话（人在 TUI/GUI 里开的那种）。
|
||||
// 插件据此决定 resume 还是新建；空串就是过去的行为。
|
||||
platformID := repo.PlatformIDOf(ctx, m.SessionID)
|
||||
// 这条会话是否接管了一条平台侧已存在的会话（人在 TUI/GUI 里开的那种），
|
||||
// 以及那条平台会话属于哪个 Agent。插件据此决定 resume 还是新建；
|
||||
// 空串就是过去的行为。
|
||||
//
|
||||
// **owner 必须参与分发判据**：platform_id 是会话级的一个值，而一封邮件
|
||||
// 可以有多个参与方。无差别下发会让抄送方拿一个属于别的平台的会话 id
|
||||
// 去自己磁盘上找文件，找不到就抛「平台侧会话已删」—— 邮件静默消失。
|
||||
// 生产实测过：pi 的会话 `01a05a5e-…` 被推给了抄送方 dsh。
|
||||
platformID, platformOwner := repo.PlatformSessionFor(ctx, m.SessionID)
|
||||
|
||||
mailType := m.MailType
|
||||
if mailType == "" {
|
||||
@ -87,6 +93,18 @@ func Recipients(ctx context.Context, m Mail) {
|
||||
replyTo = m.From
|
||||
}
|
||||
|
||||
// platformFor 只把 platform_session_id 给归属方。
|
||||
//
|
||||
// owner 为空（镜像里没这条、sessions.from_agent 也空）时一律不下发：
|
||||
// 宁可退回「当普通会话处理」（插件新建一条，人在界面上看不到），
|
||||
// 也不能让一个抽不到归属的 id 把邮件弄丢。
|
||||
platformFor := func(forName string) string {
|
||||
if platformID == "" || platformOwner == "" || forName != platformOwner {
|
||||
return ""
|
||||
}
|
||||
return platformID
|
||||
}
|
||||
|
||||
payload := func(role, workspace, forName string) map[string]interface{} {
|
||||
p := map[string]interface{}{
|
||||
"mail_id": m.MailID.String(),
|
||||
@ -112,7 +130,10 @@ func Recipients(ctx context.Context, m Mail) {
|
||||
//
|
||||
// 插件必须 resume 而不是新建：新建会让人在 TUI 里看不到这封邮件
|
||||
// 带来的对话，而那正是接管这条会话的目的。
|
||||
"platform_session_id": platformID,
|
||||
//
|
||||
// **只发给归属方**：其余参与方拿到它只会去自己磁盘上找一个
|
||||
// 不存在的会话文件，然后按 N-8 报错丢掉这封邮件。
|
||||
"platform_session_id": platformFor(forName),
|
||||
}
|
||||
if m.Origin != "" {
|
||||
p["origin"] = m.Origin
|
||||
|
||||
171
gateway/internal/notify/notify_test.go
Normal file
171
gateway/internal/notify/notify_test.go
Normal file
@ -0,0 +1,171 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/agentmail/gateway/internal/models"
|
||||
"github.com/agentmail/gateway/internal/repo"
|
||||
"github.com/agentmail/gateway/internal/sse"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func setupTestDB(t *testing.T) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
if err := db.Connect(context.Background(), filepath.Join(dir, "test.db")); err != nil {
|
||||
t.Fatalf("connect: %v", err)
|
||||
}
|
||||
if err := db.Migrate(context.Background()); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
t.Cleanup(db.Close)
|
||||
}
|
||||
|
||||
func seedAgent(t *testing.T, name string) {
|
||||
t.Helper()
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`INSERT INTO agents (agent_name, secret, platform, status) VALUES ($1, 'x', $1, 'online')`,
|
||||
name); err != nil {
|
||||
t.Fatalf("seed agent %s: %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// attach 挂一个真实的 SSE 客户端并返回「读出这个 Agent 收到的 new_mail payload」的闭包。
|
||||
//
|
||||
// 走真实的 sse.Default 而不是替换发送函数：要验的正是「谁收到什么」，
|
||||
// 而分发逻辑就在 Manager 里 —— 把它换掉等于不验。
|
||||
func attach(t *testing.T, agentName string) func() map[string]any {
|
||||
t.Helper()
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/events/stream", nil)
|
||||
c := sse.Default.AddClient(rec, req, agentName, "")
|
||||
if c == nil {
|
||||
t.Fatalf("AddClient(%s) 返回 nil", agentName)
|
||||
}
|
||||
t.Cleanup(func() { sse.Default.RemoveClient(c.ID) })
|
||||
|
||||
return func() map[string]any {
|
||||
// SSE 帧形如 `id: N\nevent: new_mail\ndata: {…}\n\n`
|
||||
for _, frame := range strings.Split(rec.Body.String(), "\n\n") {
|
||||
if !strings.Contains(frame, "event: new_mail") {
|
||||
continue
|
||||
}
|
||||
for _, line := range strings.Split(frame, "\n") {
|
||||
if !strings.HasPrefix(line, "data: ") {
|
||||
continue
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &m); err == nil {
|
||||
return m
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// seedAdopted 建一条接管了 owner 的平台会话的本侧会话。
|
||||
func seedAdopted(t *testing.T, owner, platformID, workspace string) uuid.UUID {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
if err := repo.ReplacePlatformSessions(ctx, owner, []repo.PlatformSession{
|
||||
{PlatformID: platformID, Workspace: workspace, Slug: "项目定位", Title: "项目定位"},
|
||||
}); err != nil {
|
||||
t.Fatalf("ReplacePlatformSessions: %v", err)
|
||||
}
|
||||
id, err := repo.AdoptPlatformSession(ctx, owner, platformID, "项目定位", workspace, "项目定位")
|
||||
if err != nil {
|
||||
t.Fatalf("AdoptPlatformSession: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
// platform_session_id 只该发给归属方。
|
||||
//
|
||||
// 生产事故：会话接管了 pi 的 `01a05a5e-…`，而那封邮件抄送了 dsh。DSH 收到同一个
|
||||
// id，在 `~/.dsh/sessions/` 里查不到（那是 `/root/.pi/agent/sessions/` 下的文件），
|
||||
// 于是按 N-8 抛「平台侧会话已删」——邮件静默消失，日志里一个字都没有。
|
||||
func TestRecipients_PlatformIDOnlyToOwner(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "pi")
|
||||
seedAgent(t, "dsh")
|
||||
readPi, readDsh := attach(t, "pi"), attach(t, "dsh")
|
||||
|
||||
sessionID := seedAdopted(t, "pi", "pid-pi-1", "/w")
|
||||
|
||||
Recipients(context.Background(), Mail{
|
||||
SessionID: sessionID,
|
||||
MailID: uuid.New(),
|
||||
From: "jianf",
|
||||
To: models.Address{Name: "pi", Path: "/w"},
|
||||
CC: []models.Address{{Name: "dsh", Path: "/w"}},
|
||||
Subject: "任务",
|
||||
})
|
||||
|
||||
pi, dsh := readPi(), readDsh()
|
||||
if pi == nil {
|
||||
t.Fatal("归属方 pi 没收到 new_mail")
|
||||
}
|
||||
if dsh == nil {
|
||||
t.Fatal("抄送方 dsh 没收到 new_mail（抄送方必须单独推）")
|
||||
}
|
||||
if v := pi["platform_session_id"]; v != "pid-pi-1" {
|
||||
t.Errorf("归属方 pi 的 platform_session_id = %v, want pid-pi-1", v)
|
||||
}
|
||||
if v := dsh["platform_session_id"]; v != "" {
|
||||
t.Errorf("抄送方 dsh 的 platform_session_id = %v, want 空串（那是 pi 的会话文件）", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 归属方在抄送位上也要拿到：归属与收件角色无关。
|
||||
func TestRecipients_PlatformIDToOwnerEvenAsCC(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "pi")
|
||||
seedAgent(t, "dsh")
|
||||
readPi, readDsh := attach(t, "pi"), attach(t, "dsh")
|
||||
|
||||
sessionID := seedAdopted(t, "pi", "pid-pi-2", "/w")
|
||||
|
||||
Recipients(context.Background(), Mail{
|
||||
SessionID: sessionID,
|
||||
MailID: uuid.New(),
|
||||
From: "jianf",
|
||||
To: models.Address{Name: "dsh", Path: "/w"},
|
||||
CC: []models.Address{{Name: "pi", Path: "/w"}},
|
||||
Subject: "任务",
|
||||
})
|
||||
|
||||
if v := readPi()["platform_session_id"]; v != "pid-pi-2" {
|
||||
t.Errorf("抄送位上的归属方 pi = %v, want pid-pi-2", v)
|
||||
}
|
||||
if v := readDsh()["platform_session_id"]; v != "" {
|
||||
t.Errorf("主收件人 dsh = %v, want 空串", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 普通（非接管）会话：谁都不该拿到 platform id。
|
||||
func TestRecipients_PlainSessionNoPlatformID(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "pi")
|
||||
readPi := attach(t, "pi")
|
||||
|
||||
id, err := repo.CreateSession(context.Background(), nil, "pi", "普通", "/w")
|
||||
if err != nil {
|
||||
t.Fatalf("CreateSession: %v", err)
|
||||
}
|
||||
Recipients(context.Background(), Mail{
|
||||
SessionID: id, MailID: uuid.New(), From: "jianf",
|
||||
To: models.Address{Name: "pi", Path: "/w"}, Subject: "任务",
|
||||
})
|
||||
|
||||
if v := readPi()["platform_session_id"]; v != "" {
|
||||
t.Errorf("普通会话 = %v, want 空串", v)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user