#!/usr/bin/env node /** * ZCode 的 AgentMail 驱动:**收到来信 → 起一轮 ZCode → 把结论回信**。 * * 这是让 ZCode 成为一等 Agent 的那一半(另一半是插件:MCP 工具面 + 授权钩子)。 * * # 与另三个桥的关系 * * 结构对齐 pi / dsh / opencode 三桥:SSE 订阅 → 去重 → 解析工作目录与档位 → * 跑一轮 → 按策略回信 → 心跳。可复用的部分一律走 `lib/`(逐字节同源): * 事件补投、工作目录解析、回信策略、去重判据、SSE 帧解析。 * * 差别只在「怎么跑一轮」:ZCode 用 **headless CLI** * (`--prompt … --output-format stream-json`),不是 SDK。 * * # 三个必须记住的约束 * * 1. **`--mode` 必传**。`--prompt` 的默认 mode 是 `yolo`,而 yolo 会绕过全部 * 权限询问 —— 授权钩子根本不会触发,授权系统会**静默消失**(不报错, * 只是没有任何询问)。档位映射见 `src/turn-mode.mjs`。 * 2. **失败必须回信**。邮件驱动的会话没有本地界面,一轮跑不起来而什么都不发, * 发件人只会觉得「信发出去了,然后再无音讯」。 * 3. **模型自己发过信就不再自动转发**。工具跑在 ZCode 派生的 MCP 服务器进程里, * 与驱动不是同一个进程,所以经 `lib/explicit-sends.mjs` 落盘对齐。 * * # 串行 * * 一轮一次。ZCode 的会话与工作目录是重资源,同一目录并发跑两轮会互相踩; * 代价是一封长信会挡住后面的信 —— 这是显式取舍,不是遗漏(见 README 的已知缺口)。 */ import { basename, join } from 'node:path'; import { homedir } from 'node:os'; import { readFileSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import { GatewayClient, GatewayError } from '../lib/gateway.mjs'; import { createSSEClient } from '../lib/sse-client.js'; import { BoundedSet, BoundedMap, MAX_TRACKED_MAILS, MAX_TRACKED_SESSIONS } from '../lib/bounded.js'; import { selectCatchup } from '../lib/catchup.js'; import { autoRelayDecision } from '../lib/relay-policy.js'; import { shouldSkipAutoRelay } from '../lib/relay-dedup.js'; import { resolveWorkspaceCwd, ensureCwd } from '../lib/workspace.js'; import { normalizeMode } from '../lib/permission-mode.js'; import { clampRelayKey } from '../lib/relay-key.js'; import { explicitSendsFile, readExplicitSends } from '../lib/explicit-sends.mjs'; import { zcodeModeForTier, modeReachesPermissionHook, describeTier } from './turn-mode.mjs'; import { buildMailPrompt, replySubject, renderTurnFailure } from './prompt.mjs'; import { runTurn, DEFAULT_CLI } from './zcode-run.mjs'; const log = (...parts) => console.error('[zcode-mail-bridge]', ...parts); const CONFIG = { gatewayURL: process.env.AGENTMAIL_GATEWAY_URL || 'http://127.0.0.1:8180', agentName: process.env.AGENTMAIL_AGENT_NAME || 'zcode', turnTimeoutMs: Number(process.env.AGENTMAIL_TURN_TIMEOUT_MS || 20 * 60 * 1000), maxTurns: Number(process.env.AGENTMAIL_MAX_TURNS || 0) || undefined, workspaceRoot: process.env.AGENTMAIL_WORKSPACE_ROOT || '', cliPath: process.env.AGENTMAIL_ZCODE_CLI || DEFAULT_CLI }; /** * 没有 `to_workspace` 时的兜底目录。 * * **不能用共用的 `mailSessionFallback`** —— 那个函数的目录名写的是 `~/.dsh` * (它注释里也写明是「没有天然兜底的平台(DSH)用这个」)。各平台的会话存储 * 各不相同,把 ZCode 的会话塞进 `~/.dsh` 下会造成两个平台的会话目录互相污染。 * * @param {string} sessionKey */ export function zcodeSessionFallback(sessionKey, rootOverride) { const root = rootOverride || CONFIG.workspaceRoot; if (root) return join(root, String(sessionKey || 'default')); return join(homedir(), '.zcode', 'mail-sessions', String(sessionKey || 'default')); } /** * 自报给网关的强制力。 * * **只声明得出来的事**:ZCode 上的档位强制全部靠插件里的 PermissionRequest 钩子, * 钩子没被注册(插件没启用 / 被禁 / 清单被改坏)时我们什么也拦不住, * 那时还报 native 就是在替一个不存在的能力背书 —— 而这个声明的用途正是 * 让人相信「这一档在这里是被强制的」。 * * 查的是插件自己的 `hooks/hooks.json`(驱动就住在这个插件里), * 不需要额外的配置项。 */ export function detectModeEnforcement({ hooksFile } = {}) { const file = hooksFile || fileURLToPath(new URL('../hooks/hooks.json', import.meta.url)); try { const cfg = JSON.parse(readFileSync(file, 'utf8')); const entries = cfg?.hooks?.PermissionRequest; if (Array.isArray(entries) && entries.some(e => Array.isArray(e?.hooks) && e.hooks.length > 0)) { return { enforcement: 'native', reason: `钩子已注册(${file})` }; } return { enforcement: 'advisory', reason: `钩子清单里没有 PermissionRequest(${file})` }; } catch (e) { return { enforcement: 'advisory', reason: `读不到钩子清单(${file}):${e?.message || e}` }; } } /** 描述错误:把「网关可达但返回 4xx」与「连不上」分开 —— 两者的应对完全不同。 */ export function describeError(e) { if (e instanceof GatewayError) { return `网关返回 HTTP ${e.status}(${e.path}):${typeof e.body === 'string' ? e.body : e.message}`; } return e?.message || String(e); } /** * 造一个驱动实例。 * * 依赖全部注入,所以整条流水线(事件 → 提示词 → 一轮 → 回信判定 → 发信载荷) * 可以在没有模型、没有 ZCode 的情况下被端到端断言。 */ export function createDriver({ client, runTurnFn = runTurn, logFn = log, env = process.env, config } = {}) { // 配置可覆盖:测试需要把工作目录指到临时目录,不能碰真实的家目录。 const CFG = { ...CONFIG, ...(config || {}) }; const delivered = new BoundedSet(MAX_TRACKED_MAILS); /** AgentMail 会话 id → { zcodeSessionId, cwd, tier, turns } */ const sessions = new BoundedMap(MAX_TRACKED_SESSIONS); const queue = []; let running = false; /** 当前在途回合的杀进程函数(关停时要终止它,否则会留下跑工具的孤儿)。 */ let currentKill = null; /** 注册表只用于日志与自检:它让「为什么一轮授权询问都没发生」有据可查。 */ const stats = { turns: 0, relays: 0, skippedRelay: 0, failures: 0 }; function resolveCwd(data) { const sessionId = data?.session_id || data?.mail_id || 'unknown'; const fallback = zcodeSessionFallback(sessionId, CFG.workspaceRoot); const { cwd, grouped } = resolveWorkspaceCwd(data?.to_workspace, fallback); if (!grouped) logFn(`会话 ${sessionId} 没有可用的 to_workspace,用兜底目录 ${cwd}`); ensureCwd(cwd, grouped); return cwd; } async function relay({ data, text, kind }) { const fromHuman = data?.from_human === true; const decision = autoRelayDecision({ fromHuman, replyTo: data?.from_name }); if (!decision.relay) { logFn(`不自动转发(${decision.reason})`); stats.skippedRelay++; return false; } const sessionId = data?.session_id || ''; const sent = readExplicitSends(explicitSendsFile(env), { sessionId, // 只认本轮之后的记录:早于本轮的发信属于上一次往返,不该让这一轮沉默。 since: Date.now() - CFG.turnTimeoutMs }); if (shouldSkipAutoRelay(sent, data.from_name, data.mail_id)) { logFn(`本轮模型已主动回信 ${data.from_name},跳过自动转发`); stats.skippedRelay++; return false; } // relay + relay_key 走免配额通道:模型已经把话说完了,驱动只是把它搬进邮件。 // 对搬运收配额会让「配额用尽」变成「连交代都做不到」。 const relayKey = clampRelayKey(`zcode:${data.mail_id || kind}`); await client.post('/mail/send', { to: data.from_name, subject: replySubject(data.subject), body: text, reply_to: data.mail_id || '', relay: 'summary', relay_key: relayKey }); stats.relays++; logFn(`已回信给 ${data.from_name}(${text.length} 字)`); return true; } async function processMail(data) { const sessionId = data?.session_id || ''; const tier = normalizeMode(data?.permission_mode); const mode = zcodeModeForTier(tier); const cwd = resolveCwd(data); const prev = sessions.get(sessionId); const resume = prev?.zcodeSessionId || ''; logFn(`处理 ${data.mail_id}|${describeTier(tier, mode)}|cwd=${cwd}${resume ? `|续会话 ${resume}` : ''}`); if (!modeReachesPermissionHook(mode) && tier !== 'plan') { // 只有 full 档会走到这里,且是刻意的。写日志是因为「没有权限询问」 // 在 yolo 下是预期行为,在 build 下则是缺陷 —— 两者必须能区分。 logFn(`注意:--mode ${mode} 不会产生权限询问(本档如此设计)`); } const prompt = buildMailPrompt({ agentName: CONFIG.agentName, data }); const outcome = await runTurnFn( { prompt, cwd, mode, maxTurns: CFG.maxTurns, resumeSessionId: resume || undefined, turnTimeoutMs: CFG.turnTimeoutMs, cliPath: CFG.cliPath, // 注入给 ZCode 进程(→ 继承给插件、钩子、MCP 服务器): // 授权钩子靠 AGENTMAIL_SESSION_ID 判断「有没有本地界面」, // 靠 AGENTMAIL_PERMISSION_MODE 决定档位。 env: { AGENTMAIL_SESSION_ID: sessionId, AGENTMAIL_PERMISSION_MODE: tier, AGENTMAIL_MAIL_SUBJECT: data?.subject || '', AGENTMAIL_REPLY_TO: data?.mail_id || '' } }, { log: logFn, onChild: kill => { currentKill = kill; } } ); currentKill = null; stats.turns++; if (outcome.sessionId) { sessions.set(sessionId, { zcodeSessionId: outcome.sessionId, cwd, tier, turns: (prev?.turns || 0) + 1 }); } const failed = outcome.timedOut || (outcome.exitCode !== 0 && !outcome.response); if (failed) { stats.failures++; const reason = outcome.timedOut ? `回合超时(${Math.round(CFG.turnTimeoutMs / 1000)} 秒),已终止进程树` : `ZCode 退出码 ${outcome.exitCode}${outcome.stderrTail ? `:\n${outcome.stderrTail}` : ''}`; logFn(`一轮失败:${reason}`); // 失败必须回信:否则发件人只看到「信发出去了,然后再无音讯」。 try { await client.post('/mail/send', { to: data?.from_name, subject: `处理失败: ${data?.subject || '(无主题)'}`, body: renderTurnFailure([{ kind: outcome.timedOut ? '超时' : 'CLI 失败', error: reason }], data?.subject), reply_to: data?.mail_id || '', relay: 'summary', relay_key: clampRelayKey(`zcode-failure:${data?.mail_id || sessionId}`) }); } catch (e) { logFn(`失败回报也发不出去:${describeError(e)}`); } return { ok: false, reason }; } const text = String(outcome.response || '').trim(); if (!text) { // 退出码 0 但没有最终文本:常见于模型只调了工具就结束。 // 这时**不冒充**回信(会让收件人以为模型什么都没做),但要留下日志。 logFn('这一轮没有产出最终文本,不自动回信(若模型自己发过信,那封就是答复)'); return { ok: true, relayed: false }; } return { ok: true, relayed: await relay({ data, text, kind: 'mail' }) }; } async function drain() { if (running) return; running = true; try { while (queue.length) { const data = queue.shift(); try { await processMail(data); } catch (e) { // 一封邮件处理崩了不能把驱动带走:后面还有很多信。 logFn(`处理 ${data?.mail_id} 时异常:${describeError(e)}`); } } } finally { running = false; } } /** SSE 事件入口。必须廉价 —— 它跑在读循环上。 */ function handleEvent(type, data) { if (type !== 'new_mail') return; if (data?.role && data.role !== 'to' && data.role !== 'cc') return; const id = data?.mail_id; if (!id || delivered.has(id)) return; delivered.add(id); queue.push(data); void drain(); } async function catchUp(pendingMails) { const mails = selectCatchup(pendingMails, delivered); if (!mails.length) return 0; logFn(`补投 ${mails.length} 封停机期间到达的邮件`); for (const m of mails) handleEvent('new_mail', m.data ?? m); return mails.length; } return { handleEvent, catchUp, processMail, stats, sessions, delivered, /** * 关停:终止在途回合。 * * 不做这件事的后果是——systemd 杀掉驱动之后,那个 ZCode 进程还在跑工具, * 而既没有驱动看着它,也没有本地界面看着它。宁可丢掉这一轮的工作。 */ abort() { // 先取后清:杀过就算完,重复关停(SIGTERM 后再来一个)不该重复杀。 const kill = currentKill; currentKill = null; if (kill) { logFn('关停:终止在途的 ZCode 回合'); try { kill('SIGTERM'); } catch { /* 已经结束了 */ } } queue.length = 0; } }; } // ─── 真实入口 ─────────────────────────────────────────────────────── async function main() { const client = new GatewayClient(process.env); const missing = client.checkConfig(); if (missing.length) { log(`配置不完整,缺少 ${missing.join('、')};驱动不会启动(静默启动会让信永远没人处理)`); process.exit(1); } const driver = createDriver({ client, logFn: log }); let caughtUp = false; let timer; try { await client.register(); log(`已接入 ${client.baseURL},身份 ${client.agentName}`); } catch (e) { // 密钥未登记时说清该做什么,别只留一句 401。 log(`注册失败:${describeError(e)}`); log('若提示密钥无效,请让管理员在 AgentMail 后台登记这把密钥。'); } const beat = async () => { try { // mode_enforcement 只声明得出来的事:挡得住工具的是插件里的授权钩子, // 钩子没注册时我们什么也拦不住(见 detectModeEnforcement)。 const res = await client.post('/agent/heartbeat', { mode_enforcement: detectModeEnforcement().enforcement }); if (!caughtUp) { caughtUp = true; await driver.catchUp(res?.pending_mails); } } catch { // 心跳失败不刷错误日志:真连不上时网关会把它判成离线,那才是可见信号。 } }; await beat(); timer = setInterval(beat, 30_000); createSSEClient({ authHeaders: () => client.authHeaders(), baseURL: client.baseURL, path: '/api/v1/events/stream', log, onEvent: (type, data) => driver.handleEvent(type, data) }); const shutdown = reason => { log(`收到 ${reason},关停中…(已处理 ${driver.stats.turns} 轮,回信 ${driver.stats.relays} 封)`); if (timer) clearInterval(timer); driver.abort(); client.stopSSE?.(); // 给杀进程留一点时间再退:自己先死会把 ZCode 变成孤儿。 setTimeout(() => process.exit(0), 1200); }; for (const sig of ['SIGINT', 'SIGTERM']) process.on(sig, () => shutdown(sig)); const enforcement = detectModeEnforcement(); log(`档位强制力自报:${enforcement.enforcement}(${enforcement.reason})`); log(`驱动就绪:CLI ${basename(CONFIG.cliPath)},回合上限 ${Math.round(CONFIG.turnTimeoutMs / 1000)} 秒`); } // 直接执行时启动;被 import 时只导出(测试要用 createDriver)。 const isDirect = process.argv[1] && import.meta.url === `file://${process.argv[1]}`; if (isDirect) { main().catch(e => { log(`启动失败:${describeError(e)}`); process.exit(1); }); }