MailUI4Agents/plugins/pi-mail-bridge/src/index.mjs

#!/usr/bin/env node
/**
 * AgentMail ↔ pi 桥（pi-mail-bridge）
 *
 * 形态是**常驻守护进程**，不是 pi 扩展。原因见 src/session-pool.mjs 顶部：
 * 扩展被加载进一条已存在的会话，cwd 由启动 pi 的人决定；而 B-3.1 要求每封邮件的
 * to_workspace 成为会话 cwd。
 *
 * # 进程结构
 *
 * 主进程**只做 I/O 与调度**：SSE、心跳、去重、把邮件派给子进程。模型工作全部
 * 下到 `src/worker.mjs`（一封邮件一个进程，跑完就退），由 `src/pool.mjs` 调度。
 *
 * 这不是为了并行度，是为了**不阻塞事件循环**。pi 的会话装载是同步的：
 * `SessionManager.open()` 走 `openSync` + `readSync` 循环把整个 `.jsonl` 读进内存
 * 并逐行 JSON.parse。实测本机最大那条会话 23MB，`open` 一次阻塞事件循环 118ms；
 * 模型跑起来之后 SDK 内部还有更多同步工作。原来这些都在主线程上 —— SSE 读循环
 * 在那期间完全停住，后续邮件卡在 TCP 缓冲区，久到 Gateway 认为连接死了，
 * 重连又触发重放。实测同样的活在 fork 出的子进程里跑，主进程阻塞 0ms。
 *
 * 权限询问期间的挂起也随之只影响那一个 worker：原来 `await new Promise(...)`
 * 等人做决定，整座桥在那段时间不再收信。
 *
 * 契约实现对照（docs/PLUGIN-CONTRACT.md）：
 *   B-1 启动     → main()
 *   B-2 心跳     → beat()，30 秒
 *   B-3 new_mail → pool.submit()（投递本体在 worker.mjs）
 *   B-4 决策     → handlePermissionDecision()
 *   B-5 转发     → worker.mjs 的 relaySummary()
 *   B-6 失败回信 → worker.mjs 末尾的 renderFailureReport
 *   B-7 补拉     → catchUp()
 *   B-8 权限     → worker.mjs 的 permissionExtension()
 *   B-9 关停     → shutdown()
 */

import { mkdirSync, openSync, closeSync, unlinkSync, readFileSync, writeFileSync } from 'node:fs';
import { homedir } from 'node:os';
import { join } from 'node:path';
import { ModelRuntime, getAgentDir } from '@earendil-works/pi-coding-agent';

import { GatewayClient, readLocalKey, generateLocalKey, saveConfig } from './gateway.mjs';
import { createWorkerPool } from './pool.mjs';
import { createSessionScanner } from './session-scan.mjs';
import { describeError } from './turn.mjs';
import { BoundedSet, MAX_TRACKED_MAILS } from '../lib/bounded.js';
import { snapshotPiModels } from '../lib/model-scope.js';
import { snapshotPiSessions } from '../lib/session-snapshot.js';
import { selectCatchup } from '../lib/catchup.js';
import { selectWaitingMails } from '../lib/waiting-mails.js';

// ─── 配置 ───

const GATEWAY_URL = process.env.AGENTMAIL_GATEWAY_URL || 'http://127.0.0.1:8180';
const AGENT_NAME = process.env.AGENTMAIL_AGENT_NAME || 'pi';
const AGENT_SECRET = process.env.AGENTMAIL_AGENT_SECRET || '';
const REPLY_PROVIDER = process.env.AGENTMAIL_REPLY_PROVIDER || '';
const REPLY_MODEL = process.env.AGENTMAIL_REPLY_MODEL || '';

/**
 * 单轮超时。
 *
 * 从 60 秒放宽到 10 分钟：60 秒那个数字是「主进程要腾出手来收下一封邮件」的
 * 产物 —— 超时按成功返回，好让 deliverMail 早点结束。worker 没有这个理由，
 * 它只为这封邮件活着，等真结论更准。带工具调用的一轮跑几分钟很正常，
 * 60 秒返回会让转发落在一个还没说完的结论上。
 */
const TURN_TIMEOUT_MS = Number(process.env.AGENTMAIL_TURN_TIMEOUT_MS || 600_000);

/**
 * 并发上限。
 *
 * 每个 worker 约 140MB RSS（实测），并且每个都会对上游 provider 发请求。
 * 3 是内存与吞吐的折中；同一条会话无论如何都是串行的（见 pool.mjs）。
 */
const MAX_WORKERS = Number(process.env.AGENTMAIL_MAX_WORKERS || 3);

/**
 * worker 硬超时。
 *
 * 比轮次超时留出余量：正常情况下 worker 自己会在轮次超时后收尾退出，
 * 这个数字兜的是「连收尾都没做」（进程卡死、权限等不到决策而决策事件也丢了）。
 * 到点 SIGKILL —— 否则那条会话的后续邮件永远排队。
 */
const WORKER_MAX_MS = Number(process.env.AGENTMAIL_WORKER_MAX_MS || TURN_TIMEOUT_MS + 120_000);

const LOCK_FILE = join(process.env.AGENTMAIL_CONFIG_DIR || join(homedir(), '.agentmail'), 'pi-bridge.lock');

/** 日志一律 console.error：它一定进 journalctl（契约 9.8）。 */
const log = (...args) => console.error('[pi-mail-bridge]', ...args);

// ─── 进程内状态 ───
//
// 主进程只留「调度需要的」那几样，全部在内存（重启即丢，契约第六节的已知取舍）。
// 会话映射、权限授权、命名指纹都下沉到 pool 里按邮件会话存 —— 主进程不再持有
// AgentSession 对象（那东西跨不了进程边界）。

// 已投过的 mail_id（SSE 与补拉共用，B-7.3）。
//
// 有界：桥是守护进程，跑几十天下来这里会攒下每一封处理过的邮件 id 而永远
// 没有出口。淘汰是安全的 —— 它防的两种重复（心跳与 SSE 建连之间的窗口、
// SSE 断线重放）都发生在秒到分钟级，几千封之前的 id 不可能再来。
const deliveredMails = new BoundedSet(MAX_TRACKED_MAILS);

/*
★ 2026-09-26（用户报的）：**投递即标已读**。

# 缺陷：桥投了信却不动 status ⇒ 每次重启都重投一遍 ⇒ 回声

投递路径（SSE `new_mail` / 心跳补投）只做两件事：起 worker、把 id 记进
`deliveredMails`。**没有任何一处调 `/mail/read`** —— 桥里唯一那处标已读
在 `read_inbox` 工具里，要等模型自己去读收件箱。

于是每封被投递的信**永远是 unread**；而 `catchUp` 按 `status=unread` 拉。
⇒ 桥一重启（每次部署都会），积压的"未读"被当成离线漏投**再投一遍**。

实测（2026-09-26）：
  · 5 个 mail_id 各出现在**两条不同 pi 会话**里（04:54 一条、08:01 一条）
  · 同一封信被投两次 ⇒ 两个 worker 各回一封 ⇒ 对方收到两封 ⇒ 各回两封…
  · pi 的收件箱 287 封 unread 中 **187 封已经回过信了**
    （`EXISTS(SELECT 1 FROM mails r WHERE r.parent_mail_id=m.mail_id)`）
  · 两条会话各烧到 463 / 268 封，用户原话：「我都不记得我下达这个任务，
    是你的桥自动重投存在 bug」「就是你的错误的重投机制造成了回声」

# 修法：把"投递"与"已读"合成一个动作

`deliveredMails` 是**内存**里的去重集合，重启即丢。数据库的 `status` 才是
跨重启的那个"我接管过了"记录 —— 两者必须同时写，否则口径不一致：

  内存说"投过了"、库说"没读过" ⇒ 重启后桥只认库 ⇒ 重投。

因此凡是把一封标记进 `deliveredMails` 的地方，都要同时把它在**库里**标成已读。
这就是 `markDelivered` 存在的理由：不漏一处（漏一处就是一条重投路径）。

# 为什么标已读不会"吞掉"信

标已读只改 status，不改内容、不删行；`read_inbox` 传 `status=all` 依然看得到。
而"这封信已经被交给一个 worker 处理了"正是它此刻的真实状态 —— 库里记的
本来就该是这件事，而不是"模型有没有顺手调过 read_inbox"。

# 失败不阻塞

标不上不该让投递失败（正文已经在 worker 手里，代价只是下次重启可能再投一次，
与旧行为一致、不更差）。写日志，好让"标已读坏了"这件事可见。
*/
function markDelivered(mailId) {
  if (!mailId) return;
  deliveredMails.add(mailId);
  client?.post('/mail/read', { mail_ids: [mailId] }).catch((e) =>
    log(`投递标已读失败 ${mailId}: ${e?.message || e}`));
}

let allowedModels = [];
let modelRuntime = null;
let client = null;
let pool = null;
let sessionScanner = null;
let heartbeatTimer = null;
let shuttingDown = false;

// ─── 单实例锁 ───
//
// 两个桥同时跑的后果不是「慢一点」而是错的：两条 SSE 各收到同一封邮件，
// 各起一条 pi 会话，发件人收到两封回信；而 deliveredMails 在各自内存里，去重不了。

function acquireLock() {
  mkdirSync(join(LOCK_FILE, '..'), { recursive: true, mode: 0o700 });
  try {
    // O_EXCL 原子创建。存在则说明有别的实例（或上次崩溃留下的陈锁）。
    const fd = openSync(LOCK_FILE, 'wx');
    writeFileSync(fd, String(process.pid));
    closeSync(fd);
    return true;
  } catch (e) {
    if (e?.code !== 'EEXIST') throw e;
  }
  // 陈锁判定：文件里的 pid 还活着吗
  let pid = 0;
  try { pid = Number(readFileSync(LOCK_FILE, 'utf8').trim()); } catch { /* 读不到当陈锁 */ }
  if (pid > 0) {
    try {
      // signal 0 只探测存在性，不真的发信号
      process.kill(pid, 0);
      log(`已有实例在运行（pid ${pid}），本进程退出。`);
      return false;
    } catch {
      // ESRCH：进程没了，是陈锁
    }
  }
  log(`清理陈锁 ${LOCK_FILE}（原 pid ${pid || '未知'} 已不存在）`);
  try { unlinkSync(LOCK_FILE); } catch { /* 竞态下别人清掉了也行 */ }
  return acquireLock();
}

function releaseLock() {
  try {
    // 只删自己的锁：pid 不符说明这把锁已被别的实例接管
    if (Number(readFileSync(LOCK_FILE, 'utf8').trim()) === process.pid) unlinkSync(LOCK_FILE);
  } catch { /* 已经没了 */ }
}

// ─── 权限决策回来（B-4）───

/**
 * 把决策路由给发起询问的那个 worker。
 *
 * 找不到 worker 有两种情形，都不该新开会话（B-4.3）：
 *   - 桥重启了：那次工具调用早已随进程消失。但人刚刚点了「同意」——
 *     什么都不做的话人以为自己批准了、Agent 却毫无反应，所以退化为把决策
 *     当一封通知投进原会话（B-4.2）。
 *   - 那条邮件会话从没被处理过：连通知都无处可投，只能记一行日志。
 */
function handlePermissionDecision(data) {
  // 决策回执的内容马上随这次恢复交给 worker（备注 + 等人期间新到的邮件都在里面），
  // 所以先把它记成"已交付"：稍后那封同内容的邮件就不会被当成新任务再起一轮。
  const decisionMailID = data?.decision_mail_id || '';
  if (decisionMailID) markDelivered(decisionMailID);

  // 拉"等人期间新到的邮件"要发一次 HTTP，而本函数跑在 SSE 读循环上（必须廉价）
  // —— 所以整体转异步：先把事件收下，几毫秒后带着上下文去唤醒 worker。
  routeDecision(data).catch((e) => log(`决策投递失败: ${describeError(e)}`));
}

/**
 * 把决策连同**人类说了什么**一起交给挂起的 worker。
 *
 * 缺了 `note` 就是这次线上缺陷：人类写「我说了让你拉取仓库到program下你听不懂吗」，
 * 模型只收到「用户拒绝了这次 bash 调用」，于是把同一条命令改个写法又问一遍（连问 9 次）。
 */
async function routeDecision(data) {
  const relayKey = data?.relay_key || '';
  const decision = String(data?.decision || '拒绝');
  const note = typeof data?.note === 'string' ? data.note : '';
  const sessionID = data?.session_id || '';
  /*
    ★ 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`
      + (note ? '，带备注' : '')
      + (waiting.length ? `，并附上等人期间新到的 ${waiting.length} 封邮件` : ''));
    return;
  }

  if (!sessionID || !pool.hasSession(sessionID)) {
    // **不得凭空新开会话**（B-4.3）
    log(`权限决策 ${relayKey} 无对应会话，忽略`);
    return;
  }
  log(`权限 ${relayKey} 无挂起项，退化为通知投递`);
  pool.submit('permission', { ...data, note });
}

/**
 * 等人点头期间新到的邮件（人类很可能在这期间补一句更正）。
 *
 * 拉取失败**不能拖垮决策投递**：备注还在，最坏情况是模型晚一步看到更正。
 */
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&workspace=${encodeURIComponent(workspace)}`,
    );
    return selectWaitingMails(box, { sessionID, seen: deliveredMails });
  } catch (e) {
    log(`等人期间的邮件拉取失败（不影响决策投递）: ${describeError(e)}`);
    return [];
  }
}

// ─── 心跳（B-2）───

async function reportSessions() {
  try {
    // **不用 `SessionManager.listAll()`**：它为了拿 id/cwd/name/modified 四个
    // 字段，把 ~/.pi/agent/sessions 下每个 .jsonl 的每一行都读进来并 JSON.parse，
    // 还把所有消息正文拼成一个 allMessagesText 大字符串。本机实测（115 个文件 /
    // 145MB）单次 1431ms、堆里瞬时 240MB —— 而这 282MB 每 30 秒分配一次随即
    // 变成垃圾，且那 1.4 秒是同步解析，跑在事件循环上（SSE 读循环那期间停着）。
    //
    // sessionScanner 只读 header 的首行 + 增量扫尾部找 session_info：
    // 稳态下未变化的文件一个字节都不读（实测 3ms / 0 字节）。
    const all = await sessionScanner.scan();
    const driven = pool.mailDrivenIDs();
    return snapshotPiSessions(all, (id) => driven.has(id));
  } catch (e) {
    // 拉不到就**省略字段**而不是传 []（N-7 / W-3）：
    // 空数组的语义是「平台确实一条会话都没有」，会把服务端镜像抹掉。
    log(`会话列表读取失败: ${describeError(e)}`);
    return undefined;
  }
}

async function reportModels() {
  try {
    // getAvailable 而不是 getModels：后者本机有 1221 条，其中真能调起来的只有 1 条。
    // 上报目录的全部意义就是让管理员别选中一个注定失败的路由。
    const available = await modelRuntime.getAvailable();
    return snapshotPiModels(available);
  } catch (e) {
    log(`模型目录读取失败: ${describeError(e)}`);
    return undefined;
  }
}

/**
 * 补投离线期间积压的未读邮件（B-7）。
 *
 * SSE 只推连上之后的事件，插件重启前发来的邮件不会再推一次。
 *
 * 与旧版的差别：**不再 await 每一封**。旧版串行是因为「每封都要起一轮模型，
 * 并发放出去等于对上游打 N 个并发请求」—— 那个约束现在由 pool 的 maxWorkers
 * 承担，而且它比串行更好：同一条会话仍然串行，不同会话可以并行。
 */
/**
 * 补投离线期间的邮件。
 *
 * ★★ 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;
  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）
        markDelivered(ev.mail_id);
        pool.submit('mail', ev);
        delivered += 1;
      }
    } catch (e) {
      // 单个工作区失败不影响其余：与"心跳失败不报错"同一原则。
      log(`补投工作区 ${ws} 失败（不影响其余）: ${describeError(e)}`);
    }
  }
  if (delivered) log(`补投 ${delivered} 封离线期间的邮件（共 ${pending} 封未读，跨 ${list.length} 个工作区）`);
}

// ─── 启动 / 关停 ───

async function main() {
  if (!acquireLock()) process.exit(0);

  // B-1.1：环境变量 → ~/.agentmail/agent.key → 本地生成并打印全文
  let agentKey = process.env.AGENTMAIL_AGENT_KEY || readLocalKey();
  if (!agentKey && !AGENT_SECRET) agentKey = generateLocalKey(log);

  client = new GatewayClient({
    url: GATEWAY_URL,
    agentName: AGENT_NAME,
    agentKey,
    agentSecret: AGENT_SECRET,
  });

  // ModelRuntime 主进程也要一个：心跳的 reportModels 用它。worker 各自再建
  // 一个（跨进程传不了），代价是每个 worker 多 ~20ms（实测 11–23ms）。
  //
  // allowModelNetwork 保持默认的 false：桥启动时不去网上拉模型目录。
  // 拉了也没用 —— 上报给 Gateway 的是 getAvailable()（有凭证、真能调起来的），
  // 而那取决于本机 auth.json，不取决于目录里有多少条。开着只会让
  // 启动多等一个网络往返，而且断网时启动路径上多一个可失败点。
  modelRuntime = await ModelRuntime.create();
  const runtimeErr = modelRuntime.getError?.();
  if (runtimeErr) log(`模型运行时告警: ${runtimeErr}`);

  // 会话目录扫描器。**必须建一次并复用** —— 它的省内存全靠跨拍存活的
  // size 缓存（稳态下未变化的文件一个字节都不读）。每拍新建一个等于
  // 每拍都冷启动，退回 listAll 那种全量读的开销。
  //
  // 路径自己拼而不是 import getSessionsDir：SDK 只导出 getAgentDir，
  // getSessionsDir 是内部函数（dist/config.js 里 `join(getAgentDir(), "sessions")`）。
  sessionScanner = createSessionScanner({
    sessionsDir: join(getAgentDir(), 'sessions'),
  });

  // 工作进程池。config() 每次派活时取一次 —— allowedModels 随心跳变，
  // 取快照会让 worker 用上一轮的模型范围。
  pool = createWorkerPool({
    log,
    config: () => ({
      gatewayURL: client.baseURL,
      agentName: AGENT_NAME,
      agentKey: client.agentKey,
      agentSecret: AGENT_SECRET,
      allowedModels,
      replyProvider: REPLY_PROVIDER,
      replyModel: REPLY_MODEL,
      turnTimeoutMs: TURN_TIMEOUT_MS,
    }),
    // worker 里 connect_to_server 换了坐标：worker 马上就退了，改在它自己身上
    // 等于没改。主进程据此重建 SSE，后续 worker 的 job 也会带上新坐标。
    onReconfigure: (url, key) => {
      if (!client.reconfigure({ url, agentKey: key })) return;
      log(`Gateway 坐标已更新为 ${client.baseURL}，重建 SSE`);
      client.stopSSE();
      client.startSSE(handleSSEEvent, log);
    },
    maxWorkers: MAX_WORKERS,
    workerMaxMs: WORKER_MAX_MS,
  });

  // 主进程不跑模型，因此**不建**邮件工具：工具是给模型调的，而这里没有会话。
  // （工具 schema 的约束由 test/tool-schema.test.mjs 直接验证 createMailTools，
  //   不需要在这里建一份没人用的副本。）
  //
  // connect_to_server 换坐标的闭环在 pool 的 onReconfigure 里 —— 那个工具
  // 跑在 worker 里，worker 把新坐标回报给主进程，主进程据此重建 SSE。

  try {
    await client.register();                          // B-1.2
    saveConfig({ gateway_url: GATEWAY_URL, agent_name: AGENT_NAME, registered_at: new Date().toISOString() });
    // ★ 2026-10-05：把**密钥**也印出一段。原先只印 AGENT_NAME（本地配置里的），
    // 而身份实际由密钥决定 —— 于是「配置的 name 与密钥归属不一致」在日志里
    // 完全看不出来（读起来一切正常，只有收信方在库里看到另一个署名才会发现）。
    // 印 key 前 8 位后，那类漂移一眼能对出来；不发全串，避开日志泄密。
    log(`已接入 ${GATEWAY_URL}，身份 ${AGENT_NAME}（${agentKey ? `密钥认证 ${String(agentKey).slice(0, 8)}…` : 'name/secret 认证'}）。`);
  } catch (e) {
    // 密钥未登记时这里报「密钥无效」—— 必须说清该做什么，
    // 否则用户只看到一句 401，不知道要拿密钥去后台登记。
    const why = describeError(e);
    log(`注册失败: ${why}`);

    // ★ 2026-10-05：身份绑定冲突必须**停下**，不能只记一行日志就继续跑。
    //
    // 服务端已经内置这道闸（实测）：
    //     {"error":"该密钥已绑定到 Agent \\"dsh\\"，不能用于注册 \\"pi\\""}
    //
    // 也就是说「这把 key 到底是谁的」服务端能权威回答。旧代码把它当普通
    // 错误打印一句就往下走，于是**继续以错误身份发信/收信/上报心跳** ——
    // 而且日志里那句「身份 ${AGENT_NAME}」印的是**本地配置里的名字**，
    // 读起来一切正常；只有收信方在库里看到另一个署名才会发现。
    //
    // 与「配置在模块加载期被缓存」同族（换 key 后不自知），但那道无法在本文件
    // 修（何时读 env 由 pi 决定），这一道可以：服务端已经给了确定答复的时候
    // 把进程停住。
    //
    // ★ 只对「绑定冲突」停。密钥无效 / 网络不通属于可恢复的，
    //   停掉会连带 SSE 与心跳（那两样本就该重试后自愈）。
    if (/已绑定到 Agent/.test(why)) {
      log('');
      log('★★ 身份冲突，已停止：上面这把密钥在服务端属于另一个 Agent。');
      log('   本进程配置的 agent_name 与密钥实际归属不一致，继续运行会');
      log('   以错误身份发信 —— 收信方在库里看到的署名会与预期不符。');
      log('');
      log('   查这把 key 实际属于谁：');
      log('     sqlite3 /opt/agentmail/data/agentmail.db \\');
      log("       \"SELECT agent_name FROM agent_keys WHERE key_token LIKE '<key前12位>%';\"");
      log('   然后把 pi.env 里的 AGENTMAIL_AGENT_NAME 改成那个名字（或换一把匹配的密钥），');
      log('   并**重启进程** —— 配置在模块加载期读入，改 env 不热生效。');
      log('');
      throw new Error(`身份冲突：${why}`);
    }

    if (agentKey) log(`若提示密钥无效，请让管理员在 AgentMail 后台登记这把密钥。`);
  }

  let caughtUp = false;
  const beat = async () => {
    const [platform_sessions, models] = await Promise.all([reportSessions(), reportModels()]);
    const body = {};
    body.mode_enforcement = 'native'; // tool_call hook 能 block bash/write/edit
    if (platform_sessions) body.platform_sessions = platform_sessions;
    if (models) body.models = models;
    try {
      const res = await client.post('/agent/heartbeat', body);
      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, res?.pending_workspaces);
      }
    } catch {
      // B-2.1：心跳失败不重试不报错。真连不上时 Gateway 会把它判成离线，
      // 那才是可见的信号；桥自己打一串错误日志只会淹掉真正的问题。
    }
  };
  await beat();                                       // B-1.3：不等第一个 30 秒周期
  heartbeatTimer = setInterval(beat, 30_000);         // B-1.5

  client.startSSE(handleSSEEvent, log);

  for (const sig of ['SIGINT', 'SIGTERM']) process.on(sig, () => shutdown(sig));
}

/**
 * SSE 事件分派。
 *
 * 这个函数**必须保持廉价**：它跑在读循环上。派活给 pool 是同步的（fork 是
 * 异步的，pool.submit 只是入队），所以读循环不会因为一封邮件停下。
 *
 * 提成命名函数是因为 connect_to_server 换地址后要用同一个处理器重建长连 ——
 * 内联箭头函数在那里拿不到，只能复制一遍，而复制出来的两份迟早会分叉。
 */
function handleSSEEvent(type, data) {
  if (type === 'permission_decision') {
    handlePermissionDecision(data);
    return;
  }
  if (type === 'session_archived') {
    // 会话归档 = 那条会话再也不会收信（别名 404），pool 里的 sessionState
    // 可以确定性地清掉，不必等上限淘汰去猜。
    if (pool?.forget(data?.session_id || '')) {
      log(`会话 ${data.session_id} 已归档，清除本地状态`);
    }
    return;
  }
  if (type !== 'new_mail') return;
  if (data?.role && data.role !== 'to' && data.role !== 'cc') return;
  const id = data?.mail_id;
  if (!id || deliveredMails.has(id)) return;        // B-3 第 1 步：去重
  markDelivered(id);
  // ★ 决策回执**不是新任务**（2026-09-13 线上缺陷）。
  //
  // 它长得像普通邮件（"Re: 权限请求 - 拒绝"），内容却已经随 SSE 的
  // permission_decision 直接交给了发起询问的 worker。以前它会**再**起一轮：
  // 同一件事处理两次，还把人类真正的新邮件挤在这条会话的队列后面 ——
  // 实测人类的更正 22:18:08 发出，直到 22:26:30 才被模型看到。
  //
  // 走到这里说明决策事件还没到（事件先到就已经记进 deliveredMails 了）——
  // 那就按 B-4.3 当**通知**投进原会话，而不是凭空当新任务。
  if (data?.mail_type === 'permission_decision') {
    if (!data?.session_id || !pool.hasSession(data.session_id)) {
      log(`决策回执 ${id} 无对应会话，忽略（不凭空新开）`);
      return;
    }
    log(`决策回执 ${id} 按通知投递（决策事件尚未到达）`);
    pool.submit('permission', data);
    return;
  }

  pool.submit('mail', data);
}

function shutdown(reason) {
  if (shuttingDown) return;
  shuttingDown = true;
  log(`收到 ${reason}，关停中…`);

  if (heartbeatTimer) clearInterval(heartbeatTimer);  // B-9.1
  client?.stopSSE();

  // pool.stop 先给每个 worker 发 shutdown（让它把未决权限询问 fail closed，
  // B-9.2 / N-9），再给 2 秒自己退，然后 SIGKILL。
  //
  // 不直接杀：pi 侧那些 await 不会返回，而 worker 里可能正握着会话文件。
  pool?.stop();

  releaseLock();
  // 留出 pool.stop 的宽限窗口再退：主进程先死会让子进程变成孤儿
  // （systemd 的 KillMode 会兜住，但那时 fail closed 已经来不及做了）。
  setTimeout(() => process.exit(0), 2500).unref?.();
  // B-9.3：不发「插件下线」通知邮件
}

main().catch((e) => {
  log(`启动失败: ${describeError(e)}`);
  releaseLock();
  // 故障邮件由 systemd ExecStopPost 统一发送：它还能覆盖 SIGKILL/OOM，
  // 而进程内的 uncaughtException 钩子覆盖不了这些真正的意外终止。
  process.exit(1);
});