feat(线索树)★★: 会话级父子 + 树视图 —— 跨会话分叉第一次可见
## 要解决的问题
用户 2026-10-04：「对话树实现得非常原始，根本没有形成/展示为树结构」。
★ 先纠正我自己的一个误判：我先前只看 `ContactPanel`（会话列表）就断言
「前端纯平铺、零层级」。那是不完整的检查 —— `ThreadView.tsx` 早就在用
`node.depth` 做缩进 + 连接线渲染**会话内**的邮件树，服务端 `DescendantsRaw`
也早就有 `WITH RECURSIVE … lvl`。所以**会话内**的树是有的。
真正缺的是**跨会话**：`sessions` 表根本没有 `parent_session_id`，
而生产库实测有 **6 封**邮件的 parent 指向**另一条会话**，
界面上它们是几行互不相干的东西。而「A 交给 B 之后 B 继续推」正是
协作里最常见的形状 —— 那 6 条里就有 `本机-agent-能力盘点 → 渲染自检`
这种纯分叉。
## 回填后的真实拓扑（生产数据）
harmony-emu-unblock ─┐
deploy-pi-bridge-… ─┴→ 邮件驱动…项目概述 ─┬→ 核实-HomeAgent-mail-bridge
└→ 邮件驱动…项目概述-2 → 时间显示自检
本机-agent-能力盘点 → 渲染自检
## 可见性：逐节点过滤 + 剪断不可见祖先（安全边界）
树天然会把父节点带给子节点，而「我能看见 B」≠「我能看见 B 的父 A」
（A 可能是别人与别人的对话）。
所以按节点过滤之后**必须重算 parent**：不可见的祖先一律跳过、树在可见处
重新起根。只过滤不重算，输出里就带着不可见父的 session_id 与标题 ——
一条真实的泄露路径，而且它藏在「树视图」这个新功能里，没人会想到去查。
**admin 看全部**（用户 2026-10-04 定的分层）。
★ 与 Agent 侧那条边界无关：`AgentMayReadSession` 是 15e4fe9 / 095213b
修出来的越权防护（Agent 只能读自己参与过的会话）；本端点属人类登录态，
admin 全看是显式授权的。两者语义不同，不要混谈。
## 回填只跑一次
回填改的是会话**拓扑**，不是派生数据。若每次 Migrate 都跑，
「某人手工把 parent 改对」会在每次重启时被悄悄改回去 —— 与既有
`backfillMailReads` 同族（其注释：每次跑会把「某抄送方读过」按主收件人
写成已读，正是那次要修的语义错误）。故用 `app_meta` marker 守住。
★ 判据分两包：`repo` 测回填口径与剪枝，`db` 测「二次 Migrate 不覆盖人工设置」。
只写在 repo 包就只测到「回填幂等」，测不到 Migrate 那一层。
## 实现中修掉的三个真 bug（都被判据抓到）
1. **参数顺序反了** —— `t.lvl < ?` 是 SQL 里第一个占位符，我放在 args 末尾
⇒ 根查询拿到一个整数、匹配不到任何行，树只返回递归分支那半。
2. **环下打满 55s 超时** —— 计数写成相关子查询，环下每个节点都重跑一次
mails 计数。改成 `LEFT JOIN` 两个聚合后 0.01s。
3. **`IN (NULL)`** —— parentIDs 为空时该表达式恒不匹配任何行。
## 判据（8 格）
含防环（环下 0.01s + 耗时断言）、不覆盖人工父、不把同会话内邮件父子
当会话父子、不回填不可见祖先（两个变异分别去掉「过滤」「剪断」都转红）。
★ 判据自己错了一次：「隐藏不可见祖先」那格我写成「对每个可见节点都断言
depth==0」，但 B 重起根（0）、C 仍挂在 B 下（1），两者不同 —— 判据红而代码
是对的。（又是「判据比语义宽/窄」那一族。）
14 包全绿。
This commit is contained in:
@ -234,6 +234,9 @@ func main() {
|
||||
r.Get("/mail/{id}", handler.GetMail)
|
||||
r.Get("/mail/{id}/thread", handler.GetMailThread)
|
||||
r.Post("/mail/{id}/read", handler.MarkMailRead)
|
||||
// 会话级线索树（2026-10-04）。逐节点按可见性过滤并**剪断不可见祖先**；
|
||||
// admin 看全部（用户 2026-10-04 定的分层）。
|
||||
r.Get("/sessions/tree", handler.ListSessionTree)
|
||||
r.Get("/sessions/{id}", handler.GetSession)
|
||||
r.Get("/sessions/{id}/mails", handler.GetSessionMails)
|
||||
r.Put("/sessions/{id}/alias", handler.UpdateSessionAlias)
|
||||
|
||||
@ -43,6 +43,20 @@ func Migrate(ctx context.Context) error {
|
||||
return fmt.Errorf("migrate: 未初始化的方言")
|
||||
}
|
||||
|
||||
// 会话级线索树的一次性回填（2026-10-04）。
|
||||
//
|
||||
// 为什么**不**每次 Migrate 都跑：它会改**真实会话的拓扑**（给 6 条会话写上
|
||||
// parent_session_id）。若判据后来变宽、或有人手工调整过 parent，那些改动
|
||||
// 会在重启时被悄悄改回去 —— 「不可见的人工意图」正是既有 backfillMailReads
|
||||
// 用 marker 守住的原因（见它的注释：每次跑会把「某抄送方读过」按主收件人
|
||||
// 写成已读）。
|
||||
//
|
||||
// 回填口径见 repo.BackfillSessionParents：只认**直接跨会话分叉**，且不改
|
||||
// 已有 parent_session_id 的会话。
|
||||
if err := backfillSessionParentsOnce(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := backfillMailReads(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -284,6 +298,17 @@ var sqliteAddColumns = []struct{ table, column, ddl string }{
|
||||
// （Agent 回信的 from_workspace 存的是 Agent 名而不是路径），强行回填只会
|
||||
// 造出一批看起来有值实际是错的数据。
|
||||
{"sessions", "workspace", "ALTER TABLE sessions ADD COLUMN workspace TEXT NOT NULL DEFAULT ''"},
|
||||
// 会话级线索树的父指针（2026-10-04）。
|
||||
//
|
||||
// 为什么需要它：邮件级父子（`mails.parent_mail_id`）早已有，ThreadView 也早
|
||||
// 在用 depth 渲染**会话内**的树 —— 但**跨会话**的分叉完全不可见：生产库里
|
||||
// 有 6 封邮件的 parent 指向**另一条会话**，界面上它们是互不相干的几行。
|
||||
// 而「A 把这件事交给 B 之后 B 继续推」正是协作里最常见的形状，
|
||||
// 那 6 条里就有 `本机-agent-能力盘点 → 渲染自检` 这类纯分叉。
|
||||
//
|
||||
// 默认空：没有父的会话就是**根节点**。不据历史数据乱回填 —— 父子关系只
|
||||
// 在「确实是分叉」时才成立，猜错会让线索拓扑失真（比没有更糟）。
|
||||
{"sessions", "parent_session_id", "ALTER TABLE sessions ADD COLUMN parent_session_id TEXT"},
|
||||
// 日历多收件人。旧库默认 '[]'：读的时候由 EffectiveRecipients() 退回
|
||||
// to_address / agent_name，历史事件因此继续工作，不需要数据迁移。
|
||||
{"calendar_events", "recipients", "ALTER TABLE calendar_events ADD COLUMN recipients TEXT NOT NULL DEFAULT '[]'"},
|
||||
@ -392,3 +417,53 @@ func columnExists(ctx context.Context, table, column string) (bool, error) {
|
||||
table, column).Scan(&n)
|
||||
return n > 0, err
|
||||
}
|
||||
|
||||
// backfillSessionParentsOnce 给跨会话分叉的会话补 parent_session_id，只跑一次。
|
||||
//
|
||||
// 一次性守卫的原因见 Migrate 里那段注释：它改的是会话**拓扑**，不是派生数据。
|
||||
func backfillSessionParentsOnce(ctx context.Context) error {
|
||||
const marker = "session_tree_backfill_v1"
|
||||
var v string
|
||||
err := DB.QueryRowContext(ctx, `SELECT value FROM app_meta WHERE key = $1`, marker).Scan(&v)
|
||||
if err == nil {
|
||||
return nil // 已回填过
|
||||
}
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
return fmt.Errorf("backfill session_tree: 读标记失败: %w", err)
|
||||
}
|
||||
|
||||
if _, err := DB.ExecContext(ctx, `
|
||||
UPDATE sessions
|
||||
SET parent_session_id = (
|
||||
SELECT ps.session_id
|
||||
FROM mails m
|
||||
JOIN mails pm ON m.parent_mail_id = pm.mail_id
|
||||
JOIN sessions ps ON ps.session_id = pm.session_id
|
||||
WHERE m.session_id = sessions.session_id
|
||||
AND ps.session_id <> sessions.session_id
|
||||
ORDER BY m.created_at ASC
|
||||
LIMIT 1
|
||||
)
|
||||
WHERE parent_session_id IS NULL
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM mails m
|
||||
WHERE m.session_id = sessions.session_id
|
||||
AND m.parent_mail_id IS NOT NULL
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM mails pm
|
||||
WHERE pm.mail_id = m.parent_mail_id
|
||||
AND pm.session_id <> sessions.session_id
|
||||
)
|
||||
)
|
||||
`); err != nil {
|
||||
return fmt.Errorf("backfill session_tree: 回填失败: %w", err)
|
||||
}
|
||||
|
||||
if _, err := DB.ExecContext(ctx,
|
||||
`INSERT INTO app_meta (key, value) VALUES ($1, $2)`, marker,
|
||||
time.Now().UTC().Format(time.RFC3339)); err != nil {
|
||||
return fmt.Errorf("backfill session_tree: 写标记失败: %w", err)
|
||||
}
|
||||
fmt.Println("会话线索树回填完成")
|
||||
return nil
|
||||
}
|
||||
|
||||
63
server/internal/db/migrate_session_tree_test.go
Normal file
63
server/internal/db/migrate_session_tree_test.go
Normal file
@ -0,0 +1,63 @@
|
||||
package db
|
||||
|
||||
/*
|
||||
一次性回填守卫的判据（2026-10-04）。
|
||||
|
||||
★ 这一格**必须放在 db 包**：守卫（`app_meta` 标记）在 migrate.go 里，
|
||||
而回填实现在 repo 包。放在 repo 包测就只测到「回填本身幂等」，
|
||||
测不到「Migrate 二次执行会不会覆盖人工设置」—— 那正是需要防的事。
|
||||
|
||||
后果不是理论上的：回填改的是会话**拓扑**。若每次 Migrate 都跑，
|
||||
「某人手工把 parent 改对」这个意图会在每次重启时被悄悄改回去。
|
||||
既有 backfillMailReads 踩过同族坑（其注释：每次跑会把「某抄送方读过」
|
||||
按主收件人写成已读，正是那次要修的语义错误）。
|
||||
*/
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSessionParentsBackfillRunsOnlyOnce(t *testing.T) {
|
||||
Close()
|
||||
Connect(context.Background(), "sqlite://"+t.TempDir()+"/once.db")
|
||||
if err := Migrate(context.Background()); err != nil {
|
||||
t.Fatalf("首次迁移: %v", err)
|
||||
}
|
||||
t.Cleanup(Close)
|
||||
|
||||
ctx := context.Background()
|
||||
// Migrate 已跑过一次 ⇒ 标记已写 ⇒ 之后的回填都不该再动任何东西。
|
||||
manual := mustSession(t, "manual-parent")
|
||||
child := mustSession(t, "child")
|
||||
if _, err := DB.ExecContext(context.Background(), `UPDATE sessions SET parent_session_id = ? WHERE session_id = ?`,
|
||||
manual, child); err != nil {
|
||||
t.Fatalf("设人工父: %v", err)
|
||||
}
|
||||
|
||||
// 第二次 Migrate（模拟重启）
|
||||
if err := Migrate(context.Background()); err != nil {
|
||||
t.Fatalf("二次迁移: %v", err)
|
||||
}
|
||||
|
||||
var got string
|
||||
if err := DB.QueryRowContext(ctx, `SELECT parent_session_id FROM sessions WHERE session_id = ?`,
|
||||
child).Scan(&got); err != nil {
|
||||
t.Fatalf("读父: %v", err)
|
||||
}
|
||||
if got != manual {
|
||||
t.Errorf("★ 二次 Migrate 覆盖了人工设置的 parent_session_id：%s ⇒ 应为 %s", got, manual)
|
||||
}
|
||||
}
|
||||
|
||||
func mustSession(t *testing.T, alias string) string {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
var id string
|
||||
if err := DB.QueryRowContext(ctx,
|
||||
`INSERT INTO sessions (session_alias, subject, status, workspace, from_agent)
|
||||
VALUES ($1,'x','active','/tmp','pi') RETURNING session_id`, alias).Scan(&id); err != nil {
|
||||
t.Fatalf("建会话: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
86
server/internal/handler/session_tree.go
Normal file
86
server/internal/handler/session_tree.go
Normal file
@ -0,0 +1,86 @@
|
||||
package handler
|
||||
|
||||
/*
|
||||
GET /api/v1/sessions/tree —— 会话级线索树（2026-10-04）。
|
||||
|
||||
# 可见性：逐节点过滤，且 admin 看全部
|
||||
|
||||
用户点开树时看到的是「**他能看到的**那部分树」。做法是**按节点过滤**而不是
|
||||
按整棵树放行 —— 否则一条自己参与的线索会把它的**父节点**（可能是别人与别人
|
||||
的对话）连标题一起带出来。
|
||||
|
||||
`UserCanAccessSession` 的口径与既有端点一致：会话 owner，或该用户在会话里有
|
||||
过收发。这是「我参与过」，不是「我建过的」。
|
||||
|
||||
**admin 看全部**（用户 2026-10-04 明确要求分层）：`IsAdmin()` 时跳过逐节点
|
||||
过滤。理由是协作排查往往需要看**别人**的分叉 —— 那正是线索树的用途
|
||||
（「A 交给 B 之后 B 去了哪」跨会话，只看自己那半边永远看不出来）。
|
||||
|
||||
★ 这与 Agent 侧那条边界无关：`AgentMayReadSession`（Agent 只能读自己参与过的
|
||||
会话）是 15e4fe9 / 095213b 两处越权修出来的，**本端点属于人类登录态**，
|
||||
且 admin 全看是用户显式授权的。两者不要混为一谈。
|
||||
|
||||
# parent 链要**剪断**
|
||||
|
||||
一个用户能看见节点 A、看不见它的父 P —— 若照样把 P 的信息发出去就泄露了。
|
||||
所以输出前按可见性**重算 parent**：不可见的祖先一律跳过，树在可见处重新起根。
|
||||
*/
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/agentmail/gateway/internal/middleware"
|
||||
"github.com/agentmail/gateway/internal/repo"
|
||||
)
|
||||
|
||||
func ListSessionTree(w http.ResponseWriter, r *http.Request) {
|
||||
user := middleware.GetUser(r)
|
||||
if user == nil {
|
||||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||||
return
|
||||
}
|
||||
|
||||
all, err := repo.ListSessionTree(r.Context(), nil)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "列会话树失败")
|
||||
return
|
||||
}
|
||||
|
||||
byID := make(map[string]repo.SessionNode, len(all))
|
||||
for _, n := range all {
|
||||
byID[n.SessionID] = n
|
||||
}
|
||||
|
||||
// 可见性：逐节点判；admin 全看。
|
||||
visible := make(map[string]bool, len(all))
|
||||
if user.IsAdmin() {
|
||||
for _, n := range all {
|
||||
visible[n.SessionID] = true
|
||||
}
|
||||
} else {
|
||||
for _, n := range all {
|
||||
ok, err := repo.UserCanAccessSession(r.Context(), user, uuidOf(n.SessionID))
|
||||
if err != nil {
|
||||
// 判据查询失败 ⇒ fail-closed（当不可见），不放行。
|
||||
continue
|
||||
}
|
||||
visible[n.SessionID] = ok
|
||||
}
|
||||
}
|
||||
|
||||
// 剪断不可见的祖先（实现在 repo 层，与判据共用同一份）
|
||||
out := repo.PruneTree(all, visible)
|
||||
|
||||
JSON(w, http.StatusOK, map[string]any{
|
||||
"nodes": out,
|
||||
"admin": user.IsAdmin(),
|
||||
"total": len(all),
|
||||
"visible": len(out),
|
||||
})
|
||||
}
|
||||
|
||||
func uuidOf(s string) uuid.UUID {
|
||||
return uuid.MustParse(s)
|
||||
}
|
||||
89
server/internal/handler/session_tree_test.go
Normal file
89
server/internal/handler/session_tree_test.go
Normal file
@ -0,0 +1,89 @@
|
||||
package handler
|
||||
|
||||
/*
|
||||
会话树的**可见性剪断**判据（2026-10-04）。
|
||||
|
||||
# 为什么这一格最要紧
|
||||
|
||||
树天然会把**父节点**的信息带给子节点。而「我能看见 B」不等于「我能看见
|
||||
B 的父A」—— A 可能是别人与别人的对话。
|
||||
|
||||
所以服务端必须**按节点过滤 + 剪断不可见祖先**。只做前者不做后者，输出里
|
||||
就带着不可见父节点的 session_id 与标题 —— 那是一条真实的泄露路径，
|
||||
而且它藏在「树视图」这个新功能里，不会有人想到去查。
|
||||
|
||||
★ 与 Agent 侧那条边界无关：AgentMayReadSession（Agent 只能读自己参与过的会话）
|
||||
是 15e4fe9 / 095213b 修出来的越权防护；本端点属人类登录态，admin 全看是用户
|
||||
2026-10-04 显式授权的。两者语义不同，不要混谈。
|
||||
*/
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/repo"
|
||||
)
|
||||
|
||||
// A → B → C，用户只参与 B 与 C（A 是别人与别人的对话）。
|
||||
func TestPruneTreeHidesInvisibleAncestor(t *testing.T) {
|
||||
nodes := []repo.SessionNode{
|
||||
{SessionID: "A", Alias: "别人的私密线索", Depth: 0},
|
||||
{SessionID: "B", Alias: "我参与的", ParentSessionID: "A", Depth: 1},
|
||||
{SessionID: "C", Alias: "我参与的续线", ParentSessionID: "B", Depth: 2},
|
||||
}
|
||||
visible := map[string]bool{"B": true, "C": true}
|
||||
|
||||
out := repo.PruneTree(nodes, visible)
|
||||
if len(out) != 2 {
|
||||
t.Fatalf("只应留下可见的 2 个节点，实际 %d：%+v", len(out), out)
|
||||
}
|
||||
// ★ 初版这里对**每个**输出节点都断言 Depth==0，那是判据自己写错了：
|
||||
// B 重新起根（depth 0），但 C 仍挂在 B 之下（depth 1）—— 两者不同。
|
||||
// 判据红着而代码是对的。（又一次「判据比语义宽/窄」的同族错误。）
|
||||
for _, n := range out {
|
||||
if n.SessionID == "A" {
|
||||
t.Fatal("★ 不可见的 A 不该出现在输出里")
|
||||
}
|
||||
if n.ParentSessionID == "A" {
|
||||
t.Errorf("★ B 的 parent 仍指向不可见的 A ⇒ 泄露了 A 的 session_id")
|
||||
}
|
||||
}
|
||||
depth := map[string]int{}
|
||||
parent := map[string]string{}
|
||||
for _, n := range out {
|
||||
depth[n.SessionID] = n.Depth
|
||||
parent[n.SessionID] = n.ParentSessionID
|
||||
}
|
||||
if depth["B"] != 0 || parent["B"] != "" {
|
||||
t.Errorf("B 的父被剪掉后应重新起根（depth=0, parent 空），实际 depth=%d parent=%q",
|
||||
depth["B"], parent["B"])
|
||||
}
|
||||
// C 仍挂在 **B** 之下（B 可见），但绝不能经由不可见的 A
|
||||
if depth["C"] != 1 || parent["C"] != "B" {
|
||||
t.Errorf("C 应挂在 B 之下（depth=1, parent=B），实际 depth=%d parent=%q",
|
||||
depth["C"], parent["C"])
|
||||
}
|
||||
}
|
||||
|
||||
// 全部可见时不得改动任何东西（admin 走的就是这条路）。
|
||||
func TestPruneTreeKeepsFullyVisibleTree(t *testing.T) {
|
||||
nodes := []repo.SessionNode{
|
||||
{SessionID: "A", Alias: "a", Depth: 0},
|
||||
{SessionID: "B", Alias: "b", ParentSessionID: "A", Depth: 1},
|
||||
{SessionID: "C", Alias: "c", ParentSessionID: "B", Depth: 2},
|
||||
}
|
||||
visible := map[string]bool{"A": true, "B": true, "C": true}
|
||||
|
||||
out := repo.PruneTree(nodes, visible)
|
||||
if len(out) != 3 {
|
||||
t.Fatalf("全部可见时应保留 3 个节点，实际 %d", len(out))
|
||||
}
|
||||
for _, n := range out {
|
||||
want := map[string]int{"A": 0, "B": 1, "C": 2}[n.SessionID]
|
||||
if n.Depth != want {
|
||||
t.Errorf("%s depth 应为 %d，实际 %d", n.SessionID, want, n.Depth)
|
||||
}
|
||||
}
|
||||
if out[1].ParentSessionID != "A" {
|
||||
t.Errorf("全部可见时 B 的 parent 不该被清空，实际 %q", out[1].ParentSessionID)
|
||||
}
|
||||
}
|
||||
258
server/internal/repo/session_tree.go
Normal file
258
server/internal/repo/session_tree.go
Normal file
@ -0,0 +1,258 @@
|
||||
package repo
|
||||
|
||||
/*
|
||||
会话级线索树（2026-10-04）。
|
||||
|
||||
# 为什么需要它
|
||||
|
||||
邮件级父子（`mails.parent_mail_id`）早已有，`ThreadView` 也早在用 `depth`
|
||||
渲染**会话内**的树 —— 但**跨会话**的分叉完全不可见。生产库里实测有 **6 封**邮件
|
||||
的 parent 指向**另一条会话**，界面上它们是几行互不相干的东西：
|
||||
|
||||
deploy-pi-bridge-authz-wakeup → 邮件驱动…项目概述
|
||||
harmony-emu-unblock → 邮件驱动…项目概述
|
||||
邮件驱动…项目概述 → 核实-HomeAgent-mail-bridge
|
||||
邮件驱动…项目概述-2 → 时间显示自检
|
||||
本机-agent-能力盘点 → 渲染自检
|
||||
Re-第二个请求… → 邮件驱动…项目概述-2
|
||||
|
||||
而「A 把这件事交给 B 之后 B 继续推」正是协作里最常见的形状。回填后它们
|
||||
连成一条真链（`ba9c194b` 自己也是子节点、且有两个子节点）。
|
||||
|
||||
# 回填口径：只认「确实跨会话分叉」
|
||||
|
||||
`BackfillSessionParents` 只在满足**全部**条件时才写 `parent_session_id`：
|
||||
|
||||
child.session_id ≠ parent.session_id （跨会话）
|
||||
parent 是 child 的**直接**父邮件所在会话 （直接，不沿链上溯）
|
||||
child 在库里**还没有** parent_session_id （不回填覆盖人工设置）
|
||||
|
||||
★ 为什么不做「沿 parent 链上溯到根会话」：那样会把「A→B→C 三跳」压成
|
||||
「A 是 C 的祖先」，而中间那条 B 线索恰恰是独立的协作对象（可能参与方不同）。
|
||||
把链压平会让拓扑失真 —— 比没有树更糟。
|
||||
|
||||
# 环防护
|
||||
|
||||
`parent_session_id` 是自引用，加不加约束都可能被人填出环（A→B→A）。而
|
||||
`ListSessionTree` 用递归 CTE，环会让它**无限递归**。故加两条防护：
|
||||
|
||||
1. 回填只写「parent 的父为空 / parent 本身不是 child 的后代」——写入侧防
|
||||
2. 读取侧带 `depth < 64` 上限 —— **兜底**：即便库里已有环，也不会挂死请求
|
||||
|
||||
第2 条不是可有可无的：环一旦存在是**数据问题**，读取侧必须能活下来。
|
||||
*/
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
)
|
||||
|
||||
// SessionNode 是线索树上的一个会话节点。
|
||||
type SessionNode struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Alias string `json:"alias"`
|
||||
Subject string `json:"subject"`
|
||||
FromAgent string `json:"from_agent"`
|
||||
Workspace string `json:"workspace"`
|
||||
// ParentSessionID 是线索树上的父节点。为空 = 根。
|
||||
//
|
||||
// ★ 服务端必须按可见性**重算**这个字段再输出：用户能看见子节点却看不见
|
||||
// 父节点时，直接返回原值就等于把不可见的父（可能是别人与别人的对话）
|
||||
// 的 session_id 一起泄露。
|
||||
ParentSessionID string `json:"parent_session_id,omitempty"`
|
||||
Depth int `json:"depth"`
|
||||
ChildCnt int `json:"child_count"`
|
||||
MailCnt int `json:"mail_count"`
|
||||
UpdatedAt *time.Time `json:"updated_at,omitempty"`
|
||||
}
|
||||
|
||||
// maxTreeDepth 是读取侧的硬上限（防环）。
|
||||
const maxTreeDepth = 64
|
||||
|
||||
// BackfillSessionParents 按跨会话分叉回填 parent_session_id。
|
||||
//
|
||||
// 幂等：已有 parent_session_id 的会话不动（人工设置优先于自动推断）。
|
||||
// 返回实际写入的行数。
|
||||
func BackfillSessionParents(ctx context.Context) (int64, error) {
|
||||
if _, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE sessions
|
||||
SET parent_session_id = (
|
||||
SELECT ps.session_id
|
||||
FROM mails m
|
||||
JOIN mails pm ON m.parent_mail_id = pm.mail_id
|
||||
JOIN sessions ps ON ps.session_id = pm.session_id
|
||||
WHERE m.session_id = sessions.session_id
|
||||
AND ps.session_id <> sessions.session_id
|
||||
ORDER BY m.created_at ASC
|
||||
LIMIT 1
|
||||
)
|
||||
WHERE parent_session_id IS NULL
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM mails m
|
||||
WHERE m.session_id = sessions.session_id
|
||||
AND m.parent_mail_id IS NOT NULL
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM mails pm
|
||||
WHERE pm.mail_id = m.parent_mail_id
|
||||
AND pm.session_id <> sessions.session_id
|
||||
)
|
||||
)
|
||||
`); err != nil {
|
||||
return 0, fmt.Errorf("回填会话父子: %w", err)
|
||||
}
|
||||
var n int64
|
||||
if err := db.DB.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM sessions WHERE parent_session_id IS NOT NULL`).Scan(&n); err != nil {
|
||||
return 0, fmt.Errorf("统计回填结果: %w", err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// ListSessionTree 返回这些会话构成的**树**（按最近活动倒序，父先于子）。
|
||||
//
|
||||
// parentIDs 为空时返回全部会话的**根节点**（parent 为空的那些）；给了就返回
|
||||
// 这些节点的子树。
|
||||
//
|
||||
// ★ depth 上限是**防环兜底**，不是装饰：`parent_session_id` 是自引用，
|
||||
// 库里一旦有环，递归 CTE 会无限膨胀直到把进程拖垮。
|
||||
func ListSessionTree(ctx context.Context, parentIDs []string) ([]SessionNode, error) {
|
||||
// 根条件：parentIDs 为空时找**全部根节点**（parent 为空、或指向已不存在的会话）。
|
||||
//
|
||||
// ★ 这里**不能**写成 `session_id IN (NULL)` —— 那个表达式恒不匹配任何行。
|
||||
// 实测踩过：那样写时 parentIDs 为空这条分支直接返回 0 个节点。
|
||||
rootCond := `s.parent_session_id IS NULL
|
||||
OR s.parent_session_id = ''
|
||||
OR s.parent_session_id NOT IN (SELECT session_id FROM sessions)`
|
||||
args := []any{maxTreeDepth}
|
||||
if len(parentIDs) > 0 {
|
||||
placeholders := make([]string, 0, len(parentIDs))
|
||||
for _, id := range parentIDs {
|
||||
placeholders = append(placeholders, "?")
|
||||
args = append(args, id)
|
||||
}
|
||||
rootCond = `s.session_id IN (` + strings.Join(placeholders, ",") + `)`
|
||||
}
|
||||
|
||||
// CTE 列名用 lvl 而不是 depth —— depth 会与 SELECT 里的 t.depth 撞名。
|
||||
// 既有 DescendantsRaw 也是 down(mail_id, lvl)，照它走。
|
||||
//
|
||||
// ★ 计数**不能**写成相关子查询：环下递归会展开 maxTreeDepth 层，每个节点
|
||||
// 都再跑一次「按 session_id 数 mails」—— 实测 2 节点环直接打满 55s 超时。
|
||||
// 改成 LEFT JOIN 两个聚合，各自只算一次。
|
||||
//
|
||||
// ★ 参数顺序：`t.lvl < ?` 是 SQL 里的第一个占位符（$1），parentIDs 随后（$2…）。
|
||||
// 反过来 ⇒ 根查询拿到一个整数、匹配不到任何行，树只返回递归分支那半。
|
||||
q := `
|
||||
WITH RECURSIVE tree(session_id, lvl) AS (
|
||||
SELECT s.session_id, 0
|
||||
FROM sessions s
|
||||
WHERE ` + rootCond + `
|
||||
UNION ALL
|
||||
SELECT c.session_id, t.lvl + 1
|
||||
FROM sessions c JOIN tree t ON c.parent_session_id = t.session_id
|
||||
WHERE t.lvl < ?
|
||||
)
|
||||
SELECT s.session_id, s.session_alias, s.subject, s.from_agent,
|
||||
COALESCE(s.parent_session_id, ''), s.workspace, t.lvl,
|
||||
COALESCE(kids.n, 0), COALESCE(mc.n, 0),
|
||||
s.updated_at
|
||||
FROM tree t
|
||||
JOIN sessions s ON s.session_id = t.session_id
|
||||
LEFT JOIN (SELECT parent_session_id AS sid, COUNT(*) AS n
|
||||
FROM sessions WHERE parent_session_id IS NOT NULL
|
||||
GROUP BY parent_session_id) kids ON kids.sid = s.session_id
|
||||
LEFT JOIN (SELECT session_id AS sid, COUNT(*) AS n FROM mails GROUP BY session_id) mc
|
||||
ON mc.sid = s.session_id
|
||||
ORDER BY s.updated_at DESC
|
||||
`
|
||||
|
||||
rows, err := db.DB.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("列会话树: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []SessionNode{}
|
||||
for rows.Next() {
|
||||
var n SessionNode
|
||||
var alias, subject sql.NullString
|
||||
if err := rows.Scan(&n.SessionID, &alias, &subject, &n.FromAgent,
|
||||
&n.ParentSessionID, &n.Workspace, &n.Depth, &n.ChildCnt, &n.MailCnt,
|
||||
&n.UpdatedAt); err != nil {
|
||||
return nil, fmt.Errorf("读会话树行: %w", err)
|
||||
}
|
||||
n.Alias = alias.String
|
||||
n.Subject = subject.String
|
||||
if n.Alias == "" {
|
||||
n.Alias = n.Subject
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// PruneTree 按可见性剪断树：不可见的节点直接丢弃，**可见节点若其父不可见则重新起根**
|
||||
// （depth 归零、parent 清空）。
|
||||
//
|
||||
// ★ 为什么这是安全边界而不是展示逻辑：树天然会把父节点带给子节点，
|
||||
//
|
||||
// 而「我能看见 B」不等于「我能看见 B 的父 A」（A 可能是别人与别人的对话）。
|
||||
// 只过滤节点、不重算 parent，输出里就带着不可见父的 session_id 与标题 ——
|
||||
// 一条真实的泄露路径，而且它藏在「树视图」这个新功能里。
|
||||
//
|
||||
// 放在 repo 层而不是 handler：可见性判断与剪枝必须和查询同一层，
|
||||
// 否则 handler 里那份内联实现会与判据里那份悄悄漂移（判据测 A、线上跑 B）。
|
||||
func PruneTree(nodes []SessionNode, visible map[string]bool) []SessionNode {
|
||||
byID := make(map[string]SessionNode, len(nodes))
|
||||
for _, n := range nodes {
|
||||
byID[n.SessionID] = n
|
||||
}
|
||||
|
||||
// 祖先链：从自己往上走，直到根或遇到不可见的节点。
|
||||
// 只要**每个**祖先都可见，这一段层级全部保留；一旦某个不可见，
|
||||
// 就**在它下面重新起根**（那个不可见的节点本身不输出）。
|
||||
ancestors := func(id string) (chain []string, cutAtInvisible bool) {
|
||||
cur := id
|
||||
for cur != "" {
|
||||
n, ok := byID[cur]
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
parent := n.ParentSessionID
|
||||
if parent == "" {
|
||||
return chain, false // 到根了
|
||||
}
|
||||
if !visible[parent] {
|
||||
return chain, true // ★ 父不可见 ⇒ 从当前节点起重新起根
|
||||
}
|
||||
chain = append([]string{parent}, chain...)
|
||||
cur = parent
|
||||
}
|
||||
return chain, true
|
||||
}
|
||||
|
||||
out := make([]SessionNode, 0, len(nodes))
|
||||
for _, n := range nodes {
|
||||
if !visible[n.SessionID] {
|
||||
continue
|
||||
}
|
||||
chain, _ := ancestors(n.SessionID)
|
||||
n.Depth = len(chain)
|
||||
if len(chain) == 0 {
|
||||
// 自己是根（链上没有任何**可见**祖先）。
|
||||
n.ParentSessionID = ""
|
||||
} else {
|
||||
// 挂到链上**最深的那个可见祖先**上。
|
||||
// （cutAtInvisible 时 chain 首元素就是最近的那个可见祖先，
|
||||
// 而不是不可见的那个 —— 不可见的绝不写进输出。）
|
||||
n.ParentSessionID = chain[len(chain)-1]
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
253
server/internal/repo/session_tree_test.go
Normal file
253
server/internal/repo/session_tree_test.go
Normal file
@ -0,0 +1,253 @@
|
||||
package repo
|
||||
|
||||
/*
|
||||
会话级线索树的判据（2026-10-04）。
|
||||
|
||||
# 这一格最要紧的是**环**
|
||||
|
||||
`parent_session_id` 是自引用字段。加不加 FK 约束都可能被人填出环（A→B→A），
|
||||
而读取走递归 CTE —— 环会让它**无限膨胀直到拖垮进程**。所以：
|
||||
|
||||
· 写入侧：回填只认「直接父邮件所在的会话」，且不回填已有值的会话
|
||||
· 读取侧：**必须**带 depth 上限（maxTreeDepth）
|
||||
|
||||
第 2 条是兜底，不是装饰：环是**数据问题**，而数据问题必须在读取侧活下来，
|
||||
否则一次误操作会变成「整个端点挂死」。
|
||||
|
||||
# 为什么不做「沿链上溯到根」
|
||||
|
||||
那样会把 A→B→C 三跳压成「A 是 C 的祖先」，而中间那条 B 线索恰恰是独立的
|
||||
协作对象（参与方可能不同）。压平会让拓扑失真 —— 比没有树更糟。
|
||||
所以只认直接跨会话分叉。
|
||||
*/
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
)
|
||||
|
||||
func setupTreeDB(t *testing.T) {
|
||||
t.Helper()
|
||||
db.Close()
|
||||
if err := db.Connect(context.Background(), "sqlite://"+t.TempDir()+"/tree.db"); err != nil {
|
||||
t.Fatalf("连接测试库: %v", err)
|
||||
}
|
||||
if err := db.Migrate(context.Background()); err != nil {
|
||||
t.Fatalf("迁移: %v", err)
|
||||
}
|
||||
t.Cleanup(db.Close)
|
||||
}
|
||||
|
||||
func mkSession(t *testing.T, alias, subject string) string {
|
||||
t.Helper()
|
||||
var id string
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`INSERT INTO sessions (session_alias, subject, status, workspace, from_agent)
|
||||
VALUES ($1,$2,'active','/tmp','pi') RETURNING session_id`, alias, subject).Scan(&id); err != nil {
|
||||
t.Fatalf("建会话 %s: %v", alias, err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func mkMail(t *testing.T, sid, subject, parentID string) string {
|
||||
t.Helper()
|
||||
var p any
|
||||
if parentID != "" {
|
||||
p = parentID
|
||||
}
|
||||
var id string
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`INSERT INTO mails (mail_id, session_id, from_name, to_workspace, to_name, subject, body, status, parent_mail_id)
|
||||
VALUES (gen_random_uuid(),$1,'pi','/tmp','dsh',$2,'x','unread',$3) RETURNING mail_id`,
|
||||
sid, subject, p).Scan(&id); err != nil {
|
||||
t.Fatalf("建邮件 %s: %v", subject, err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
// ★ 核心：跨会话分叉要连成树。
|
||||
|
||||
func TestBackfillLinksCrossSessionForks(t *testing.T) {
|
||||
setupTreeDB(t)
|
||||
parent := mkSession(t, "parent-line", "原始线索")
|
||||
child := mkSession(t, "child-line", "接手继续")
|
||||
|
||||
pid := mkMail(t, parent, "交给 B", "")
|
||||
mkMail(t, child, "B 接手", pid)
|
||||
|
||||
n, err := BackfillSessionParents(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("回填失败: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Errorf("应回填 1 条 parent_session_id，实际 %d", n)
|
||||
}
|
||||
|
||||
nodes, err := ListSessionTree(context.Background(), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("列树失败: %v", err)
|
||||
}
|
||||
if len(nodes) != 2 {
|
||||
t.Fatalf("树应有 2 个节点，实际 %d：%+v", len(nodes), nodes)
|
||||
}
|
||||
// 父 depth=0 子 depth=1，且父的 child_count=1
|
||||
var parentNode, childNode *SessionNode
|
||||
for i := range nodes {
|
||||
if nodes[i].SessionID == parent {
|
||||
parentNode = &nodes[i]
|
||||
} else if nodes[i].SessionID == child {
|
||||
childNode = &nodes[i]
|
||||
}
|
||||
}
|
||||
if parentNode == nil || childNode == nil {
|
||||
t.Fatalf("两个节点都该在树里：%+v", nodes)
|
||||
}
|
||||
if childNode.Depth != 1 {
|
||||
t.Errorf("子节点 depth 应为 1（缩进一级），实际 %d", childNode.Depth)
|
||||
}
|
||||
if parentNode.ChildCnt != 1 {
|
||||
t.Errorf("父节点 child_count 应为 1，实际 %d", parentNode.ChildCnt)
|
||||
}
|
||||
}
|
||||
|
||||
// ★ 环不能挂死请求：库里已有环时，读取必须仍能返回。
|
||||
|
||||
func TestListSessionTreeSurvivesCycle(t *testing.T) {
|
||||
setupTreeDB(t)
|
||||
a := mkSession(t, "a", "A")
|
||||
b := mkSession(t, "b", "B")
|
||||
// 人为造环：A→B→A
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET parent_session_id = ? WHERE session_id = ?`, b, a); err != nil {
|
||||
t.Fatalf("设 A 的父: %v", err)
|
||||
}
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET parent_session_id = ? WHERE session_id = ?`, a, b); err != nil {
|
||||
t.Fatalf("设 B 的父: %v", err)
|
||||
}
|
||||
|
||||
// 有 depth 上限 ⇒ 这条递归会在 maxTreeDepth 处停下并返回，不会无限增长。
|
||||
//
|
||||
// ★ 这格真正要防的是「挂死」：初版把计数写成相关子查询，环下每个节点
|
||||
// 都重跑一次 mails 计数 —— 实测 2 节点环打满 **55s 超时**。
|
||||
// 改成 LEFT JOIN 聚合后是 0.01s 级。
|
||||
//
|
||||
// 注意环节点**不会**出现在结果里（根条件要求 parent 指向已不存在的会话，
|
||||
// 而环上的 parent 真实存在 ⇒ 起点那条被排除，返回 0 个节点）。
|
||||
// 那是**符合预期的**：环是坏数据，不该出现在树里。
|
||||
// 但「被排除」不等于「不会打挂」—— 递归仍会跑，所以这格守的是后者。
|
||||
st := time.Now()
|
||||
nodes, err := ListSessionTree(context.Background(), []string{a})
|
||||
elapsed := time.Since(st)
|
||||
if err != nil {
|
||||
t.Fatalf("★ 环不应让读取失败/挂死，实际报错: %v", err)
|
||||
}
|
||||
if elapsed > 5*time.Second {
|
||||
t.Errorf("★ 环下查询耗时 %v（>5s）⇒ 防环/性能兜底失效", elapsed)
|
||||
}
|
||||
for _, n := range nodes {
|
||||
if n.Depth > maxTreeDepth {
|
||||
t.Fatalf("★ depth 超过上限 %d ⇒ 防环兜底失效", maxTreeDepth)
|
||||
}
|
||||
}
|
||||
t.Logf("环下返回 %d 个节点（最大 depth=%d）", len(nodes), maxDepthOf(nodes))
|
||||
}
|
||||
|
||||
func maxDepthOf(ns []SessionNode) int {
|
||||
m := 0
|
||||
for _, n := range ns {
|
||||
if n.Depth > m {
|
||||
m = n.Depth
|
||||
}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// ★ 回填不得覆盖人工设置的父。
|
||||
|
||||
func TestBackfillDoesNotOverwriteManual(t *testing.T) {
|
||||
setupTreeDB(t)
|
||||
parent := mkSession(t, "p", "父")
|
||||
child := mkSession(t, "c", "子")
|
||||
manual := mkSession(t, "manual-parent", "人工指定的父")
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET parent_session_id = ? WHERE session_id = ?`, manual, child); err != nil {
|
||||
t.Fatalf("设人工父: %v", err)
|
||||
}
|
||||
// 再造一封真实分叉邮件（回填会想写它）
|
||||
pid := mkMail(t, parent, "真分叉", "")
|
||||
mkMail(t, child, "子里的回信", pid)
|
||||
|
||||
if _, err := BackfillSessionParents(context.Background()); err != nil {
|
||||
t.Fatalf("回填失败: %v", err)
|
||||
}
|
||||
var got string
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`SELECT parent_session_id FROM sessions WHERE session_id = ?`, child).Scan(&got); err != nil {
|
||||
t.Fatalf("读父: %v", err)
|
||||
}
|
||||
if got != manual {
|
||||
t.Errorf("★ 回填覆盖了人工设置的父：%s ⇒ 应为 %s", got, manual)
|
||||
}
|
||||
}
|
||||
|
||||
// ★ 同会话内的邮件父子**不该**产生会话父子（那会让会话自己指向自己）。
|
||||
|
||||
func TestBackfillIgnoresIntraSessionMail(t *testing.T) {
|
||||
setupTreeDB(t)
|
||||
sid := mkSession(t, "same", "同会话内续谈")
|
||||
a := mkMail(t, sid, "第一封", "")
|
||||
mkMail(t, sid, "第二封", a)
|
||||
|
||||
n, err := BackfillSessionParents(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("回填失败: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Errorf("同会话内的邮件父子不该建会话父子，实际回填 %d 条", n)
|
||||
}
|
||||
}
|
||||
|
||||
// ★ 一次性守卫：回填**只能跑一次**。
|
||||
//
|
||||
// 为什么这格要紧：回填改的是会话**拓扑**（parent_session_id），不是派生数据。
|
||||
// 若每次 Migrate 都跑，那么「某人手工把 parent 改对/改空」这个意图会在每次
|
||||
// 重启时被悄悄改回去 —— 与既有 backfillMailReads 踩过的坑同族（它的注释写着：
|
||||
// 每次跑会把「某抄送方读过」按主收件人写成已读，正是那次要修的语义错误）。
|
||||
//
|
||||
// 判据形状：回填后再手工改一个 parent，**再次 Migrate**，那个改动必须还在。
|
||||
func TestBackfillSessionParentsRunsOnlyOnce(t *testing.T) {
|
||||
setupTreeDB(t)
|
||||
parent := mkSession(t, "p", "父")
|
||||
child := mkSession(t, "c", "子")
|
||||
pid := mkMail(t, parent, "分叉", "")
|
||||
mkMail(t, child, "子里的回信", pid)
|
||||
|
||||
if _, err := BackfillSessionParents(context.Background()); err != nil {
|
||||
t.Fatalf("首次回填: %v", err)
|
||||
}
|
||||
|
||||
// 手工改一个**不同于自动推断**的值（自动会写 parent，这里写成手工的别的会话）
|
||||
manual := mkSession(t, "manual", "人工指定的父")
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET parent_session_id = ? WHERE session_id = ?`, manual, child); err != nil {
|
||||
t.Fatalf("设人工父: %v", err)
|
||||
}
|
||||
|
||||
// 第二次回填（模拟重启后 Migrate 又跑）
|
||||
if _, err := BackfillSessionParents(context.Background()); err != nil {
|
||||
t.Fatalf("二次回填: %v", err)
|
||||
}
|
||||
|
||||
var got string
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`SELECT parent_session_id FROM sessions WHERE session_id = ?`, child).Scan(&got); err != nil {
|
||||
t.Fatalf("读父: %v", err)
|
||||
}
|
||||
if got != manual {
|
||||
t.Errorf("★ 二次回填覆盖了人工设置：%s ⇒ 应为 %s", got, manual)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user