package repo import ( "context" "database/sql" "errors" "fmt" "strings" "time" "github.com/agentmail/gateway/internal/db" "github.com/agentmail/gateway/internal/models" "github.com/google/uuid" ) // ---------- 平台会话镜像 ---------- // // Agent 平台自己也在开会话:有些经由邮件驱动,有些是人直接在平台界面上开的。 // 写信时想续谈某条会话,得先知道那个工作区下有哪些会话可续 —— 而 Gateway // 只看得见邮件驱动的那部分。 // // **由插件在心跳里上报,Gateway 不反向拉取。** // 当前架构是单向的(Agent 持密钥主动连 Gateway,Gateway 从不外呼); // 让 Gateway 去调平台接口需要它保存各平台的地址与凭证,那是另一套信任模型。 // 代价是插件没运行时同步不了 —— 但插件没运行时邮件本来也投不进去。 // PlatformSession 是插件上报的一条平台侧会话。 type PlatformSession struct { PlatformID string `json:"platform_id"` Workspace string `json:"workspace"` Slug string `json:"slug,omitempty"` Title string `json:"title,omitempty"` MailDriven bool `json:"mail_driven"` UpdatedAt *time.Time `json:"updated_at,omitempty"` } // maxPlatformSessions 限制单次上报的会话数。 // // 一个长期运行的平台可以累积上千条会话,而候选列表上千项对人没有意义。 // 插件按最近活跃排序后上报前 N 条即可。 const maxPlatformSessions = 200 /* ReplacePlatformSessions 替换某 Agent 在**本次上报所覆盖的那些工作区**里的镜像。 # 「整表替换」的域是 (agent, workspace),不是 agent(2026-09-26 修) 原先的 DELETE 域是 `agent_name` 单列 —— 而**每个上报者只知道自己一个 directory**: plugins/opencode-mail-bridge/index.js:1147 client.session.list({ query: directory ? {directory} : undefined }) 于是 A 工作区的桥上报一次,就把 B 工作区上报过的镜像全擦掉;下个工作区的桥 再上报,又擦掉 A 的。表现为「镜像按 project 轮换」。 # 生产实测(不是推断) sqlite3 agent_platform_sessions 按 workspace 分组: dsh 77 条散在 **25** 个工作区(/home/program/agentmail 25、/tmp 20 …) pi 151 条散在 **62** 个工作区 # 后果已在生产数据上可见 镜像被擦 ⇒ `notify/mail.go` 的 `PlatformSessionFor` 查不到 ⇒ `sessions.platform_id` 留空 ⇒ 收方拿不到平台会话 id。 实测 **18 条活跃会话里 17 条 `platform_id` 为空**。 而空 platform_id 的去向不止"少一个跳转":`notify/mail.go:94` 用它决定 `platform_session_id` 发给谁,owner 取错就抛「平台侧会话已删」⇒ **邮件静默消失**。 # 为什么仍然是"整表替换"而不是增量合并 镜像是平台当前状态的快照。增量合并会让已删掉的平台会话永远留在候选列表里, 而 session 位是三态语义,指向不存在的会话会直接 404("选了却送不到")。 ⇒ 保持整表替换,只把**域收窄到本次上报覆盖的工作区**。 # 上报跨多个工作区时的语义 一次上报里出现多个 workspace ⇒ 那些工作区**各自**整表替换; **本次没出现的**工作区一律不动。 # 域里含 workspace 的必要性 `agent_platform_sessions` 的主键是 `(agent_name, platform_id)`。 同一个 platform_id 出现在两个 workspace 会撞 `UNIQUE constraint failed`, 而这里没有 `ON CONFLICT` + `defer tx.Rollback()` ⇒ **整个 DELETE 回滚**、 `agents.go` 降级为 -1 ⇒ 表现为「心跳一直成功而镜像永久停滞」。 ⇒ 该主键是否也要加 workspace,见 DEBTS.json 的 platform-mirror-replace-domain-too-wide (**那一项未决**:本函数只消除了"擦错别人",没有消除"同 id 跨 ws 撞约束")。 */ func ReplacePlatformSessions(ctx context.Context, agentName string, list []PlatformSession) error { agentName = strings.TrimSpace(agentName) if agentName == "" { return nil } if len(list) > maxPlatformSessions { list = list[:maxPlatformSessions] } // 本次上报覆盖了哪些工作区 —— DELETE 域就按它圈定。 // // 用 map 去重且**保序**:同一工作区在 list 里出现多次只算一次(下面的 // seen 也会按 id 去重,但 DELETE 域要先算出来)。 wsSeen := map[string]bool{} var wsOrder []string for _, ps := range list { ws := strings.TrimSpace(ps.Workspace) if ws != "" && !wsSeen[ws] { wsSeen[ws] = true wsOrder = append(wsOrder, ws) } } tx, err := db.DB.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() // 只清本次上报覆盖的工作区。wsOrder 为空(上报里一项 workspace 都没有) // ⇒ **什么都不删**:那说明这次上报不带任何工作区信息,无从判断该清谁, // 按 agent 清等于回到那个缺陷。 // // `len(wsOrder) > 0` 这个守卫不是可有可无的:空列表会拼出 `IN ()`, // 两种方言都恰好恒假(SQLite 与 PostgreSQL 都是),所以**行为上**与 // "什么都不删"相同 —— 但那是依赖两个数据库的一条隐式巧合,而不是 // 写代码的人读得出来的保证。显式守卫把意图摆在语句上。 if len(wsOrder) > 0 { ph := make([]string, len(wsOrder)) args := make([]any, 0, len(wsOrder)+1) args = append(args, agentName) for i, w := range wsOrder { ph[i] = fmt.Sprintf("$%d", i+2) args = append(args, w) } if _, err := tx.ExecContext(ctx, `DELETE FROM agent_platform_sessions WHERE agent_name = $1 AND workspace IN (`+ strings.Join(ph, ",")+`)`, args...); err != nil { return err } } seen := map[string]bool{} for _, ps := range list { id := strings.TrimSpace(ps.PlatformID) if id == "" || seen[id] { continue } seen[id] = true driven := 0 if ps.MailDriven { driven = 1 } if _, err := tx.ExecContext(ctx, ` INSERT INTO agent_platform_sessions (agent_name, platform_id, workspace, slug, title, mail_driven, updated_at, reported_at) VALUES ($1, $2, $3, $4, $5, $6, $7, NOW()) `, agentName, id, strings.TrimSpace(ps.Workspace), strings.TrimSpace(ps.Slug), strings.TrimSpace(ps.Title), driven, ps.UpdatedAt); err != nil { return err } } return tx.Commit() } // SessionCandidate 是「续谈某条会话」的一个候选项。 type SessionCandidate struct { // Alias 是填进 session 位的值 —— 候选项的实际用途就是它 Alias string `json:"alias"` // Title 给人看,用来分辨两条别名相似的会话在谈什么 Title string `json:"title,omitempty"` // Source 说明这条候选从哪来: // mail 本侧邮件线索(可直接送达) // platform 平台侧会话镜像(本侧还没有对应线索) Source string `json:"source"` // Unread 仅 mail 来源有意义 Unread int `json:"unread,omitempty"` } // SuggestSessionCandidates 汇总某 name@path 下可续谈的会话。 // // 两个来源合并: // 1. 本侧邮件线索(sessions.workspace 匹配,或历史数据里靠 mails 反推) // 2. 平台会话镜像里带 slug 的那些 // // 本侧优先:邮件线索是「这个别名一定送得到」的保证,而镜像只是平台的说法。 // 同名时保留本侧那条,并把镜像的标题补上去(镜像通常有更新的标题)。 func SuggestSessionCandidates(ctx context.Context, forUser, peerName, path string) ([]SessionCandidate, error) { return suggestSessionCandidates(ctx, forUser, peerName, path, nil) } func suggestSessionCandidates(ctx context.Context, forUser, peerName, path string, onlyWorkspace *string) ([]SessionCandidate, error) { out := []SessionCandidate{} seen := map[string]int{} // alias -> out 下标 // 工作区收窄只在本侧线索那条查询里能加:`sessions.workspace` 才是会话的工作区, // 镜像表(来源 2)自己那份 workspace 列另算。 wsClause := "" wsArgs := []any{} if onlyWorkspace != nil { wsClause = ` AND COALESCE(s.workspace, '') = $4` wsArgs = append(wsArgs, *onlyWorkspace) } // ---- 来源 1:本侧邮件线索 ---- // // sessions.workspace 是权威来源。它是新加的列,历史会话为空串, // 因此保留 mails 反推作为兜底:`s.workspace = $2 OR (s.workspace = '' AND )`。 // 反推只看 to_workspace —— Agent 回信时 from_workspace 存的是 Agent 名而非路径, // 拿它比路径永远匹配不上。 rows, err := db.DB.QueryContext(ctx, ` SELECT s.session_alias, COALESCE(s.subject, ''), COALESCE(s.platform_id, ''), (SELECT COUNT(*) FROM mails u WHERE u.session_id = s.session_id AND u.status = 'unread') FROM sessions s WHERE s.session_alias IS NOT NULL AND s.session_alias <> '' AND s.status <> 'archived' AND EXISTS ( SELECT 1 FROM mails m WHERE m.session_id = s.session_id AND (m.to_name = $1 OR m.from_name = $1 OR `+db.CCHas("m.cc_list", 1)+`) ) AND ($2 = '' OR s.workspace = $2 OR (s.workspace = '' AND EXISTS ( SELECT 1 FROM mails w WHERE w.session_id = s.session_id AND COALESCE(w.to_workspace,'') = $2 ))) AND ($3 = '' OR s.owner_user_id = (SELECT user_id FROM users WHERE username = $3) OR EXISTS ( SELECT 1 FROM mails mm WHERE mm.session_id = s.session_id AND (mm.from_name = $3 OR mm.to_name = $3 OR `+db.CCHas("mm.cc_list", 3)+`) ))`+wsClause+` ORDER BY s.updated_at DESC `, append([]any{peerName, path, forUser}, wsArgs...)...) if err != nil { return out, err } defer rows.Close() for rows.Next() { var alias, title, pid string var unread int if err := rows.Scan(&alias, &title, &pid, &unread); err != nil { return out, err } if alias == "" { continue } seen[alias] = len(out) out = append(out, SessionCandidate{ Alias: alias, Title: title, Source: "mail", Unread: unread, }) if pid != "" { seen["pid:"+pid] = len(out) - 1 } } if err := rows.Err(); err != nil { return out, err } // ---- 来源 2:平台会话镜像 ---- // // 镜像表有自己的 workspace 列,所以这里另写一个参数位($3), // 不与来源 1 复用 $4 —— 两条查询的参数个数不同,硬凑一个编号只会让 // 下一次改动踩到"占了位子却没人绑"的坑。 mirrorWS := "" mirrorArgs := []any{peerName, path} if onlyWorkspace != nil { mirrorWS = ` AND COALESCE(workspace, '') = $3` mirrorArgs = append(mirrorArgs, *onlyWorkspace) } prows, err := db.DB.QueryContext(ctx, ` SELECT slug, title, platform_id FROM agent_platform_sessions WHERE agent_name = $1 AND slug <> '' AND ($2 = '' OR workspace = $2)`+mirrorWS+` -- 不用 NULLS LAST:它要 SQLite 3.30+,而驱动自带的版本不由我们控制。 -- COALESCE 在两个方言里都成立,语义也更直接:没有 updated_at 就用上报时间。 ORDER BY COALESCE(updated_at, reported_at) DESC `, mirrorArgs...) if err != nil { // 镜像查不到不该让整个补全失败:本侧线索已经够用了 return out, nil } defer prows.Close() for prows.Next() { var slug, title, pid string if err := prows.Scan(&slug, &title, &pid); err != nil { break } if slug == "" { continue } // 已被接管的平台会话不再单独列:选它也会落进已有的那条本侧线索, // 但候选列表出现两次会让人以为有两条不同的会话(项目定位 x2 的场景)。 if pid != "" { if _, dup := seen["pid:"+pid]; dup { continue } } if i, ok := seen[slug]; ok { // 本侧已有同名线索:保留 mail 来源(它保证送得到), // 但补上镜像的标题 —— 平台侧标题通常比会话建立时的主题更贴切 if out[i].Title == "" && title != "" { out[i].Title = title } continue } seen[slug] = len(out) out = append(out, SessionCandidate{Alias: slug, Title: title, Source: "platform"}) } return out, nil } // SetSessionWorkspace 记下会话所属的工作目录。 // // 只在为空时写入:会话的工作区在建立时就定下了,之后不该被一封发往 // 别处的邮件改掉 —— 那会让这条会话在候选列表里凭空换一个工作区。 func SetSessionWorkspace(ctx context.Context, sessionID interface{ String() string }, workspace string) error { ws := strings.TrimSpace(workspace) if ws == "" { return nil } _, err := db.DB.ExecContext(ctx, `UPDATE sessions SET workspace = $1 WHERE session_id = $2 AND workspace = ''`, ws, sessionID.String()) return err } // ---------- 接管平台会话 ---------- // // TUI 与邮箱是同一个 Agent 的**两个入口**,不是两套隔离的世界。 // 人在平台界面上开的会话,应该也能被邮件投进去 —— 补全早就把它们列为候选, // 缺的只是投递侧这一跳。 // // 「接管」= 在本侧建一条会话并把 platform_id 记上。之后: // - 这条会话在 sessions 表里有正式身份(可寻址、有预算、能归档) // - 插件收到投递事件时看到 platform_id,就去 resume 那条平台会话 // 而不是新建一条 // // 一条平台会话只能被接管一次:第二次投递复用第一次建的本侧会话, // 否则同一条 TUI 对话会在邮箱里裂成多条互不相干的线索。 // FindPlatformSession 按 (agent, slug, workspace) 找一条平台会话镜像。 // // workspace 为空表示不限(地址省略 path 位时)。返回 platform_id 与它的 // 真实 workspace —— 后者是权威的:**会话的 cwd 在它创建时就定了**, // 地址里的 path 位若与之不同,以会话为准。人是从候选列表里选的, // 他要的是「那条会话」而不是「那个目录」。 func FindPlatformSession(ctx context.Context, agentName, slug, workspace string) (platformID, realWorkspace, title string, err error) { agentName = strings.TrimSpace(agentName) slug = strings.TrimSpace(slug) if agentName == "" || slug == "" { return "", "", "", ErrSessionNotFound } ws := strings.TrimSpace(workspace) err = db.DB.QueryRowContext(ctx, ` SELECT platform_id, workspace, title FROM agent_platform_sessions WHERE agent_name = $1 AND slug = $2 AND ($3 = '' OR workspace = $3) ORDER BY COALESCE(updated_at, reported_at) DESC LIMIT 1 `, agentName, slug, ws).Scan(&platformID, &realWorkspace, &title) if errors.Is(err, sql.ErrNoRows) { return "", "", "", ErrSessionNotFound } return platformID, realWorkspace, title, err } // FindSessionByPlatformID 找出已经接管了某条平台会话的本侧会话。 // // 返回 ErrSessionNotFound 表示还没被接管。归档的也算 —— 让归档过的会话 // 重新被接管会造出第二条本侧会话,同一条 TUI 对话在邮箱里就裂成两截。 // 需要恢复的话人应该去取消归档。 func FindSessionByPlatformID(ctx context.Context, agentName, platformID string) (uuid.UUID, error) { var id uuid.UUID err := db.DB.QueryRowContext(ctx, ` SELECT s.session_id FROM sessions s WHERE s.platform_id = $1 AND s.status <> 'archived' AND EXISTS ( SELECT 1 FROM mails m WHERE m.session_id = s.session_id AND (m.to_name = $2 OR m.from_name = $2 OR `+db.CCHas("m.cc_list", 2)+`) ) ORDER BY s.updated_at DESC LIMIT 1 `, platformID, agentName).Scan(&id) if errors.Is(err, sql.ErrNoRows) { return uuid.Nil, ErrSessionNotFound } return id, err } // AdoptPlatformSession 接管一条平台会话:建本侧会话并绑定 platform_id。 // // alias 用平台自己的 slug —— 「别名复用平台命名」是既定决策,而且人在补全里 // 看到的就是那个 slug,投递后别名换成别的会让他找不到自己刚发的信。 // // workspace 用平台会话的真实 cwd 而不是地址里的 path 位,理由见 // FindPlatformSession 的注释。 func AdoptPlatformSession(ctx context.Context, agentName, platformID, slug, workspace, subject string) (uuid.UUID, error) { // slug 可能与本侧某条无关会话撞名(别名全局唯一)。撞了就加后缀 —— // EnsureSessionAlias 已有这套逻辑,这里先建后命名即可。 id, err := CreateSession(ctx, nil, agentName, subject, workspace) if err != nil { return uuid.Nil, err } if _, err := db.DB.ExecContext(ctx, `UPDATE sessions SET platform_id = $1 WHERE session_id = $2`, platformID, id); err != nil { return uuid.Nil, err } // 显式写入档位与强制力:接管一条平台会话没有父会话, // 只靠 DB 默认值会在「schema 列定义变动」或「迁移补列给了不同默认」时 // 静默偏离预期 —— 显式写 'workspace' 是唯一可靠表述「这条会话是新接管的, // 没有继承来源」的方式。与 me.go 新建会话那条路径一致。 if _, err := SetSessionPermissionMode(ctx, id, models.DefaultPermissionMode); err != nil { return uuid.Nil, err } _ = SetSessionEnforcement(ctx, id, AgentModeEnforcement(ctx, agentName)) // 别名尽量用 slug;撞名时 EnsureSessionAlias 自动加后缀 _, _ = EnsureSessionAlias(ctx, id, slug) return id, nil } // PlatformIDOf 读一条本侧会话绑定的平台会话 id(空 = 不是接管来的)。 // // 投递时要把它放进 SSE 事件:插件据此决定 resume 还是新建。 // // 只在「已知这条会话只有一个参与方」时用它。有抄送时必须用 // PlatformSessionFor 拿到归属方 —— 理由见那个函数。 func PlatformIDOf(ctx context.Context, sessionID uuid.UUID) string { pid, _ := PlatformSessionFor(ctx, sessionID) return pid } // PlatformSessionFor 返回一条本侧会话绑定的平台会话 id **及其归属 Agent**。 // // # 为什么归属方是必须的 // // `platform_id` 是**会话级**的一个值,而一封邮件可以有多个参与方。 // 把它无差别推给所有人,收到的一方会拿它去自己的磁盘上找会话文件 —— // 那个 id 属于别的平台。 // // 生产实测:会话 `16845133` 接管了 pi 的会话 `01a05a5e-…`,而那封邮件抄送了 // `dsh@/home/program/agentmail.new`。DSH 收到同一个 platform_session_id, // 在 `~/.dsh/sessions/` 里查不到(那是 `/root/.pi/agent/sessions/` 下的文件), // 于是走进「平台侧会话已删」那条防线抛错。那道防线本身是对的(N-8: // 不能退回新建,否则人在界面上看不到这封邮件带来的对话),它拦下的却是 // 「别人的会话」—— 邮件因此静默消失,而插件侧的日志走的是不进 journalctl // 的通道,连线索都没有。 // // 归属方以镜像(`agent_platform_sessions.agent_name`,Agent 自己上报的)为准; // 镜像整表替换,平台侧删了会话那行就没了,此时退回 `sessions.from_agent` —— // `AdoptPlatformSession` 建会话时把归属 Agent 写在那里,是可靠的第二来源。 func PlatformSessionFor(ctx context.Context, sessionID uuid.UUID) (platformID, owner string) { var pid, fromAgent string var mirrored *string if err := db.DB.QueryRowContext(ctx, ` SELECT COALESCE(s.platform_id, ''), COALESCE(s.from_agent, ''), aps.agent_name FROM sessions s LEFT JOIN agent_platform_sessions aps ON aps.platform_id = s.platform_id AND COALESCE(s.platform_id, '') <> '' WHERE s.session_id = $1`, sessionID).Scan(&pid, &fromAgent, &mirrored); err != nil { return "", "" } if pid == "" { return "", "" } if mirrored != nil && *mirrored != "" { return pid, *mirrored } return pid, fromAgent }