#!/usr/bin/env node /** * pi-mail-bridge 的**投递工作进程** —— 一封邮件一个,跑完就退。 * * # 为什么必须是独立进程 * * 桥的主进程要一直读 SSE。而 pi 的会话装载是**同步**的: * `SessionManager.open()` 走 `openSync` + `readSync` 循环把整个 `.jsonl` * 读进内存并逐行 JSON.parse(SDK core/session-manager.js 的 * loadEntriesFromFile)。实测本机最大那条会话 23MB,`open` 一次 * **阻塞事件循环 118ms**;模型跑起来之后 SDK 内部还有大量同步工作。 * 全都发生在主线程上,SSE 读循环在那期间完全停住 —— 后续邮件卡在 TCP * 缓冲区里,久到 Gateway 认为连接死了,重连又触发重放。 * * 把模型工作搬进子进程后,主进程只剩「收事件 → 去重 → 分派」, * 实测同样的活在 fork 出的子进程里跑,主进程事件循环阻塞 0ms。 * * # 为什么是 child_process 而不是 worker_threads * * 两者实测都能建起 AgentSession。选进程的理由是**隔离**: * 模型会跑 bash/write/edit,一次 OOM 或 uncaughtException 不该带走 * 整座桥;而 `worker_threads` 与主线程共享堆和进程生命周期。 * 代价是每个 worker 约 140MB RSS 和 ~500ms 启动,由主进程的并发上限约束。 * * # 一封邮件一个进程带来的简化 * * 原来的「接管会话短暂持有」(adopted / adoptTimers / releaseAdopted) * 整套机制没有了:worker 退出**就是**释放,而且对普通会话和接管会话 * 一视同仁 —— 每封邮件都是「open → 跑一轮 → 还回去」。 * pi 假定「一个文件一个持有者」,主进程按会话串行分派保证了这一点。 * * # 进程边界上传什么 * * 主进程持有的是**路径与标量**(sessionFile / cwd / piSessionId / 授权表), * 不是 AgentSession 对象 —— 那东西跨不了进程。worker 每次从 sessionFile * 重新装载,拿到的是包含 TUI 期间写入的全部历史。 * * 协议见 src/pool.mjs 顶部。 */ import { existsSync } from 'node:fs'; import { homedir } from 'node:os'; import { join } from 'node:path'; import { ModelRuntime } from '@earendil-works/pi-coding-agent'; import { GatewayClient } from './gateway.mjs'; import { renderDecisionReason } from '../lib/denial-reason.js'; import { createMailTools } from './tools.mjs'; import { openSession, runTurn } from './session-pool.mjs'; import { buildMailPrompt, lastAssistantText, replySubject, relayKeyFor, describeError, renderResumeFailure } from './turn.mjs'; import { planNamingSync, planWriteBack } from './naming.mjs'; import { resolveWorkspaceCwd, ensureCwd } from '../lib/workspace.js'; import { piMailFallback } from './paths.mjs'; import { resolveSessionReuse } from './turn-cwd.mjs'; import { modelAttemptOrder, renderFailureReport } from '../lib/model-scope.js'; import { explicitSends, shouldSkipAutoRelay } from '../lib/relay-dedup.js'; import { autoRelayDecision } from '../lib/relay-policy.js'; import { adoptedSessionID, adoptMissingMessage } from '../lib/adopt.js'; import { isApproval, isAlwaysDecision } from '../lib/permission-grants.js'; import { normalizeMode, MODE_FULL, MODE_PLAN } from '../lib/permission-mode.js'; import { guardDecision, verifySandboxActive } from '../lib/sandbox.js'; import { writeFileSync, unlinkSync } from 'node:fs'; import { clampRelayKey, isPermanentFailure, isDuplicateRelay } from '../lib/relay-key.js'; // ─── 与主进程的通道 ─── /** 日志一律回传主进程:worker 的 stdout 会混在一起,加 pid 前缀才分得清。 */ const log = (...args) => send({ type: 'log', line: args.join(' ') }); function send(msg) { try { process.send?.(msg); } catch { /* 主进程已经走了,这条日志没有去处 */ } } /** relay_key -> resolve;决策由主进程的 SSE 收到后路由进来。 */ const pending = new Map(); /** * 这一轮里人点过「一直同意」的工具。 * * 初值由主进程在 job 里给(跨 worker 持久),新增的回报给主进程。 * 不用 lib/permission-grants.js 的 store:那份按 (会话, 工具) 存, * 而 worker 只服务一条会话,一个 Set 就够,且要能整体回传。 */ const grants = new Set(); // ─── 单封邮件的全部状态 ─── // // worker 只处理一封邮件、只碰一条会话,所以这些原来在主进程里 // 按 sessionId 分桶的 Map 在这里都退化成单个变量。 let job = null; let client = null; // 每次权限询问「人类说了什么」(备注 + 等人期间新到的邮件),按 relayKey 暂存。 // 见 lib/denial-reason.js 的说明。 const decidedExtra = new Map(); let modelRuntime = null; 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 lastSyncedName = ''; let relayedKey = ''; let finished = false; // ─── 权限钩子(B-8)─── /** * 与主进程版本逐条对应,差别只有两处: * - 「是不是邮件驱动的会话」不必查表:worker 只为邮件而存在。 * - 等决策的 promise 由主进程通过 IPC 唤醒,而不是本进程的 SSE。 * * 决策等待期间**只有这个 worker 停住**,主进程照常读 SSE、照常给别的会话 * 派活 —— 这正是原来最难受的一处:权限询问会让整座桥不再收信。 */ /** * 这个 worker 是不是真跑在沙箱里。 * * 父进程只设标记(`AGENTMAIL_PI_SANDBOXED=1`),但**标记不等于事实** —— * 判断错会让闸门既不问也不拦(最坏的一类)。所以现场自证一次:往界外写一个 * 金丝雀,写不进去才算数(见 lib/sandbox.js 的 verifySandboxActive)。 * * 结果缓存在进程级:自证会在界外留一个瞬时文件,不值得每轮重来。 */ let sandboxState = null; function sandboxActive() { if (sandboxState) return sandboxState.active; if (process.env.AGENTMAIL_PI_SANDBOXED !== '1') { sandboxState = { active: false, reason: '父进程没标记(这一轮没套沙箱)' }; } else { sandboxState = verifySandboxActive({ writeFile: writeFileSync, unlink: unlinkSync }); } log(`沙箱自证:${sandboxState.active ? '在边界内' : '不在边界内'} —— ${sandboxState.reason}`); return sandboxState.active; } function permissionExtension() { const GUARDED = new Set(['bash', 'write', 'edit']); return (pi) => { pi.on('tool_call', async (event, ctx) => { // ── 档位判定 ── // plan: 被守卫的工具一律直接拒绝(该档语义是「不动手」,没什么可问人的) // full: 不拦截(已声明全权) // workspace: 走原有问人流程 const mode = normalizeMode(mailContext.permissionMode); // 决策收在 lib/sandbox.js 的 guardDecision 里(纯函数,可单测): // full → 放行;plan → 拒;workspace + **沙箱在** → 放行(内核兜住边界); // workspace + 没沙箱 → 逐条问人(回退到原来那唯一一层)。 // 这一段以前只有"一律问",因为那时没有任何东西能判界内/界外。 const gate = guardDecision({ mode, sandboxed: sandboxActive(), toolName: event.toolName, guarded: GUARDED.has(event.toolName), }); if (gate === 'pass' || gate === 'allow') return; if (gate === 'block') { return { block: true, reason: `plan 档下不允许执行 ${event.toolName}。本档只允许读与查,请把方案写在回信里。如需动手请让发件人把档位改成 workspace。`, }; } const sid = ctx?.sessionManager?.getSessionId?.() || ''; // 只管自己那条会话。worker 里不该出现第二条,出现了说明有 bug —— // 让位(返回 undefined)比拦错一个安全。 if (piSessionId && sid && sid !== piSessionId) return; // 人点过「一直同意」→ 直接放行。必须在 POST 之前:否则每条命令都发一封 // 邮件,那个选项形同虚设(生产实测同一条会话被问了 15 次 bash)。 if (grants.has(event.toolName)) return; // relay_key 用 pi 给的 toolCallId(B-8.1):服务端随决策事件回传它。 // // clampRelayKey 不是防御性冗余:启用 extended thinking 时 Bedrock 把 // 思考签名拼进 toolCallId,实测长到 437 ~ 13601 字节,键直接超服务端 // 160 字节列宽 → 400。同一条会话里长短两种形态混着出现,于是权限询问 // 随机成功随机失败(生产日志 21:54:02 失败、21:54:24 同会话成功)。 const relayKey = clampRelayKey(`${sid || piSessionId}:${event.toolCallId}`); try { // 不传 `to`:决策人由服务端按 会话 owner → 线索里最近的人类 → 409 // 解析。插件只有本地上下文,猜不出「这条 Agent 链最初是谁派的活」。 const accepted = await client.post('/permission/request', { question: `是否允许执行 ${event.toolName}?`, options: ['同意', '一直同意', '拒绝'], context: [ describeToolCall(event), mailContext.subject ? `\n触发任务:${mailContext.subject}` : '', mailContext.replyTo ? `任务来自:${mailContext.replyTo}` : '', ].filter(Boolean).join('\n'), session_id: job.data?.session_id || '', relay_key: relayKey, }); // ★ 幂等命中:网关回 200 `{status:"duplicate_relay"}` 并**提前返回** —— // 没有产生新的询问,也永远不会有人来决策。而下面的 `await` 只由 // `permission_decision` 事件唤醒,重复的键永远等不到它。 // // 不认这个回包的代价是**静默挂死**:模型干等、人以为在跑, // 直到回合超时才以「处理失败」出现(而不是以「没人可问」出现, // 于是排查方向从一开始就是错的)。 // // 为什么会出现重复:键是确定性的(会话 + toolCallId),所以插件重启 // 后重放同一轮、SDK 重放同一个 tool call、以及上一次询问已被人决定过 // 而这一侧没收到那个决策(重启/断线)都会命中它。这些都是正常重试。 if (isDuplicateRelay(accepted)) { log(`权限询问被判定为重复(key=${relayKey}),本次没有产生新请求`); return { block: true, reason: [ `这个授权询问之前已经发过一次(key=${relayKey}),本次没有产生新的询问。`, '可能原因:插件重启后重放同一轮,或上一次询问已经有人决定过但这个决策没有回到这里。', '请改用不需要授权的方式完成,或在回信里说明需要人工执行哪一步(也可以请对方重新发一次任务)。', ].join('\n'), }; } } catch (e) { // 409 = 服务端判定这条任务链上没有人类,永远不会有人来点头。 // 当场 block 并把服务端建议原文当 reason:模型从工具报错里看到 // 「没人可问,换不需要权限的方式」才能自己改道,挂死时连重试机会都没有。 if (e?.status === 409) { const b = e.body || {}; /* * 409 里服务端会带上**真实档位**(`handler/permission.go` 的 * `"permission_mode": mode`)。若它说这条会话是 full 档,那 409 的 * 含义不是「没人可问」,而是「**本档根本不该问**」—— full 档的工具调用 * 无需审批,正确反应是**放行**,不是拦截。 * * 什么时候会走到这里:补投路径漏传档位(`lib/catchup.js`,2026-09-14 已修), * 于是 worker 拿默认 workspace 档去申请。没有这条兜底时,一条 full 档会话 * 只要有一封补投邮件,这一轮 bash/write/edit 全被 409 拦死。 * * 兜底**只认服务端明说的 full**:plan 档(该档语义是「不动手」,拦住是对的) * 与「这条链上没有人类」的 409 照旧 fail closed —— 猜宽了就是提权。 */ if (b.permission_mode === MODE_FULL) { log(`服务端判定本会话为 full 档,放行 ${event.toolName}(${relayKey}):无需审批`); return; } const reason = [ b.error || '权限询问无法送达:这条任务链上没有人类用户', b.detail || '', b.suggestion || '', ].filter(Boolean).join('\n'); log(`权限询问无人可投,当场拒绝 ${relayKey}:${b.error || ''}`); return { block: true, reason }; } // 其余 4xx(400 / 401 / 403 / 404 / 422…)同样永远不会因重试成功。 // // 这里原来一律「让位给本地决策」,而邮件驱动的 worker 没有 TUI —— // 让位等于守卫消失。生产实测:relay_key 过长报 400 被当暂时失败, // 那次 bash 在无人批准的情况下执行了(21:54:02 让位,同会话 // 21:54:24 另一次 key 正常,于是权限询问被随机吞掉)。 // // fail closed:宁可让模型看到「权限系统坏了」并自己改道, // 也不能悄悄放行一条没人看过的命令。 if (isPermanentFailure(e)) { const detail = describeError(e); log(`权限转发遇到永久失败(HTTP ${e?.status}),当场拒绝:${detail}`); return { block: true, reason: [ `无法把 ${event.toolName} 的授权请求送达给人类:${detail}`, '这是一个不会因重试而改变的失败(请求本身被服务端拒绝)。', '请改用不需要授权的方式完成,或在回信里说明需要人工执行哪一步。', ].join('\n'), }; } // 暂时失败(5xx / 408 / 429 / 网络层)→ 让位给 pi 本地决策(B-8.2)。 log(`权限转发暂时失败,让位给本地决策: ${describeError(e)}`); return; } log(`权限询问已发出(${event.toolName},key=${relayKey}),等待决策…`); send({ type: 'permission_pending', relayKey }); const decision = await new Promise((resolve) => pending.set(relayKey, resolve)); // fail closed(B-9.2 / N-9):只有明确同意才放行。 if (isApproval(decision)) { // 「一直同意」要真的记住,否则这个选项在骗人。判定交给 isAlwaysDecision —— // 「同意」是单次授权,把它当 always 会放行人没看过的后续命令。 if (isAlwaysDecision(decision)) { grants.add(event.toolName); send({ type: 'permission_grant', toolName: event.toolName }); log(`本会话的 ${event.toolName} 已获「一直同意」,后续不再询问`); } log(`权限 ${relayKey} 获批(${decision}),放行 ${event.toolName}`); return; } const extra = decidedExtra.get(relayKey) || {}; decidedExtra.delete(relayKey); return { block: true, reason: renderDecisionReason({ toolName: event.toolName, decision, note: extra.note, waiting: extra.waiting, }), }; }); }; } /** 把一次工具调用摘要成人能判断的文本(B-8.4)。 */ function describeToolCall(event) { const input = event?.input ?? {}; if (event.toolName === 'bash') return `命令:\n${String(input.command ?? '').slice(0, 800)}`; if (event.toolName === 'write' || event.toolName === 'edit') { return `文件:${input.file_path ?? input.path ?? '(未给出)'}`; } return JSON.stringify(input).slice(0, 800); } // ─── 会话装载 ─── /** * 找到这封邮件该落进的会话文件,装载它。 * * 三条路,优先级从高到低: * 1. 主进程给了 sessionFile —— 这条邮件会话之前有 worker 跑过,接着谈。 * 2. 服务端说接管了平台会话(人在 TUI 里开的那条)→ 按 platform id 找文件。 * 3. 都没有 → 新建,cwd 取地址的 path 位(B-3.1)。 * * 返回 `reused` 供提示词与模型降级判断:有历史的会话不重新自我介绍, * 也不做模型降级(换模型要换会话,会丢掉整条上下文,而上下文正是 * 发件人指定这条会话的原因)。 */ async function loadSession(mailTools) { const data = job.data; const given = job.session?.sessionFile; // ★ **复用判定消费父进程的结论**,不再自己 `existsSync` 一份(pi 评审 §三): // 父进程的判定与 worker 的判定各写一份时,`resolvedCwd` 就可能被交给一个 // 不消费它的分支 —— 那正是"回报了没用的 cwd"那个缺口的根子。 // 判定唯一的实现是 `src/turn-cwd.mjs` 的 `resolveSessionReuse`。 // 父进程没给(老协议)时才退回自己判一次(保底,不是常规路径)。 // ★ 父进程没给判定时(老协议 / 异常帧),**委托给唯一那份规则**判一次, // 而不是就地再写一个谓词(pi 评审 2026-09-15 §三:那份谓词比唯一规则**松** —— // 只看"文件在不在",不要求 `storedCwd`,于是"有会话文件但没有 cwd"这一支 // 会得出与唯一规则不同的结论)。 // // 这条路径**当前不可达**:`workerLaunch` 只有一个调用者,而它无条件注入 // `sessionReused`。但一旦有人新增第二个启动点,它立刻复活 —— // 所以它必须有信号(下面那行 log),否则将来复活时唯一的证据是"行为不对"。 const decidedReused = job.session?.sessionReused; if (decidedReused === undefined && given) { log('复用判定:父进程没给结论(老协议?),按唯一规则 resolveSessionReuse 就地判一次'); } const reuse = decidedReused === undefined ? resolveSessionReuse({ sessionFile: given, storedCwd: job.session?.cwd, exists: existsSync }) : { reused: Boolean(decidedReused) }; if (reuse.reused && given) { // cwd 取**父进程已经算好的那个**(`job.session.resolvedCwd`)—— // 父进程用它算沙箱的 `--rw`,worker 必须用同一个值,否则 rw 里可能没有 // 我们真正要写的目录(症状"界内也写不了")。见 src/turn-cwd.mjs 头注释。 const cwd = job.session?.resolvedCwd || job.session.cwd || resolveWorkspaceCwd( data.to_workspace, piMailFallback(data.session_id)).cwd; const opened = await openSession({ cwd, modelRuntime, customTools: mailTools, extension: permissionExtension(), sessionFile: given, }); return { ...opened, cwd, reused: true }; } const adoptID = adoptedSessionID(data); if (adoptID) { // 用 sessionScanner 而不是 `SessionManager.listAll()`:这里只要 id → path, // 而那两个字段全在会话文件的**首行** header 里。listAll 为了拿它们会把 // 每个 .jsonl 的每一行读进来并 JSON.parse,还把所有消息正文拼成一个大字符串 // (本机 115 个文件 / 145MB 实测:1431ms、堆里瞬时 240MB)。 // // worker 是短命进程,拿不到跨拍缓存的好处,但冷启动也依然便宜得多: // 没有任何一行巨大的 message 被 materialize(实测单行最长 2.63MB)。 const { createSessionScanner } = await import('./session-scan.mjs'); const { getAgentDir } = await import('@earendil-works/pi-coding-agent'); const all = await createSessionScanner({ sessionsDir: join(getAgentDir(), 'sessions'), }).scan(); const info = all.find((e) => e?.id === adoptID); if (!info?.path) { // 镜像是快照,可以过期。**不能**退回「新建一条」—— 那会让人在 TUI 里 // 看不到这封邮件带来的对话,而那正是接管的目的(N-8:静默改语义比报错糟)。 throw new Error(adoptMissingMessage(adoptID, '磁盘上已无这个会话文件')); } // cwd 取会话自己的(SessionInfo.cwd 来自持久化 header); // 老会话的 cwd 是空串,那种情况退回地址里的 path 位。 const { cwd } = resolveWorkspaceCwd( info.cwd || data.to_workspace, piMailFallback(data.session_id)); const opened = await openSession({ cwd, modelRuntime, customTools: mailTools, extension: permissionExtension(), sessionFile: info.path, }); log(`接管 pi 会话 ${adoptID}(cwd=${cwd},文件 ${info.path})`); return { ...opened, cwd, reused: true }; } // 目录不存在时**不创建**(N-2:笔误会在磁盘上落下真目录,而 Agent 在 // 里面一无所获),拒绝相对路径(N-3)。 const { cwd, grouped } = resolveWorkspaceCwd( data.to_workspace, piMailFallback(data.session_id)); if (!grouped && data.to_workspace) { log(`工作目录 ${data.to_workspace} 不可用,回退到 ${cwd}`); } // ★ 与父进程对齐:父进程已经按**同一个纯函数**算过这一轮的 cwd 并据此算了 // `--rw`。若两者不一致(理论上不该),取父进程那个 —— 因为 rw 是按它算的, // 用别的值就是"界内也写不了"。父进程没给时才用本地推导的结果。 const turnCwd = job.session?.resolvedCwd || cwd; if (job.session?.resolvedCwd && job.session.resolvedCwd !== cwd) { log(`注意:父进程定的 cwd=${job.session.resolvedCwd} 与本地推导的 ${cwd} 不同,按父进程的走`); } ensureCwd(turnCwd, grouped || Boolean(job.session?.resolvedCwd)); const opened = await openSession({ cwd: turnCwd, modelRuntime, customTools: mailTools, extension: permissionExtension(), }); log(`新建 pi 会话 ${opened.session.sessionId}(cwd=${turnCwd})`); // ★ 回报的必须是**实际用的那个**(`turnCwd`),不是本地推导的 `cwd`: // 这个返回值经 `session_opened` → `state.cwd`,而 `state.cwd` 正是下一轮 // `resolveTurnCwd` 的 `storedCwd`(也即下一轮 `--rw` 的输入)。 // 回报成 `cwd` 就等于让"记下来的必须是实际用的"这条不变量在这一支上不成立 // ⇒ 等式只在"本轮 rw vs 本轮 openSession"上闭合,没在"本轮 rw vs 下一轮 rw"上闭合 // (pi 评审 §三 抓到;起因是 `openSession` 用 `turnCwd` 而 `return` 用 `cwd`)。 return { ...opened, cwd: turnCwd, reused: false }; } // ─── 命名一致(C-11 / W-7)─── async function syncNaming(session, platformName) { const mailSessionID = job.data?.session_id; if (!mailSessionID) return; const plan = planNamingSync({ platformName, mailSubject: mailContext.subject, lastSynced: lastSyncedName, }); if (plan.skip) return; // 先记指纹再发请求:响应回来时 setSessionName 会再次触发 // session_info_changed,这一步是防自激循环的关键。 lastSyncedName = plan.signature; send({ type: 'name_synced', signature: plan.signature }); const res = await client.post(`/sessions/${mailSessionID}/sync`, { alias: plan.alias, title: plan.title, }); const back = planWriteBack({ finalAlias: res?.alias, currentPiName: session.sessionName }); log(`命名同步 ${piSessionId}: alias=${res?.alias || '(未变)'} 来源=${plan.source}`); if (back.write) { // 顺序要紧:先更新指纹,再改名。setSessionName **同步**触发 // session_info_changed(实测),本函数会被重入;指纹后更新的话重入那次 // 看到旧指纹,于是又打一次 sync —— 每条会话两次内容相同的请求。 lastSyncedName = `platform:${back.name}|${back.name}`; send({ type: 'name_synced', signature: lastSyncedName }); // 只用 setSessionName(走 pi 自己的写入路径)。绝不自己拼路径写会话文件: // 首条 assistant 消息落盘前文件还不存在,pi 首次落盘用 openSync(file,"wx"), // 抢先创建会让它抛 EEXIST(实测)。 session.setSessionName(back.name); log(`别名回写 pi:${back.name}(${back.reason})`); } } // ─── 自动转发(B-5)─── /** * 把这一轮的结论转回发件人。 * * 与主进程版本的差别:不再挂在 `agent_end` 订阅上,而是在 `runTurn` 返回后 * **确定性地**调一次。原来必须靠事件是因为主进程的 60 秒超时会先返回、 * 会话还在跑;worker 没有那个约束(它就为这封邮件活着),等真结束再转发。 * * @param {boolean} adopted 接管会话跳过命名同步 —— 它的别名是人从补全里 * 选中的平台 slug,同步会双向改坏:撞名时 Gateway 加后缀,定稿别名又回写进 * pi 会话文件,于是下次心跳上报的 slug 变成带后缀那个。实测撞出来过一次。 */ async function relaySummary(session, sessionManager, adopted) { if (!adopted) { await syncNaming(session, session.sessionName) .catch((e) => log(`命名同步失败: ${describeError(e)}`)); } // **只给人类来信自动转发**(详见 lib/relay-policy.js)。 // // 对方是 Agent 时它那边的插件也会自动回一封,于是两个模型都以为 // 「我只要把话说完就行」,实际在持续互相唤醒 —— 生产实测 pi 与 dsh // 客套 6 轮直到撞上 hop 上限。这一步在取文本之前判:早退比白做一轮清楚。 const policy = autoRelayDecision({ fromHuman: job.data?.from_human === true, replyTo: mailContext.replyTo, }); if (!policy.relay) { log(`本轮不自动转发:${policy.reason}`); return; } // 只取 type==='text' 的块(B-5.1 / N-6):thinking 是思考过程,不是结论。 const text = lastAssistantText(session.messages); if (!text) return; // 空文本不发空邮件(B-5.4) // 幂等键用 pi 会话 id + 会话树叶子 id:两者都落盘,重放也是同一个键。 const relayKey = relayKeyFor(piSessionId, sessionManager.getLeafId?.()); if (relayedKey === relayKey) return; // 模型这一轮已亲手回过这条线索 → 让位(B-5.3)。 // 否则收件箱里是两封说同一件事的邮件(生产实测过)。 if (shouldSkipAutoRelay(explicitSends.get(piSessionId), mailContext.replyTo, mailContext.mailID)) { relayedKey = relayKey; log(`本轮模型已主动回信 ${mailContext.replyTo},跳过自动转发`); return; } await client.post('/mail/send', { to: mailContext.replyTo, subject: replySubject(mailContext.subject), body: text, reply_to: mailContext.mailID || '', // relay + relay_key 走免配额通道(I-2):模型已经把话说完了,桥只是把它 // 搬到邮件里。对搬运收费会让配额用尽时 Agent 连交代都做不了。 relay: 'summary', relay_key: relayKey, }); relayedKey = relayKey; log(`已转发本轮总结给 ${mailContext.replyTo}(${text.length} 字)`); } // ─── 主流程 ─── async function run() { const data = job.data; const kind = job.kind; client = new GatewayClient({ url: job.config.gatewayURL, agentName: job.config.agentName, agentKey: job.config.agentKey, agentSecret: job.config.agentSecret, }); // ModelRuntime 每个 worker 建一次。allowModelNetwork 保持默认 false: // 上报给 Gateway 的是 getAvailable()(有凭证、真能调起来的), // 那取决于本机 auth.json,不取决于目录里有多少条。 modelRuntime = await ModelRuntime.create(); const runtimeErr = modelRuntime.getError?.(); if (runtimeErr) log(`模型运行时告警: ${runtimeErr}`); // connect_to_server 在 worker 里换了坐标要让主进程知道:worker 马上就退了, // 改在自己身上等于没改。主进程收到后重建 SSE 并写进后续 worker 的 job。 const mailTools = createMailTools({ // read_inbox 用它在服务端把列表收窄到自己这条会话。 getMailSessionId: () => mailContext.sessionID, client, log, agentName: job.config.agentName, onReconnect: () => send({ type: 'reconfigure', url: client.baseURL, agentKey: client.agentKey, }), }); const { session, sessionManager, diagnostics, cwd, reused } = await loadSession(mailTools); for (const d of diagnostics) log(`扩展诊断: ${d?.message ?? JSON.stringify(d)}`); piSessionId = session.sessionId; // 「这条会话是接管来的吗」以 Gateway 给的 platform_session_id 为准, // **不能**从 `reused && !job.session.sessionFile` 推断:第二封邮件进同一条 // 接管会话时主进程给了 sessionFile,那样推断会得出 false,于是命名同步跑起来 // 把人从补全里选的 slug 冲掉 —— 那正是上一轮修掉的那个 bug。 const adopted = Boolean(adoptedSessionID(data)); send({ type: 'session_opened', piSessionId, sessionFile: session.sessionFile || '', cwd, reused, }); // pi 侧改名(人在 TUI 里 /name)→ 同步给 Gateway。接管会话不同步,理由见 // relaySummary 的 adopted 参数注释。 session.subscribe((event) => { if (event?.type === 'session_info_changed' && !adopted) { syncNaming(session, event.name).catch((e) => log(`命名同步失败: ${describeError(e)}`)); } }); const prompt = buildMailPrompt({ agentName: job.config.agentName, data, kind, reused }); // 轮次超时给得很宽(默认 10 分钟,由主进程的 workerMaxMs 派生)。 // // 原来是 60 秒「超时按成功返回」,因为主进程要腾出手来收下一封邮件; // worker 没有这个理由 —— 它只为这封邮件活着,等真结论更准。带工具调用的 // 一轮跑几分钟很正常,60 秒返回会让转发落在一个还没说完的结论上。 const turnTimeout = job.config.turnTimeoutMs; if (reused) { // 续谈不做模型降级:换模型要换会话,会丢掉整条上下文 —— 而上下文正是 // 发件人指定这条会话的原因。但失败要可见。 const outcome = await runTurn(session, prompt, turnTimeout); log(`续谈 ${piSessionId}(mail ${data.mail_id}${outcome.queued ? ',已排队' : ''})`); if (!outcome.ok) { // ★ 续谈失败**也必须回一封失败信**,与新会话那条路一致(B-6)。 // // 实测缺口(2026-09-12):模型侧 402(余额不足)时,新会话那条路 // 会回「处理失败」,而续谈这条路只写了日志就 throw —— 发件人那边 // **什么也不会收到**(续谈不触发自动转发,因为会话里没有新的 // assistant 文本),而邮件驱动的会话没有本地界面可以看。 // // 复现条件很普通:往一条**已存在**的会话再发一封信,此时模型侧报错。 if (kind === 'mail' && data.from_name) { try { await client.post('/mail/send', { to: data.from_name, subject: `处理失败: ${data.subject || '(无主题)'}`, // 用续谈专用文案:共用那段说的是「范围内的模型都试过了」, // 而这条路**故意不降级**,照抄等于让人去调一个无效的旋钮。 body: renderResumeFailure(data.subject, outcome.error), reply_to: data.mail_id || '', relay: 'summary', relay_key: clampRelayKey(`model-failure:${data.mail_id || piSessionId}`), }); log(`已回报续谈失败给 ${data.from_name}`); } catch (e) { log(`失败回报也发不出去: ${describeError(e)}`); } } throw new Error(`续谈失败: ${outcome.error}`); } await relaySummary(session, sessionManager, adopted); return; } // 按管理员划定的范围逐个尝试(D-3)。 // 关键点:`prompt()` resolve **不代表模型跑成功了** —— 无凭证的 provider // 会让它 reject(实测 `No API key found for amazon-bedrock.`), // 而上游报错走 stopReason==='error'。判定交给 classifyTurnOutcome。 const attempts = modelAttemptOrder(job.config.allowedModels, { provider: job.config.replyProvider, model: job.config.replyModel, }); // 平台既没划范围、env 也没指定 → 这一轮的模型由**宿主**决定,平台无法控制。 // 这不该是静默的:实测它让每封来信都撞上宿主默认指向的一个没余额的直连 // provider,而日志读起来像「平台的模型没钱了」(见上面 label 的注释)。 if (attempts.length === 1 && !attempts[0]) { log( '注意:平台未指定模型(模型范围为空 且 AGENTMAIL_REPLY_MODEL 未设置),' + '本轮将使用**宿主的**默认模型,平台无法控制它是哪一个。' + '要固定下来请设 AGENTMAIL_REPLY_PROVIDER / AGENTMAIL_REPLY_MODEL,或在配置页划定范围。' ); } const failures = []; let live = { session, sessionManager }; for (const route of attempts) { // 标签要说出**语义**,不能写「平台默认」——那是错的,而且误导过。 // // `modelAttemptOrder` 在「平台未划范围 + env 未指定」时返回 `[undefined]`, // 意思是「交给宿主 SDK 用它自己的默认模型」。那是**宿主的**默认 // (pi 的 /root/.pi/agent/settings.json),平台管不着它。 // // 写成「平台默认」之后,日志里的 `模型 (平台默认) 失败: 402` 看起来像 // 「平台选的模型没钱了」,而真相是宿主默认指向了一个平台从未选用过的 // 直连 provider(实测:deepseek/deepseek-v4-flash,直连云、余额为零)。 const label = route ? `${route.provider}/${route.model}` : '宿主默认(平台未指定模型)'; if (route) { const model = modelRuntime.getModel(route.provider, route.model); if (!model) { // 目录里根本没有这个路由:同步就能判定,不必起一轮。 failures.push({ ...route, error: `平台目录里没有 ${label}` }); log(`模型 ${label} 不存在,跳过`); continue; } // 换模型要换会话:pi 的模型在 createAgentSession 时绑定。 // 上一次尝试失败的会话没有任何 assistant 消息,丢掉不损失内容。 try { live.session.dispose?.(); } catch { /* 已经没了 */ } const retried = await openSession({ cwd, modelRuntime, model, customTools: mailTools, extension: permissionExtension(), }); live = { session: retried.session, sessionManager: retried.sessionManager }; piSessionId = retried.session.sessionId; send({ type: 'session_opened', piSessionId, sessionFile: retried.session.sessionFile || '', cwd, reused: false, }); } const outcome = await runTurn(live.session, prompt, turnTimeout); if (outcome.ok) { if (failures.length) log(`${label} 成功(前 ${failures.length} 个失败)`); await relaySummary(live.session, live.sessionManager, false); return; } failures.push({ ...(route || {}), error: outcome.error }); log(`模型 ${label} 失败: ${outcome.error}`); } // 全部失败 → 必须回信(B-6):模型一次都没跑起来,会话里没有任何 assistant // 消息,自动转发因此什么也不会发 —— 发件人只会看到再无音讯。 if (kind === 'mail' && data.from_name) { try { await client.post('/mail/send', { to: data.from_name, subject: `处理失败: ${data.subject || '(无主题)'}`, body: renderFailureReport(failures, data.subject), reply_to: data.mail_id || '', relay: 'summary', relay_key: clampRelayKey(`model-failure:${data.mail_id || piSessionId}`), }); log(`已回报模型调用失败给 ${data.from_name}`); } catch (e) { log(`失败回报也发不出去: ${describeError(e)}`); } } // 发完仍要 throw(B-6.4):静默会让这次失败只存在于邮件里,日志上看不出来。 throw new Error(`范围内 ${failures.length} 个模型全部失败:${failures.map((f) => f.error).join(' | ')}`); } // ─── 入口 ─── process.on('message', (msg) => { if (msg?.type === 'job') { if (job) return; // 一个 worker 只接一封 job = msg; mailContext = { replyTo: msg.data?.from_name || '', subject: msg.data?.subject || '', mailID: msg.data?.mail_id || '', sessionID: msg.data?.session_id || '', permissionMode: msg.data?.permission_mode || 'workspace', }; lastSyncedName = msg.lastSyncedName || ''; for (const t of msg.grants || []) grants.add(t); run() .then(() => finish({ ok: true, error: '' })) .catch((e) => finish({ ok: false, error: describeError(e) })); return; } if (msg?.type === 'permission_decision') { const resolve = pending.get(msg.relayKey); if (!resolve) return; pending.delete(msg.relayKey); // 备注与「等人期间新到的邮件」按 relayKey 暂存,等那次工具调用回来渲染 // 拒绝理由时取走。不存全局变量:一个回合里可能有多次询问先后落地。 decidedExtra.set(msg.relayKey, { note: typeof msg.note === 'string' ? msg.note : '', waiting: Array.isArray(msg.freshMails) ? msg.freshMails : [], }); /* * ★★ 必须**最后**唤醒等待者,而且**必须唤醒**——这一行曾经不存在。 * * 顺序:上面的 `decidedExtra.set` 要先落地。hook 醒来后会去读它渲染拒绝理由, * 先 resolve 就会读到空备注(2026-09-13 那次修的正是这个:人写了 * 「我说了让你拉取仓库到 program 下你听不懂吗」,模型只看到「拒绝」, * 转头把同一条命令又问了一遍,连问 9 次)。 * * 但那次修**只补了备注,把唤醒弄丢了**:`const decision = await new Promise(...)` * (文件上方的 tool_call 钩子)永远等不到 resolve ⇒ 后果不是"备注少了", * 而是**整条授权链断掉**: * ① 工具调用挂着不动 → 一轮跑到 10 分钟 `TURN_TIMEOUT_MS` 才结束; * ② 桥把模型那半句开场白当"本轮总结"发回去(生产实测 59 字); * ③ 会话里留下一个**没有 toolResult 的 toolCall** → 下一轮 pi SDK 给它补一条 * `isError: true` 的「No result provided」→ 模型重试 bash → 人又被问一遍。 * 实测(webui4frpc 那条会话,2026-09-14):同一条会话一天被问了 6 次 bash, * 每一次都是人 8 秒内点了「同意」,而每一轮都恰好烧满 10 分钟。 * * 判据侧当时为什么没红:`permission-note.test.mjs` 的 WIRING 只钉了 * `decidedExtra.set(...)`(装饰)与 `renderDecisionReason(...)`, * **没有一条钉"唤醒"**。现在补上了。 */ resolve(msg.decision); return; } if (msg?.type === 'shutdown') { // fail closed(B-9.2 / N-9):唤醒所有未决询问,让 pi 侧那些 await 返回。 // 不唤醒的话会话挂死;而默认放行一个没人批准的危险操作更糟。 for (const [key, resolve] of pending) { log(`未决权限 ${key} fail closed`); resolve('shutdown'); } pending.clear(); } }); function finish(result) { if (finished) return; finished = true; send({ type: 'done', ...result }); // **一律 exit 0**:`done` 已经把成败说清楚了,用退出码再说一遍会让主进程 // 把「模型全部失败」这种已处理的结果也打成「worker 异常退出」。 // 非零退出码留给真正的崩溃(没来得及发 done 的那种)。 // // 给 IPC 一个 tick 把 done 送出去再退:未 flush 的消息会丢,而主进程靠它判成败。 // unref 不影响触发(IPC 通道本身在保持事件循环存活),只是不由这个计时器兜着。 const t = setTimeout(() => process.exit(0), 50); if (typeof t.unref === 'function') t.unref(); } // 未捕获异常也要回报:静默退出会让主进程只看到 exit code, // 而那条邮件的失败原因就此丢失。 process.on('uncaughtException', (e) => finish({ ok: false, error: `未捕获异常: ${describeError(e)}` })); process.on('unhandledRejection', (e) => finish({ ok: false, error: `未处理拒绝: ${describeError(e)}` })); send({ type: 'ready' });