fix(inbox): read_inbox 按会话收窄 —— 修「不同 session 的 agent 都能看到全部邮件」
用户问：「你之前不是说你已经处理了不同 session 的 agent 都可以看到全部邮件的 问题了吗？」——**我得先纠正事实：上一轮我只做了诊断并问要不要动手，没有实施。** 这是我的表述问题（把"已定位并给了方案"说成了像"已处理"）。现在实施。 ## 缺陷 `read_inbox` 是**按 Agent** 的：列的是该 Agent 的全部未读（含别的会话的来信）， 并按契约把列出来的都标成已读 ⇒ A 会话的 worker 标掉 B 会话的未读。平时看不出来 （SSE 事件在途时队列兜着），但桥重启/漏事件后的补投判据是 `?status=unread` —— 被标掉的那封**再也不会补投** ⇒ 静默丢信。现场实例：另一条会话的来信在 `mail_reads` 里的 reader=pi、时间正是我读自己收件箱的那一刻。 ## 改动 - **网关**：`GET /mail/inbox` 与 `POST /mail/read` 支持可选 `session_id`。 不带 = 旧语义（整个 Agent 的收件箱，浏览器/脚本仍可用）；带了就只在这条会话内 列与标。`ListInbox` / `MarkAllInboxReadFor` 保持原签名并委托给新变体 —— 老调用点一个都不用改。 - **pi 桥**：`read_inbox` 把自己那条会话拼进 URL（worker 通过闭包把**邮件会话 id** 递给工具，而不是在启动时取快照）。 ## 判据 - repo 三条：列表按会话收窄（含"不带会话时两条都在"的反向对照）、 ★"标会话 A 不动会话 B"、会话内计数与列表口径一致（否则界面会出现"徽标 2、列表 1"）。 - handler/网关：非法 `session_id` ⇒ 400（不静默忽略）。 - pi 接线三条（URL 拼了收窄、worker 递了 id、判据自检：旧写法必须判红）。 - 线上只读 E2E：两条真实会话 A/B 列表**无交集**、不带会话能列出全部、非法 id 400。 ## 过程中测试当场抓到"只改了一半" `MarkAllInboxReadForSession` 里插 `mail_reads` 的语句我加了会话条件， **刷新冗余列的 UPDATE 忘了加** ⇒ 返回的"标掉几封"变成 2（应 1）。 判据一眼看出来了 —— 这类"改一半"正是这次要防的。 ## 范围（诚实说明） 另外四家桥（dsh/opencode/zcode/homeagent）的 `read_inbox` 工具签名里**没有会话上下文** （`execute(args)` / `execute(args, ctx)` 各不相同），要按各自框架的上下文 API 接线， 不是一行改动 ⇒ **未做**，列为待办（位置已定位）。所以：pi 上这个缺陷已消除， 另外四家仍在。 ## 部署 网关已部署并线上验证；pi 桥的部署**延迟到本轮结束后 150 秒**执行 （重启 pi 桥会掐掉我自己这一轮 —— 之前真发生过），日志 `/var/log/agentmail-pi-redeploy.log`，可用 `node deploy/check-deploy-drift.mjs` 核对。
This commit is contained in:
@ -539,7 +539,22 @@ func GetInbox(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
mails, err := repo.ListInbox(r.Context(), agentName, status, limit)
|
||||
// 可选会话收窄：桥的 read_inbox 会带上自己那条会话。
|
||||
//
|
||||
// 不带 = 整个 Agent 的收件箱（旧语义，浏览器/脚本仍可用）；带了就只列这条线索 ——
|
||||
// 否则 A 会话的 worker 会把 B 会话的未读也列出来并标成已读，桥重启后的补投
|
||||
// 判据（?status=unread）就再也看不到那封信（用户报的"都能看到全部邮件"）。
|
||||
var sessionID uuid.UUID
|
||||
if raw := r.URL.Query().Get("session_id"); raw != "" {
|
||||
id, perr := uuid.Parse(raw)
|
||||
if perr != nil {
|
||||
Error(w, http.StatusBadRequest, "非法的 session_id")
|
||||
return
|
||||
}
|
||||
sessionID = id
|
||||
}
|
||||
|
||||
mails, err := repo.ListInboxScoped(r.Context(), agentName, status, limit, sessionID)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to list inbox")
|
||||
return
|
||||
@ -637,6 +652,8 @@ func parseInt(s string) (int, error) {
|
||||
}
|
||||
|
||||
type markReadRequest struct {
|
||||
// SessionID 可选：带了就只在这条会话内标（见 GetInbox 里那段说明）。
|
||||
SessionID string `json:"session_id"`
|
||||
// MailIDs 要标记为已读的邮件；省略/为空 = 把收件箱里全部未读标掉。
|
||||
MailIDs []string `json:"mail_ids"`
|
||||
}
|
||||
@ -667,12 +684,22 @@ func MarkInboxRead(w http.ResponseWriter, r *http.Request) {
|
||||
// 不给 id 就把收件箱里全部未读标掉。
|
||||
// 这是 Agent 最常见的用法：一轮处理完，剩下的都不必再看。
|
||||
if len(req.MailIDs) == 0 {
|
||||
n, err := repo.MarkAllInboxReadFor(r.Context(), agentName)
|
||||
var sessionID uuid.UUID
|
||||
scope := "all"
|
||||
if raw := strings.TrimSpace(req.SessionID); raw != "" {
|
||||
id, perr := uuid.Parse(raw)
|
||||
if perr != nil {
|
||||
Error(w, http.StatusBadRequest, "非法的 session_id")
|
||||
return
|
||||
}
|
||||
sessionID, scope = id, "session"
|
||||
}
|
||||
n, err := repo.MarkAllInboxReadForSession(r.Context(), agentName, sessionID)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to mark read")
|
||||
return
|
||||
}
|
||||
JSON(w, http.StatusOK, map[string]any{"status": "read", "marked": n, "scope": "all"})
|
||||
JSON(w, http.StatusOK, map[string]any{"status": "read", "marked": n, "scope": scope})
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@ -504,7 +504,26 @@ func MarkMailRead(ctx context.Context, id uuid.UUID, reader string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
/*
|
||||
─── 会话维度（2026-09-14）───────────────────────────────────────────────
|
||||
|
||||
用户报的缺陷：「不同 session 的 agent 都可以看到全部邮件」。
|
||||
|
||||
原先 `read_inbox` 是**按 Agent** 的：列的是该 Agent 的全部未读（含别的会话的来信），
|
||||
并且按契约把列出来的都标成已读 ⇒ A 会话的 worker 会把 B 会话的未读标掉。
|
||||
平时看不出来（SSE 事件在途时队列兜着），但桥重启/漏事件后的补投判据是
|
||||
`?status=unread` —— 被别人标掉的那封**再也不会补投** ⇒ 静默丢信。
|
||||
|
||||
修法：列表与"全部标已读"都支持按 `session_id` 收窄，桥把自己的会话传进来。
|
||||
原函数保持原语义（不带会话 = 整个 Agent 的收件箱），新增带会话的变体 ——
|
||||
老调用点一个都不用改。
|
||||
*/
|
||||
func ListInbox(ctx context.Context, agentName, status string, limit int) ([]models.Mail, error) {
|
||||
return ListInboxScoped(ctx, agentName, status, limit, uuid.Nil)
|
||||
}
|
||||
|
||||
// ListInboxScoped 与 ListInbox 相同，但 `sessionID` 非零时只列该会话的邮件。
|
||||
func ListInboxScoped(ctx context.Context, agentName, status string, limit int, sessionID uuid.UUID) ([]models.Mail, error) {
|
||||
q := `SELECT m.mail_id, m.session_id, m.parent_mail_id,
|
||||
m.from_name, m.from_workspace, m.to_name, m.to_workspace,
|
||||
m.cc_list, m.subject, m.body, m.mail_type, COALESCE(m.permission_result,'') AS permission_result,
|
||||
@ -520,6 +539,12 @@ func ListInbox(ctx context.Context, agentName, status string, limit int) ([]mode
|
||||
WHERE (m.to_name = $1 OR ` + db.CCHas("m.cc_list", 1) + `)
|
||||
AND s.status <> 'archived'`
|
||||
args := []any{agentName}
|
||||
if sessionID != uuid.Nil {
|
||||
// 会话收窄：只列这条线索里的邮件（见上面「会话维度」的说明）
|
||||
args = append(args, sessionID)
|
||||
q += fmt.Sprintf(` AND m.session_id = $%d`, len(args))
|
||||
}
|
||||
|
||||
if status != "" && status != "all" {
|
||||
// 未读/已读都按**这个读者**算（原先直接比 m.status，于是被抄送方读掉
|
||||
// 别人的未读也跟着变 —— 这就是要修的那条）
|
||||
@ -1564,19 +1589,37 @@ func MarkMailsReadFor(ctx context.Context, recipient string, ids []uuid.UUID) (i
|
||||
// 排除已归档会话：那些邮件在收件箱里根本看不到，
|
||||
// 标掉它们只会让「标记了 N 封」这个数字与用户看到的对不上。
|
||||
func MarkAllInboxReadFor(ctx context.Context, recipient string) (int, error) {
|
||||
if err := markReadFor(ctx, recipient,
|
||||
`(m.to_name = $1 OR `+db.CCHas("m.cc_list", 1)+`)
|
||||
AND m.session_id IN (SELECT session_id FROM sessions WHERE status <> 'archived')`,
|
||||
recipient); err != nil {
|
||||
return MarkAllInboxReadForSession(ctx, recipient, uuid.Nil)
|
||||
}
|
||||
|
||||
// MarkAllInboxReadForSession 只标掉某条会话里发给 recipient 的未读。
|
||||
//
|
||||
// 为什么需要：Agent 的「不给 mail_ids，全部标掉」在会话驱动的 worker 里会跨会话
|
||||
// 误伤（见 ListInboxScoped 上面那段说明）。不带 sessionID（uuid.Nil）时是旧语义。
|
||||
func MarkAllInboxReadForSession(ctx context.Context, recipient string, sessionID uuid.UUID) (int, error) {
|
||||
scope := `(m.to_name = $1 OR ` + db.CCHas("m.cc_list", 1) + `)
|
||||
AND m.session_id IN (SELECT session_id FROM sessions WHERE status <> 'archived')`
|
||||
args := []any{recipient}
|
||||
if sessionID != uuid.Nil {
|
||||
scope += ` AND m.session_id = $2`
|
||||
args = append(args, sessionID)
|
||||
}
|
||||
if err := markReadFor(ctx, recipient, scope, args...); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// 同上：刷新冗余列，未读判据在 mail_reads
|
||||
res, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE mails SET status = 'read'
|
||||
// 同上：刷新冗余列，未读判据在 mail_reads。
|
||||
// ★ 这条 UPDATE 也必须跟着同一个 scope —— 只给上面的 INSERT 收窄、漏掉它，
|
||||
// 返回的计数与"实际标掉多少"都会跨会话（测试当场抓到：应当 1 封、实际 2 封）。
|
||||
upd := `UPDATE mails SET status = 'read'
|
||||
WHERE status = 'unread'
|
||||
AND (to_name = $1 OR `+db.CCHas("cc_list", 1)+`)
|
||||
AND session_id IN (SELECT session_id FROM sessions WHERE status <> 'archived')
|
||||
`, recipient)
|
||||
AND (to_name = $1 OR ` + db.CCHas("cc_list", 1) + `)
|
||||
AND session_id IN (SELECT session_id FROM sessions WHERE status <> 'archived')`
|
||||
updArgs := []any{recipient}
|
||||
if sessionID != uuid.Nil {
|
||||
upd += ` AND session_id = $2`
|
||||
updArgs = append(updArgs, sessionID)
|
||||
}
|
||||
res, err := db.DB.ExecContext(ctx, upd, updArgs...)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
152
server/internal/repo/session_scope_test.go
Normal file
152
server/internal/repo/session_scope_test.go
Normal file
@ -0,0 +1,152 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/agentmail/gateway/internal/models"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
/*
|
||||
会话维度：一个 Agent 同时在多条会话里干活时，**读一条不能动另一条**。
|
||||
|
||||
# 用户报的缺陷
|
||||
|
||||
「不同 session 的 agent 都可以看到全部邮件」—— 原先 `read_inbox` 按 **Agent** 列，
|
||||
并且把列出来的都标成已读 ⇒ A 会话的 worker 会把 B 会话的未读标掉。
|
||||
|
||||
平时看不出来（SSE 事件在途时队列兜着），但桥重启/漏事件之后的补投判据是
|
||||
`?status=unread` —— 被别人标掉的那封信**再也不会补投**。这不是"少看一封"，
|
||||
是静默丢信。现场实例：另一条会话的来信被我在读自己收件箱时标成已读
|
||||
（`mail_reads` 里那行的 reader=pi、时间就是那次读取的时刻）。
|
||||
|
||||
# 判据要两侧都验
|
||||
|
||||
只验"收窄后看不到别人的"是不够的：把列表直接返回空也能过。所以同时验
|
||||
"不带会话时两条都看得到"（反向对照），以及"全标已读"两个变体。
|
||||
*/
|
||||
|
||||
// seedMailInSession 在指定会话里造一封给 recipient 的未读邮件。
|
||||
func seedMailInSession(t *testing.T, sessionID uuid.UUID, recipient string) uuid.UUID {
|
||||
t.Helper()
|
||||
var id uuid.UUID
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body)
|
||||
VALUES ($1, 'sender', $2, 's', 'b') RETURNING mail_id`,
|
||||
sessionID, recipient).Scan(&id); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func twoSessionsWithUnread(t *testing.T, agent string) (uuid.UUID, uuid.UUID, uuid.UUID, uuid.UUID) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
s1, err := CreateSession(ctx, nil, "human", "会话一", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s2, err := CreateSession(ctx, nil, "human", "会话二", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s1, s2, seedMailInSession(t, s1, agent), seedMailInSession(t, s2, agent)
|
||||
}
|
||||
|
||||
func TestInboxListIsScopedBySession(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
s1, s2, m1, m2 := twoSessionsWithUnread(t, "bot")
|
||||
|
||||
// 反向对照：不带会话 = 整个 Agent 的收件箱，两条都看得到
|
||||
all, err := ListInbox(ctx, "bot", "unread", 50)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !hasMail(all, m1) || !hasMail(all, m2) {
|
||||
t.Fatalf("不带会话时应当两条都列出来（%d 条）", len(all))
|
||||
}
|
||||
|
||||
// ★ 收窄到会话一：只有它那条
|
||||
scoped, err := ListInboxScoped(ctx, "bot", "unread", 50, s1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !hasMail(scoped, m1) {
|
||||
t.Fatal("会话一的邮件应当在会话一的列表里")
|
||||
}
|
||||
if hasMail(scoped, m2) {
|
||||
t.Fatal("★ 会话二的邮件不该出现在会话一的列表里（用户报的缺陷）")
|
||||
}
|
||||
|
||||
// 会话二同理（别只验一侧 —— 方向反了也能"过"）
|
||||
scoped2, err := ListInboxScoped(ctx, "bot", "unread", 50, s2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if hasMail(scoped2, m1) || !hasMail(scoped2, m2) {
|
||||
t.Fatalf("会话二的列表应只含会话二的邮件")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkAllReadCanBeScopedToSession(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
s1, s2, _, _ := twoSessionsWithUnread(t, "bot")
|
||||
|
||||
// 只标会话一
|
||||
n, err := MarkAllInboxReadForSession(ctx, "bot", s1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("应当只标掉 1 封，实际 %d", n)
|
||||
}
|
||||
left, err := ListInboxScoped(ctx, "bot", "unread", 50, s2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(left) != 1 {
|
||||
t.Fatalf("★ 会话二的未读必须还在（实际剩 %d 条）—— 被跨会话标掉就是静默丢信", len(left))
|
||||
}
|
||||
|
||||
// 反向对照：不带会话时两条都会被标掉
|
||||
if _, err := MarkAllInboxReadFor(ctx, "bot"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again, _ := ListInbox(ctx, "bot", "unread", 50); len(again) != 0 {
|
||||
t.Fatalf("不带会话的全标应当清空整个收件箱，实际剩 %d", len(again))
|
||||
}
|
||||
}
|
||||
|
||||
func TestScopedCountsMatchScopedList(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
s1, s2, _, _ := twoSessionsWithUnread(t, "bot")
|
||||
|
||||
// 会话内的未读数（会话列表徽标用）与"只列这条会话"的口径必须一致，
|
||||
// 否则界面上会出现"徽标 2、列表 1"这种自相矛盾。
|
||||
n1, err := CountUnreadInSession(ctx, "bot", s1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
list1, err := ListInboxScoped(ctx, "bot", "unread", 50, s1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n1 != len(list1) || n1 != 1 {
|
||||
t.Fatalf("会话一：计数 %d、列表 %d，期望都是 1", n1, len(list1))
|
||||
}
|
||||
_ = s2
|
||||
}
|
||||
|
||||
func hasMail(list []models.Mail, id uuid.UUID) bool {
|
||||
for _, m := range list {
|
||||
if m.ID == id {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
Reference in New Issue
Block a user