fix(inbox): 收件箱按**工作区**收窄（三维地址的 path 位此前从未被使用）
用户 12 天前就提过（`552fbc7` 只修了 session_id 那一维），这轮才真修。 用户原话：「难道让一个不在项目工作区的 agentsession 去修工程吗？」 # 缺陷（生产实测，2026-09-26） 在 `mc` 工作区干活的 pi 读收件箱拿到 **200 封，其中 191 封属于 `/home/program/agentmail`** —— 它照着那些信里的断言去改 agentmail 的代码， 把手上的 mc 活丢在一边。用户当场问它「你怎么干着干着修 agentmail 去了？」 （这条对话就在 mc 会话的 jsonl 里） 根因：`ListInboxScoped` 的 WHERE 只有 `m.to_name = $1`（+ 可选 session_id）， **没有任何 workspace 条件**。三维地址 `name@path.session` 的 path 位 在收件箱侧从未生效 —— 那不是"另一种语义"，是没兑现契约。 # 三条守卫全部只覆盖自动转发，防不住这个 | 守卫 | 只覆盖 | 为何无效 | | --- | --- | --- | | 会话预算 | `relay != ""` 才扣 | 这批信 relay=0（模型主动发）⇒ 不扣 | | maxRelayHops=5 | 同上，只数 relay | 同上 ⇒ 不进那个分支 | | 插件自动转发守卫 | 插件代劳时 | 日志明说"本轮不自动转发" ⇒ 模型自己发的不受管 | # 服务端 · `ListInboxScoped` / `CountUnreadScoped` / `MarkAllInboxReadForSession` 三处统一加 `s.workspace = $N`（用会话的 workspace，不用 mails.to_workspace： 后者是信封字段、可能是抄送或历史遗留；"线索属于哪个工作区"是会话属性）。 ★ 三处必须是**同一个谓词** —— 列表看不到的信却被"全部标掉"标掉就是静默丢信 （session_scope_test.go 记过这个形状）。 · **workspace 在 Agent 侧必需，缺了 400**（用户裁定：「不带 workspace 是错误 发件格式，直接退回！」）。旧语义（不带=全部）正是缺陷本身，不留兼容回退。 · 人类侧**不过滤**（一个人跨工作区，WebUI 按 session_workspace 分组显示）—— 所以"必需"这条约束放在 Handler 而不是 repo 层：它是接口契约，不是数据层不变量。 · 新增 `UnreadWorkspaces`：心跳是**进程级**（一个桥服务所有工作区），没有 "我的工作区"可言；但只有总数桥不知道去哪个工作区补投 ⇒ 心跳回 `pending_workspaces` 清单，桥逐个消费。 · 决策载荷补 `workspace`（服务端知道 session→workspace，插件重启后推不出来）。 · `TouchAgentLastSeen` 从 HeartbeatAgent 拆出：middleware 在每个认证请求上都调它， 而那时工作区还没解析（请求体没读），原来在白算一次 CountUnread。 # 三个插件（pi / opencode / dsh） · 读类工具带 `workspace`；补投从"读一次全局收件箱"改为**逐工作区**读。 · pi：worker 信封的 `to_workspace` 经闭包递进工具（不是会话文件 header 的 cwd —— 后者是"会话上次落在哪"，前者是"这封信寄到哪个工作区"）。 · opencode/dsh：插件常驻、信封在 deliverMail 那刻就消费掉了 ⇒ 新增 `sessionWorkspace` 映射（键与既有 reverseMap 同一把）。 · 修一处真 bug：`UnreadWorkspaces` 原先会返回相对路径工作区（历史库里有 `workspace='root'`），桥侧实测撞 400（`补投工作区 root 失败`）⇒ 只报可寻址的。 # 实测凭据 · 改前：`pi` 的收件箱 200 封混 3 个工作区（agentmail 191 / TrueAgent 7 / huawei 2） · 改后：agentmail=100（total 228）、mc=16、TrueAgent=7 —— 各工作区独立 · 不带 workspace ⇒ **HTTP 400**，话术给出可执行步骤 · 桥日志：`rw=/home/newqqagent/plugindev/editdoc-upgrade` —— 终于是别的工作区了 （改前 78 次 worker 启动**全部**是 `/home/program/agentmail`） # 判据 · `server/internal/repo/workspace_scope_test.go`（3 条）： 两向收窄 + **反向对照**（不带时两条都看得到 ⇒ 证明是收窄不是清空）+ 未读数同口径 + 相对路径必须报错 · `plugins/pi-mail-bridge/test/inbox-workspace-scope.test.mjs`（4 条）：接线 + 取信封而非 cwd + 补投逐工作区 + 判据自检 · dsh 那条 `取不到会话时退回整体收件箱` **改了**：它钉的"退回整体"正是缺陷， 现在钉"两维各自缺席时各自不带、服务端 400 让错误可见" · 变异验证：服务端 2 处 + 插件 3 处，全部判红后恢复回绿 全量：server `go test ./...` 绿；三插件 513+340+403 全绿。
This commit is contained in:
@ -173,6 +173,20 @@ class GatewayClient {
|
||||
|
||||
const sessionMap = new BoundedMap<string, { dshSessionId: string; directory: string }>(MAX_TRACKED_SESSIONS);
|
||||
const reverseMap = new BoundedMap<string, string>(MAX_TRACKED_SESSIONS);
|
||||
|
||||
/*
|
||||
★ 2026-09-26：DSH 会话 id -> **工作区**。
|
||||
|
||||
收件箱接口现在要求 `workspace`（缺了 400），而 `read_inbox` 的 URL 是从
|
||||
`exec.agent` 现场拼的 —— 那里只有 DSH 的会话 id，推不出工作区。
|
||||
信封（`data.to_workspace`）在 `deliverMail` 那一刻就消费掉了，
|
||||
到 `read_inbox` 时已不在作用域里 ⇒ 在这里留一份映射。
|
||||
|
||||
用户报的缺陷（这一维此前完全没做）：在 `mc` 工作区干活的会话读收件箱拿到
|
||||
agentmail 的信，照着去改 agentmail 的代码。三维地址 `name@path.session`
|
||||
的 **path 位本来就该参与寻址**。
|
||||
*/
|
||||
const sessionWorkspace = new BoundedMap<string, string>(MAX_TRACKED_SESSIONS);
|
||||
const mailDrivenSessions = new BoundedSet<string>(MAX_TRACKED_SESSIONS);
|
||||
// 回信上下文。fromHuman / inReplyTo 是服务端给的两个信号：
|
||||
// 前者决定要不要自动转发（Agent 间不转，见 lib/relay-policy.js），
|
||||
@ -475,27 +489,43 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
* SSE 只推连上之后的事件，插件重启前发来的邮件不会再推一次。
|
||||
* 不补的话那封邮件永远躺在收件箱里，而发件人以为 Agent 收到了。
|
||||
*/
|
||||
async function catchUp(pending: unknown): Promise<void> {
|
||||
async function catchUp(pending: unknown, workspaces?: unknown): Promise<void> {
|
||||
if (!pending) return;
|
||||
try {
|
||||
const box = await client.get('/mail/inbox?status=unread&limit=20');
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
if (tasks.length === 0) return;
|
||||
console.error(`[dsh-mail-bridge] 补投 ${tasks.length} 封离线期间的邮件（共 ${pending} 封未读）`);
|
||||
// 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求
|
||||
for (const ev of tasks) {
|
||||
// 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封
|
||||
if (deliveredMails.has(ev.mail_id)) continue;
|
||||
deliveredMails.add(ev.mail_id);
|
||||
try {
|
||||
await deliverMail(ev, 'mail');
|
||||
} catch (e: any) {
|
||||
console.error(`[dsh-mail-bridge] 补投 ${ev.mail_id} 失败: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
} catch (e: any) {
|
||||
console.error(`[dsh-mail-bridge] 补投失败: ${e?.message || e}`);
|
||||
/*
|
||||
★ 逐工作区补投，不再读一次全局收件箱。
|
||||
旧写法不带收窄 ⇒ 会把**所有工作区**的漏投一起重放（在 mc 干活却被补投
|
||||
agentmail 的信）。清单来自心跳的 `pending_workspaces`：心跳是进程级、
|
||||
没有"我的工作区"可言，所以由它给清单，这里逐个消费。
|
||||
*/
|
||||
const list = Array.isArray(workspaces) ? (workspaces as string[]) : [];
|
||||
if (list.length === 0) {
|
||||
console.error(`[dsh-mail-bridge] 补投跳过：pending_mails=${pending} 但服务端未给出 pending_workspaces（旧版服务端？）`);
|
||||
return;
|
||||
}
|
||||
let delivered = 0;
|
||||
for (const ws of list) {
|
||||
try {
|
||||
const box = await client.get(`/mail/inbox?status=unread&limit=20&workspace=${encodeURIComponent(ws)}`);
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
if (tasks.length === 0) continue;
|
||||
// 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求
|
||||
for (const ev of tasks) {
|
||||
// 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封
|
||||
if (deliveredMails.has(ev.mail_id)) continue;
|
||||
deliveredMails.add(ev.mail_id);
|
||||
try {
|
||||
await deliverMail(ev, 'mail');
|
||||
delivered += 1;
|
||||
} catch (e: any) {
|
||||
console.error(`[dsh-mail-bridge] 补投 ${ev.mail_id} 失败: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
} catch (e: any) {
|
||||
// 单个工作区失败不影响其余（与"心跳失败不报错"同一原则）
|
||||
console.error(`[dsh-mail-bridge] 补投工作区 ${ws} 失败（不影响其余）: ${e?.message || e}`);
|
||||
}
|
||||
}
|
||||
if (delivered) console.error(`[dsh-mail-bridge] 补投 ${delivered} 封离线期间的邮件（共 ${pending} 封未读，跨 ${list.length} 个工作区）`);
|
||||
}
|
||||
|
||||
async function beat(): Promise<void> {
|
||||
@ -523,7 +553,7 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
// 每轮心跳都补的话会把「模型正在处理中、尚未标已读」的邮件重复投递。
|
||||
if (!caughtUp) {
|
||||
caughtUp = true;
|
||||
await catchUp(res?.pending_mails);
|
||||
await catchUp(res?.pending_mails, res?.pending_workspaces);
|
||||
}
|
||||
} catch {
|
||||
// 心跳失败不报错：网络抖动很常见，下一轮会补上。
|
||||
@ -871,6 +901,14 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
* 取不到返回空串（= 退回整个 Agent 的收件箱）。宁可退回旧行为，也不猜 ——
|
||||
* 猜错会把别人会话的未读标掉，那是静默丢信。
|
||||
*/
|
||||
/** 这次调用所属会话的工作区（与 mailSessionOf 同一把键）。拿不到返回空串 —— 那时
|
||||
* 收件箱请求会 400，是刻意的：错误可见，好过静默跨工作区拿到别处的信。 */
|
||||
function workspaceOf(exec: any): string {
|
||||
const dshSessionId = exec?.agent?.id ?? exec?.agent ?? '';
|
||||
if (!dshSessionId) return '';
|
||||
return sessionWorkspace.get(String(dshSessionId)) ?? '';
|
||||
}
|
||||
|
||||
function mailSessionOf(exec: any): string {
|
||||
const dshSessionId = String(exec?.agent?.id ?? '');
|
||||
if (!dshSessionId) return '';
|
||||
@ -919,6 +957,8 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
const cwd = String(agent?.session?.header?.cwd ?? '');
|
||||
sessionMap.set(mailSessionID, { dshSessionId: id, directory: cwd });
|
||||
reverseMap.set(id, mailSessionID);
|
||||
/* 会话自己的 cwd 就是它的工作区 —— read_inbox 的收窄要用它 */
|
||||
if (cwd) sessionWorkspace.set(id, cwd);
|
||||
mailDrivenSessions.add(id);
|
||||
console.error(`[dsh-mail-bridge] 由确定性 id 恢复会话映射 ${mailSessionID} -> ${id}`);
|
||||
}
|
||||
@ -963,6 +1003,8 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
if (!mailSessionID) return;
|
||||
sessionMap.set(mailSessionID, { dshSessionId, directory: cwd });
|
||||
reverseMap.set(dshSessionId, mailSessionID);
|
||||
/* 同上：cwd 即工作区 */
|
||||
if (cwd) sessionWorkspace.set(dshSessionId, cwd);
|
||||
mailDrivenSessions.add(dshSessionId);
|
||||
mailContexts.set(mailSessionID, {
|
||||
replyTo: data.from_name || '',
|
||||
@ -1341,7 +1383,15 @@ export function apply(ctx: any, config: PluginConfig): void {
|
||||
async execute(args: any, exec?: any): Promise<string> {
|
||||
const status = args.status || DEFAULT_INBOX_STATUS;
|
||||
const mailSessionID = mailSessionOf(exec);
|
||||
const scope = mailSessionID ? `&session_id=${encodeURIComponent(mailSessionID)}` : '';
|
||||
/*
|
||||
★ 两维收窄都要带：
|
||||
· session_id —— 只列**这条线索**的信（防"A 会话标掉 B 会话的未读"）
|
||||
· workspace —— 只列**这个工作区**的信（防"在 mc 干活却读到 agentmail 的信"）
|
||||
服务端对缺 workspace 直接 400 —— 刻意如此（旧语义正是那个缺陷本身）。
|
||||
*/
|
||||
const ws = workspaceOf(exec);
|
||||
const scope = (mailSessionID ? `&session_id=${encodeURIComponent(mailSessionID)}` : '')
|
||||
+ (ws ? `&workspace=${encodeURIComponent(ws)}` : '');
|
||||
const { mails } = await client.get(
|
||||
`/mail/inbox?status=${status}&limit=${args.limit || DEFAULT_INBOX_LIMIT}${scope}`
|
||||
);
|
||||
|
||||
@ -32,9 +32,20 @@ test('★ 会话来源是平台上下文（并发安全），不是模块级变
|
||||
assert.ok(!bad.test(src), '不得用模块级"当前会话"变量（同进程多会话会互相覆盖）');
|
||||
});
|
||||
|
||||
test('取不到会话时退回整体收件箱（不猜）', () => {
|
||||
assert.match(src, /if \(!dshSessionId\) return ''/);
|
||||
assert.match(src, /const scope = mailSessionID \? `&session_id=[^`]*` : ''/);
|
||||
test('取不到会话时不猜（session 位与 workspace 位各自独立缺席）', () => {
|
||||
assert.match(src, /if \(!dshSessionId\) return ''/, '拿不到 DSH 会话 → 空串，不去反查');
|
||||
/*
|
||||
★★ 2026-09-26 这条断言改了：原来钉的是"取不到会话时退回**整体**收件箱"。
|
||||
那个"退回整体"正是缺陷 —— 在 mc 工作区干活的会话会因此拿到 agentmail 的信
|
||||
并照着去改 agentmail 的代码（用户当场指出）。
|
||||
|
||||
现在的规矩：两维**各自**缺席时各自不带，服务端对缺 workspace 直接 400。
|
||||
"错误可见"胜于"静默跨工作区"。（服务端那半见 workspace_scope_test.go）
|
||||
*/
|
||||
assert.match(src, /const ws = workspaceOf\(exec\);/,
|
||||
'workspace 要经 workspaceOf(exec) 取（与 mailSessionOf 同一把键）');
|
||||
assert.match(src, /ws \? `&workspace=\$\{encodeURIComponent\(ws\)\}` : ''/,
|
||||
'workspace 那维要真的拼进去');
|
||||
});
|
||||
|
||||
test('★ 判据自检：不带收窄的旧写法必须判红', () => {
|
||||
|
||||
@ -273,7 +273,16 @@ const readInboxTool = {
|
||||
const filter = args.filter || DEFAULT_INBOX_STATUS;
|
||||
const limit = args.limit || DEFAULT_INBOX_LIMIT;
|
||||
const mailSessionID = reverseMap.get(String(context?.sessionID ?? "")) || "";
|
||||
const scope = mailSessionID ? `&session_id=${encodeURIComponent(mailSessionID)}` : "";
|
||||
const workspace = sessionWorkspace.get(String(context?.sessionID ?? "")) || "";
|
||||
/*
|
||||
★ 两维收窄都要带：
|
||||
· session_id —— 只列**这条线索**的信（防"A 会话标掉 B 会话的未读"）
|
||||
· workspace —— 只列**这个工作区**的信（防"在 mc 干活却读到 agentmail 的信"）
|
||||
服务端对缺 workspace 直接 400 —— 那是刻意的（旧语义正是缺陷本身）。
|
||||
workspace 拿不到时**不带**，让服务端报 400：错误可见，好过静默跨工作区。
|
||||
*/
|
||||
const scope = (mailSessionID ? `&session_id=${encodeURIComponent(mailSessionID)}` : "")
|
||||
+ (workspace ? `&workspace=${encodeURIComponent(workspace)}` : "");
|
||||
const data = await apiGet(`/mail/inbox?status=${filter}&limit=${limit}${scope}`);
|
||||
|
||||
// 渲染与已读策略放 lib/inbox-format.js：它们与平台 SDK 无关，
|
||||
@ -603,6 +612,28 @@ function startSSE(onEvent) {
|
||||
// 条目，而 GC 收不掉（还被强引用着）。
|
||||
const sessionMap = new BoundedMap(MAX_TRACKED_SESSIONS); // agentmail session_id -> opencode session id
|
||||
const reverseMap = new BoundedMap(MAX_TRACKED_SESSIONS); // opencode session id -> agentmail session_id（供 event 钩子回写命名）
|
||||
|
||||
/*
|
||||
★ 2026-09-26：opencode session id -> **工作区**。
|
||||
|
||||
# 为什么要这一张表
|
||||
|
||||
收件箱接口现在要求 `workspace`（缺了 400），而 `read_inbox` 的 URL 是从
|
||||
`context.sessionID` 现场拼的 —— 那里只有 opencode 的会话 id，推不出工作区。
|
||||
|
||||
# 与 pi 桥的差别（同一件事的两种做法）
|
||||
|
||||
pi 的 worker 一个进程只服务一封邮件，信封就在手上（`data.to_workspace`），
|
||||
所以它直接把闭包传进工具。opencode 是**插件常驻、按会话分派**，
|
||||
信封在 `resolveSessionForMail` 那一刻消费掉了，到 `read_inbox` 时已经不在作用域里。
|
||||
⇒ 在这里留一份映射，键与 reverseMap 同一个（都是 opencode 的 session id）。
|
||||
|
||||
# 用户报的缺陷（这一维此前完全没做）
|
||||
|
||||
在 `mc` 工作区干活的会话读收件箱拿到 agentmail 的 191 封信，照着去改 agentmail
|
||||
的代码。三维地址 `name@path.session` 的 **path 位本来就该参与寻址**。
|
||||
*/
|
||||
const sessionWorkspace = new BoundedMap(MAX_TRACKED_SESSIONS); // opencode session id -> 工作区绝对路径
|
||||
const syncedTitles = new BoundedMap(MAX_TRACKED_SESSIONS); // opencode session id -> 已回写过的标题（去重，避免 session.updated 刷屏）
|
||||
|
||||
// 权限询问的双向定位。
|
||||
@ -666,6 +697,7 @@ async function resolveSessionForMail(client, directory, data, kind) {
|
||||
if (mailSessionID) {
|
||||
sessionMap.set(mailSessionID, adoptedID);
|
||||
reverseMap.set(adoptedID, mailSessionID);
|
||||
if (wantDir) sessionWorkspace.set(adoptedID, wantDir);
|
||||
// 标记为邮件驱动：接管之后这条会话**开始**参与邮件往来，
|
||||
// 轮次结束要把总结转回发件人。不标记的话邮件投进去了却永远没有回音。
|
||||
mailDrivenSessions.add(adoptedID);
|
||||
@ -706,6 +738,7 @@ async function resolveSessionForMail(client, directory, data, kind) {
|
||||
if (mailSessionID) {
|
||||
sessionMap.set(mailSessionID, sessionID);
|
||||
reverseMap.set(sessionID, mailSessionID);
|
||||
if (wantDir) sessionWorkspace.set(sessionID, wantDir);
|
||||
mailDrivenSessions.add(sessionID);
|
||||
|
||||
// slug 在创建时就有（如 nimble-lagoon），立即作为寻址别名回写；
|
||||
@ -1186,28 +1219,46 @@ export default async function mailBridge(input) {
|
||||
* SSE 只推连上之后的事件，插件重启前发来的邮件不会再推一次。
|
||||
* 不补的话那封邮件永远躺在收件箱里，而发件人以为 Agent 收到了。
|
||||
*/
|
||||
async function catchUp(pending) {
|
||||
async function catchUp(pending, workspaces) {
|
||||
if (!pending) return;
|
||||
try {
|
||||
const box = await apiGet("/mail/inbox?status=unread&limit=20");
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
if (tasks.length === 0) return;
|
||||
console.error(`[mail-bridge] 补投 ${tasks.length} 封离线期间的邮件（共 ${pending} 封未读）`);
|
||||
// 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求
|
||||
for (const ev of tasks) {
|
||||
// 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封
|
||||
// （selectCatchup 只在拉完那一刻去过重）
|
||||
if (deliveredMails.has(ev.mail_id)) continue;
|
||||
deliveredMails.add(ev.mail_id);
|
||||
try {
|
||||
await deliverMail(client, directory, ev, "mail");
|
||||
} catch (e) {
|
||||
console.error(`[mail-bridge] 补投 ${ev.mail_id} 失败:`, e?.message || e);
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
console.error("[mail-bridge] 补投失败:", e?.message || e);
|
||||
/*
|
||||
★ 逐工作区补投，不再读一次全局收件箱。
|
||||
|
||||
旧写法 `apiGet("/mail/inbox?status=unread&limit=20")` 不带任何收窄，
|
||||
会把**所有工作区**的漏投一起重放 —— 在 mc 干活却被补投 agentmail 的信。
|
||||
清单来自心跳的 `pending_workspaces`（心跳是进程级、没有"我的工作区"，
|
||||
所以由它给清单，这里逐个消费）。
|
||||
*/
|
||||
const list = Array.isArray(workspaces) ? workspaces : [];
|
||||
if (!list.length) {
|
||||
console.error(`[mail-bridge] 补投跳过：pending_mails=${pending} 但服务端未给出 pending_workspaces（旧版服务端？）`);
|
||||
return;
|
||||
}
|
||||
let delivered = 0;
|
||||
for (const ws of list) {
|
||||
try {
|
||||
const box = await apiGet(`/mail/inbox?status=unread&limit=20&workspace=${encodeURIComponent(ws)}`);
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
if (tasks.length === 0) continue;
|
||||
// 串行：每封都要起一轮模型，并发放出去等于对上游打 N 个并发请求
|
||||
for (const ev of tasks) {
|
||||
// 逐封再查一次：拉收件箱和逐封投递之间 SSE 可能已经投过其中某封
|
||||
// （selectCatchup 只在拉完那一刻去过重）
|
||||
if (deliveredMails.has(ev.mail_id)) continue;
|
||||
deliveredMails.add(ev.mail_id);
|
||||
try {
|
||||
await deliverMail(client, directory, ev, "mail");
|
||||
delivered += 1;
|
||||
} catch (e) {
|
||||
console.error(`[mail-bridge] 补投 ${ev.mail_id} 失败:`, e?.message || e);
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
// 单个工作区失败不影响其余（与"心跳失败不报错"同一原则）
|
||||
console.error(`[mail-bridge] 补投工作区 ${ws} 失败（不影响其余）:`, e?.message || e);
|
||||
}
|
||||
}
|
||||
if (delivered) console.error(`[mail-bridge] 补投 ${delivered} 封离线期间的邮件（共 ${pending} 封未读，跨 ${list.length} 个工作区）`);
|
||||
}
|
||||
|
||||
let caughtUp = false;
|
||||
@ -1232,7 +1283,7 @@ export default async function mailBridge(input) {
|
||||
// 每轮心跳都补的话会把「模型正在处理中、尚未标已读」的邮件重复投递。
|
||||
if (!caughtUp) {
|
||||
caughtUp = true;
|
||||
await catchUp(res?.pending_mails);
|
||||
await catchUp(res?.pending_mails, res?.pending_workspaces);
|
||||
}
|
||||
} catch {
|
||||
// 心跳失败不报错：网络抖动很常见，下一轮会补上。
|
||||
|
||||
@ -183,7 +183,13 @@ async function routeDecision(data) {
|
||||
const decision = String(data?.decision || '拒绝');
|
||||
const note = typeof data?.note === 'string' ? data.note : '';
|
||||
const sessionID = data?.session_id || '';
|
||||
const waiting = sessionID ? await collectWaitingMails(sessionID) : [];
|
||||
/*
|
||||
★ workspace 由服务端给（决策载荷里的 `workspace`）—— 插件推不出来：
|
||||
重启后待决映射丢光，此时手上只有这个事件。缺了它下面的收件箱读会 400，
|
||||
而这条路的用途是"把人在等待期间补的更正一并交给模型"，读不到 = 更正丢失。
|
||||
*/
|
||||
const workspace = data?.workspace || '';
|
||||
const waiting = sessionID ? await collectWaitingMails(sessionID, workspace) : [];
|
||||
|
||||
if (relayKey && pool.routePermission(relayKey, decision, note, waiting)) {
|
||||
log(`权限 ${relayKey} 决策 ${data.decision}（决策人 ${data.decided_by || '?'}）已转交 worker`
|
||||
@ -206,9 +212,18 @@ async function routeDecision(data) {
|
||||
*
|
||||
* 拉取失败**不能拖垮决策投递**：备注还在，最坏情况是模型晚一步看到更正。
|
||||
*/
|
||||
async function collectWaitingMails(sessionID) {
|
||||
async function collectWaitingMails(sessionID, workspace = '') {
|
||||
// ★ 必须带 workspace（服务端缺它 400）。旧版这里读的是**全局**收件箱 ——
|
||||
// 那正是"在 mc 干活却拿到 agentmail 的信"那条缺陷的另一处载体。
|
||||
// workspace 为空时不读：宁可少给一封更正，也不要跨工作区误取。
|
||||
if (!workspace) {
|
||||
log('等人期间的邮件拉取跳过：决策载荷未带 workspace（旧版服务端？）');
|
||||
return [];
|
||||
}
|
||||
try {
|
||||
const box = await client.get('/mail/inbox?status=unread&limit=20');
|
||||
const box = await client.get(
|
||||
`/mail/inbox?status=unread&limit=20&workspace=${encodeURIComponent(workspace)}`,
|
||||
);
|
||||
return selectWaitingMails(box, { sessionID, seen: deliveredMails });
|
||||
} catch (e) {
|
||||
log(`等人期间的邮件拉取失败（不影响决策投递）: ${describeError(e)}`);
|
||||
@ -260,21 +275,57 @@ async function reportModels() {
|
||||
* 并发放出去等于对上游打 N 个并发请求」—— 那个约束现在由 pool 的 maxWorkers
|
||||
* 承担，而且它比串行更好：同一条会话仍然串行，不同会话可以并行。
|
||||
*/
|
||||
async function catchUp(pending) {
|
||||
/**
|
||||
* 补投离线期间的邮件。
|
||||
*
|
||||
* ★★ 2026-09-26 改：**逐工作区**补投，而不是读一次全局收件箱。
|
||||
*
|
||||
* ── 原来错在哪 ──
|
||||
* 它调 `/mail/inbox?status=unread&limit=20` —— **不带任何收窄**。那时的语义是
|
||||
* "该 Agent 的全部未读"，于是桥会把**所有工作区**的漏投邮件一起重放：
|
||||
* 我在 `mc` 工作区干活，却被补投一堆 `agentmail` 的信（用户报的那类现象）。
|
||||
*
|
||||
* ── 现在怎么走 ──
|
||||
* 心跳返回 `pending_workspaces`（哪些工作区有未读）。逐个工作区去读 ——
|
||||
* 每次读都带上那个工作区的 `workspace`，与服务端要求的收窄口径一致。
|
||||
* 服务端在缺 workspace 时直接 400，所以这里不可能"忘了带"。
|
||||
*
|
||||
* ── 为什么不是"心跳带一个 workspace" ──
|
||||
* 心跳是**进程级**的（一个桥进程服务所有工作区），它没有"我的工作区"可言。
|
||||
* 而收件箱是 **worker 级**的（每个 worker 手上那封信有明确的 path 位）。
|
||||
* 在进程级强制 workspace 是概念错配 —— 所以心跳给清单，这里逐个消费。
|
||||
*
|
||||
* @param {number} pending 全局未读总数（仅用于日志；不再是过滤依据）
|
||||
* @param {string[]} workspaces 有未读的工作区清单
|
||||
*/
|
||||
async function catchUp(pending, workspaces) {
|
||||
if (!pending) return;
|
||||
try {
|
||||
const box = await client.get('/mail/inbox?status=unread&limit=20');
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
if (!tasks.length) return;
|
||||
log(`补投 ${tasks.length} 封离线期间的邮件（共 ${pending} 封未读）`);
|
||||
for (const ev of tasks) {
|
||||
if (deliveredMails.has(ev.mail_id)) continue; // 逐封再查（B-7.6）
|
||||
deliveredMails.add(ev.mail_id);
|
||||
pool.submit('mail', ev);
|
||||
}
|
||||
} catch (e) {
|
||||
log(`补投失败: ${describeError(e)}`);
|
||||
const list = Array.isArray(workspaces) ? workspaces : [];
|
||||
if (!list.length) {
|
||||
// 有未读却给不出工作区：只可能是上报侧出了问题。明说，别静默跳过。
|
||||
log(`补投跳过：pending_mails=${pending} 但服务端未给出 pending_workspaces（旧版服务端？）`);
|
||||
return;
|
||||
}
|
||||
let delivered = 0;
|
||||
for (const ws of list) {
|
||||
try {
|
||||
// ★ workspace 是必填参数 —— 缺了服务端 400（这是刻意的，见服务端 GetInbox）。
|
||||
const box = await client.get(
|
||||
`/mail/inbox?status=unread&limit=20&workspace=${encodeURIComponent(ws)}`,
|
||||
);
|
||||
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
|
||||
for (const ev of tasks) {
|
||||
if (deliveredMails.has(ev.mail_id)) continue; // 逐封再查（B-7.6）
|
||||
deliveredMails.add(ev.mail_id);
|
||||
pool.submit('mail', ev);
|
||||
delivered += 1;
|
||||
}
|
||||
} catch (e) {
|
||||
// 单个工作区失败不影响其余：与"心跳失败不报错"同一原则。
|
||||
log(`补投工作区 ${ws} 失败（不影响其余）: ${describeError(e)}`);
|
||||
}
|
||||
}
|
||||
if (delivered) log(`补投 ${delivered} 封离线期间的邮件（共 ${pending} 封未读，跨 ${list.length} 个工作区）`);
|
||||
}
|
||||
|
||||
// ─── 启动 / 关停 ───
|
||||
@ -370,7 +421,7 @@ async function main() {
|
||||
if (Array.isArray(res?.allowed_models)) allowedModels = res.allowed_models; // B-2.2
|
||||
if (!caughtUp) { // B-7.1：只在首个成功心跳后补一次
|
||||
caughtUp = true;
|
||||
await catchUp(res?.pending_mails);
|
||||
await catchUp(res?.pending_mails, res?.pending_workspaces);
|
||||
}
|
||||
} catch {
|
||||
// B-2.1：心跳失败不重试不报错。真连不上时 Gateway 会把它判成离线，
|
||||
|
||||
@ -61,30 +61,65 @@ const text = (s) => ({ content: [{ type: 'text', text: s }] });
|
||||
* opencode / dsh / homeagent 三个平台都没写这一行，只有这里写了 —— 它不是
|
||||
* 「更严格更好」，而是与 pi 的参数传递机制直接冲突。
|
||||
*/
|
||||
export function createMailTools({ client, log, agentName = '', onReconnect, getMailSessionId = () => ''}) {
|
||||
// ─── 会话收窄参数（读类工具的公共前缀）───
|
||||
export function createMailTools({ client, log, agentName = '', onReconnect,
|
||||
getMailSessionId = () => '', getWorkspace = () => '' }) {
|
||||
// ─── 收窄参数（读类工具的公共前缀）：两维，都必须由 worker 闭包递进来 ───
|
||||
//
|
||||
// 服务端拿它干两件事：
|
||||
// ① 收件箱只列/只标本会话的邮件（缺了它，A 会话的 worker 会把 B 会话的未读标掉
|
||||
// ⇒ 补投再也看不到那封信 = 静默丢信。用户原话「不同 session 的 agent
|
||||
// 都可以看到全部邮件」）；
|
||||
// ★★ 2026-09-26 补上 workspace 这一维，并订正下面那段注释。
|
||||
//
|
||||
// ── 这段注释原来写着什么（以及为什么它是错的）──
|
||||
// ② **工作区隔离**：服务端由这条 session 反查 workspace，只有同工作区的会话才放行。
|
||||
// 一个 Agent 同时服务所有工作区，不带这一维时在 TrueAgent 里干活的 worker
|
||||
// 能读到 agentmail 的整条线索（2026-09-14 用户报的那类越界）。
|
||||
// ……服务端不接受调用方直接声明工作区，那等于自己给自己发通行证。
|
||||
//
|
||||
// 取值只能是**邮件会话 id**，且必须由 worker 闭包递进来（模型改不了它）——
|
||||
// 服务端不接受调用方直接声明工作区，那等于自己给自己发通行证。
|
||||
// 前半句描述的能力**服务端从来没有过** —— `ListInboxScoped` 的 WHERE 里
|
||||
// 只有 `m.to_name = $1`（+ 可选的 session_id），没有任何 workspace 条件。
|
||||
// 我（写这段注释的人）把"设计意图"当成"已实现"，于是插件侧也一直只传 session_id。
|
||||
//
|
||||
// 后半句的理由**不成立**：session_id → workspace 这条反查在服务端确实可行，
|
||||
// 但它只覆盖"这条线索属于哪个工作区"，覆盖不了"**我这个 worker 在哪个工作区**"。
|
||||
// 两者的差别就是缺陷本身：见下。
|
||||
//
|
||||
// ── 缺陷（用户 2026-09-26 当场指出，此前已提过多次）──
|
||||
// 在 `mc` 工作区干活的 pi 读收件箱拿到 200 封，其中 **191 封属于
|
||||
// `/home/program/agentmail`** —— 它照着那些信里的断言去改 agentmail 的代码，
|
||||
// 把手上 mc 的活丢在一边（用户当场问「你怎么干着干着修 agentmail 去了？」）。
|
||||
//
|
||||
// ── 为什么 workspace 由**调用方声明**是对的（而不是"自己给自己发通行证"）──
|
||||
// 它是**过滤条件**，不是鉴权依据：声明错了只影响"我能看到什么"，
|
||||
// 越权不了别人的东西（服务端仍然只列 `to_name = 我` 的信）。
|
||||
// 与 session_id 同构。而且服务端没有别的办法知道它 —— 那是 worker 每回合的 cwd，
|
||||
// 随回合变，服务端拿不到可靠来源。
|
||||
//
|
||||
// ── 取值 ──
|
||||
// workspace 取**信封上的 path 位**（`data.to_workspace`），不是会话文件 header 里的
|
||||
// cwd：后者是会话上次落在哪，前者是"这封信寄到哪个工作区"—— 收件箱要回答的是后者。
|
||||
// （且 mailTools 在 loadSession **之前**装配，那时也拿不到 cwd。）
|
||||
//
|
||||
// ★ 服务端要求它**必需**：缺了直接 400（用户裁定「不带 workspace 是错误发件格式，
|
||||
// 直接退回！」）。所以这里不能再像原来那样"缺了就不带、让服务端走旧语义"——
|
||||
// 旧语义就是那个缺陷。
|
||||
const wsParam = () => {
|
||||
const ws = typeof getWorkspace === 'function' ? getWorkspace() : '';
|
||||
return ws ? `workspace=${encodeURIComponent(ws)}` : '';
|
||||
};
|
||||
const scopeQS = () => {
|
||||
const sid = typeof getMailSessionId === 'function' ? getMailSessionId() : '';
|
||||
return sid ? `session_id=${encodeURIComponent(sid)}` : '';
|
||||
const parts = [];
|
||||
const ws = wsParam();
|
||||
if (ws) parts.push(ws);
|
||||
if (sid) parts.push(`session_id=${encodeURIComponent(sid)}`);
|
||||
return parts.join('&');
|
||||
};
|
||||
// 已经带了查询串的 URL 用这个（收件箱那条要 append 到 status/limit 后面）
|
||||
const inboxScope = () => {
|
||||
const q = scopeQS();
|
||||
return q ? `&${q}` : '';
|
||||
};
|
||||
// 任意路径用这个：自己判断该用 ? 还是 &。缺了 scope 就原样返回，
|
||||
// 让服务端走"旧语义 + 记警告"那条路，而不是拼出一个半截 URL。
|
||||
// 任意路径用这个：自己判断该用 ? 还是 &。
|
||||
//
|
||||
// ★ 不再有"缺了就原样返回让服务端走旧语义"那条路 —— 服务端现在对缺 workspace
|
||||
// 的读请求直接 400。缺了就会 400，这正是我们要的：宁可失败得明确，
|
||||
// 也不要静默看到别的工作区的信。
|
||||
const withScope = (path) => {
|
||||
const q = scopeQS();
|
||||
if (!q) return path;
|
||||
|
||||
@ -103,7 +103,7 @@ let piSessionId = '';
|
||||
// sessionID = **AgentMail 的邮件会话 id**（不是 pi 的 session id）。
|
||||
// read_inbox 要靠它把自己那条会话的邮件与别的会话区分开 —— 少了它，A 会话的
|
||||
// worker 会把 B 会话的未读一起列出来并标掉（2026-09-14 用户报的缺陷）。
|
||||
let mailContext = { replyTo: '', subject: '', mailID: '', sessionID: '', permissionMode: 'workspace' };
|
||||
let mailContext = { replyTo: '', subject: '', mailID: '', sessionID: '', permissionMode: 'workspace', workspace: '' };
|
||||
let lastSyncedName = '';
|
||||
let relayedKey = '';
|
||||
let finished = false;
|
||||
@ -558,6 +558,9 @@ async function run() {
|
||||
const mailTools = createMailTools({
|
||||
// read_inbox 用它在服务端把列表收窄到自己这条会话。
|
||||
getMailSessionId: () => mailContext.sessionID,
|
||||
// ★ 收件箱的第二维收窄：**这封信寄到哪个工作区**（信封的 path 位）。
|
||||
// 缺了它，在 mc 干活的 pi 会看到 agentmail 的 191 封信并照着去改 agentmail。
|
||||
getWorkspace: () => mailContext.workspace,
|
||||
client, log, agentName: job.config.agentName,
|
||||
onReconnect: () => send({
|
||||
type: 'reconfigure', url: client.baseURL, agentKey: client.agentKey,
|
||||
@ -735,6 +738,9 @@ process.on('message', (msg) => {
|
||||
mailID: msg.data?.mail_id || '',
|
||||
sessionID: msg.data?.session_id || '',
|
||||
permissionMode: msg.data?.permission_mode || 'workspace',
|
||||
/* ★ 信封上的 path 位 = 这封信寄到哪个工作区。读类工具用它收窄收件箱
|
||||
（缺了服务端会 400 —— 见 tools.mjs 里 createMailTools 那段注释）。 */
|
||||
workspace: msg.data?.to_workspace || '',
|
||||
};
|
||||
lastSyncedName = msg.lastSyncedName || '';
|
||||
for (const t of msg.grants || []) grants.add(t);
|
||||
|
||||
65
plugins/pi-mail-bridge/test/inbox-workspace-scope.test.mjs
Normal file
65
plugins/pi-mail-bridge/test/inbox-workspace-scope.test.mjs
Normal file
@ -0,0 +1,65 @@
|
||||
/**
|
||||
* read_inbox 必须把列表收窄到**自己所在的工作区**（2026-09-26 用户报的缺陷）。
|
||||
*
|
||||
* # 缺陷（用户当场指出，此前已提过多次）
|
||||
*
|
||||
* 在 `mc` 工作区干活的 pi 读收件箱拿到 200 封，其中 **191 封属于
|
||||
* `/home/program/agentmail`** —— 它照着那些信里的断言去改 agentmail 的代码，
|
||||
* 把手上 mc 的活丢在一边。用户当场问「你怎么干着干着修 agentmail 去了？」
|
||||
*
|
||||
* 三维地址是 `name@path.session`：**path 位本来就该参与寻址**，而收件箱侧此前
|
||||
* 完全没用它（`ListInboxScoped` 的 WHERE 里只有 `to_name`）。
|
||||
*
|
||||
* # 与 inbox-session-scope 的分工（两条判据、两维收窄，别混）
|
||||
*
|
||||
* · `session_id` —— 只列**这条线索**的信。防的是"A 会话的 worker 标掉 B 会话的未读"。
|
||||
* · `workspace` —— 只列**这个工作区**的信。防的是"在 mc 干活却读到 agentmail 的信"。
|
||||
*
|
||||
* 两维都要，缺一维就漏一类越界。这里只验**接线**；服务端语义
|
||||
* （读 A 工作区不动 B）由 `server/internal/repo/workspace_scope_test.go` 负责。
|
||||
*/
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
const HERE = dirname(fileURLToPath(import.meta.url));
|
||||
const tools = readFileSync(join(HERE, '..', 'src', 'tools.mjs'), 'utf8');
|
||||
const worker = readFileSync(join(HERE, '..', 'src', 'worker.mjs'), 'utf8');
|
||||
const index = readFileSync(join(HERE, '..', 'src', 'index.mjs'), 'utf8');
|
||||
|
||||
test('read_inbox 的收窄参数带上 workspace', () => {
|
||||
assert.match(tools, /workspace=\$\{encodeURIComponent\(ws\)\}/,
|
||||
'scopeQS 要真的拼出 workspace=…（服务端缺它直接 400）');
|
||||
assert.match(tools, /const ws = wsParam\(\);/,
|
||||
'scopeQS 要消费 wsParam，而不是只留着函数不调');
|
||||
});
|
||||
|
||||
test('worker 把"这封信寄到哪个工作区"递给工具', () => {
|
||||
assert.match(worker, /getWorkspace: \(\) => mailContext\.workspace/,
|
||||
'要传闭包而不是快照（worker 每回合的 mailContext 会变）');
|
||||
assert.match(worker, /workspace: msg\.data\?\.to_workspace/,
|
||||
'★ 取**信封的 path 位**（to_workspace），不是会话文件的 cwd —— ' +
|
||||
'前者是"这封信寄到哪个工作区"，后者是"会话上次落在哪"');
|
||||
});
|
||||
|
||||
test('补投逐工作区走，不再读一次全局收件箱', () => {
|
||||
// 原来：client.get('/mail/inbox?status=unread&limit=20') —— 不带收窄，
|
||||
// 于是桥会把所有工作区的漏投一起重放。
|
||||
assert.match(index, /pending_workspaces/,
|
||||
'心跳要消费 pending_workspaces 清单');
|
||||
assert.match(index, /mail\/inbox\?status=unread&limit=20&workspace=\$\{encodeURIComponent\(ws\)\}/,
|
||||
'每个工作区各读一次，且带上 workspace');
|
||||
assert.doesNotMatch(index, /client\.get\('\/mail\/inbox\?status=unread&limit=20'\)/,
|
||||
'旧的"不带收窄读全局"那条路必须消失（它就是跨工作区重放的载体）');
|
||||
});
|
||||
|
||||
test('★ 判据自检：拿没带工作区收窄的旧写法喂进来必须判红', () => {
|
||||
const oldScope = "const sid = typeof getMailSessionId === 'function' ? getMailSessionId() : '';\n return sid ? `session_id=${encodeURIComponent(sid)}` : '';";
|
||||
assert.equal(/workspace=\$\{encodeURIComponent\(ws\)\}/.test(oldScope), false);
|
||||
|
||||
const oldCatchup = "const box = await client.get('/mail/inbox?status=unread&limit=20');";
|
||||
assert.match(oldCatchup, /client\.get\('\/mail\/inbox\?status=unread&limit=20'\)/,
|
||||
'自检：旧写法确实能被那条 doesNotMatch 抓到');
|
||||
});
|
||||
Reference in New Issue
Block a user