MailUI4Agents/plugins/zcode-mail-bridge/src/zcode-run.mjs

/**
 * 跑一轮 ZCode：起一个 headless 进程，把事件流读回来，取出最终回复。
 *
 * # 为什么用 `--output-format stream-json` 而不是 `--json`
 *
 * 两者都能在最后给出 `{sessionId, response}`（这是从 CLI 产物里读出的契约：
 * 逐条事件写 `mapSessionEvent(event)`，最后补一行 `{type:"result", …}`）。
 * 差别在**过程可见**：`--json` 全程没有输出，一个卡住的回合与一个正在干活的
 * 回合在外部看起来完全一样 —— 而邮件驱动的会话没有界面，除了日志没人能看见它。
 * stream-json 让「正在做什么」进得了日志，也让超时能被归因。
 *
 * # 会话延续
 *
 * 首轮没有 `--resume`，从 `result` 行里取回 `sess_...` 存下来；后续同一
 * AgentMail 会话的信都带上 `--resume`，于是模型记得前几轮。丢了它就等于
 * 「每封信都从零开始」，而模型会表现得像没见过之前的要求。
 *
 * # 超时
 *
 * 到期杀**进程树**：ZCode 会派生工具子进程（bash 等），只杀父进程会留下一堆
 * 孤儿继续跑，而且它们还占着工作目录。
 */

import { spawn as nodeSpawn } from 'node:child_process';

/** 默认的 CLI 入口：ZCode 自带的纯 CLI（不是 Electron 那个 GUI 入口）。 */
export const DEFAULT_CLI = '/opt/ZCode/resources/glm/zcode.cjs';

const DEFAULT_TURN_TIMEOUT_MS = 20 * 60 * 1000;

/**
 * 拼出一次 headless 调用的参数表。
 *
 * 单独抽出来是为了可测：`--mode` 漏掉会**静默绕过全部授权询问**
 * （`--prompt` 的默认 mode 是 yolo），这种缺陷不会报错，只会让授权系统消失。
 */
export function buildRunArgs({
  prompt,
  cwd,
  mode,
  maxTurns,
  resumeSessionId,
  allowedTools,
  disallowedTools,
  cliPath = DEFAULT_CLI
}) {
  const args = [
    cliPath,
    '--prompt',
    prompt,
    '--output-format',
    'stream-json',
    '--cwd',
    cwd
  ];
  // mode 必传：不传就是 yolo，等于关掉授权。
  if (!mode) throw new Error('buildRunArgs 需要 mode（不传等于 yolo，会绕过授权询问）');
  args.push('--mode', mode);
  if (maxTurns) args.push('--max-turns', String(maxTurns));
  if (resumeSessionId) args.push('--resume', resumeSessionId);
  // `--allowed-tools` 在 help 里写着，但 CLI 的解析器**不认**它
  // （`Unknown option '--allowed-tools'`），会直接退到 usage。
  // 留着这个参数会拼出一条永远跑不起来的命令行，所以在拼参数这一步就报错，
  // 而不是等到一两分钟后拿到一段 usage 文本才去查。
  if (Array.isArray(allowedTools) && allowedTools.length) {
    throw new Error(
      '本版本 ZCode CLI 不支持 --allowed-tools（解析器报 Unknown option）；' +
        '只能用 --disallowed-tools 黑名单，见 src/turn-mode.mjs 的 REVIEWED_DENYLIST'
    );
  }
  // 逗号分隔：help 说「Comma or space-separated」，实测两种都行，
  // 但逗号不受调用方是否把参数拼成单个 argv 的影响。
  if (Array.isArray(disallowedTools) && disallowedTools.length) {
    args.push('--disallowed-tools', disallowedTools.join(','));
  }
  return args;
}

/**
 * 解析一行 stream-json 输出的**纯函数**部分。
 *
 * 返回 `null` 表示这行不是我们要的（非 JSON、或没有意义的行）——
 * 调用方据此计数，好让「输出格式变了」这件事能被发现，而不是静默当成没输出。
 *
 * @returns {{kind:'event'|'result', event?:any, sessionId?:string, response?:string}|null}
 */
export function parseStreamLine(line) {
  const text = String(line ?? '').trim();
  if (!text || text[0] !== '{') return null;
  let obj;
  try {
    obj = JSON.parse(text);
  } catch {
    return null;
  }
  if (!obj || typeof obj !== 'object') return null;
  if (obj.type === 'result') {
    return {
      kind: 'result',
      sessionId: typeof obj.sessionId === 'string' ? obj.sessionId : '',
      response: typeof obj.response === 'string' ? obj.response : '',
      eventCount: typeof obj.eventCount === 'number' ? obj.eventCount : undefined,
      projection: obj.projection
    };
  }
  return { kind: 'event', event: obj };
}

/**
 * 跑一轮。
 *
 * @param {{prompt:string, cwd:string, mode:string, maxTurns?:number,
 *          resumeSessionId?:string, turnTimeoutMs?:number,
 *          disallowedTools?:string[],
 *          env?:Record<string,string>, cliPath?:string, nodePath?:string,
 *          onChild?:(kill:(signal?:string)=>void)=>void,
 *          onEvent?:(event:any)=>void}} opts
 * @param {{spawn?:Function, log?:Function}} [deps] spawn 可注入以便测试
 * @returns {Promise<{sessionId:string, response:string, events:any[], exitCode:number,
 *                    unparsable:number, timedOut:boolean, killed:boolean,
 *                    stderrTail:string, argv:string[]}>}
 */
export function runTurn(opts, deps = {}) {
  const spawn = deps.spawn || nodeSpawn;
  const log = deps.log || (() => {});
  const nodePath = opts.nodePath || process.execPath;
  const args = buildRunArgs(opts);
  const timeoutMs = opts.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS;
  // 宽限期：SIGTERM 之后等多久 SIGKILL；再等多久就无论如何收尾。
  // 可配是为了测试 —— 生产用默认值。
  const killGraceMs = opts.killGraceMs ?? 5000;
  const settleGraceMs = opts.settleGraceMs ?? 2000;
  const startedAt = Date.now();

  return new Promise(resolve => {
    let child;
    try {
      child = spawn(nodePath, args, {
        cwd: opts.cwd,
        env: { ...process.env, ...(opts.env || {}) },
        stdio: ['ignore', 'pipe', 'pipe'],
        // 自己建进程组，超时时能整组杀 —— 否则 ZCode 派生的工具子进程会活下来。
        detached: process.platform !== 'win32'
      });
    } catch (e) {
      resolve({
        sessionId: '',
        response: '',
        events: [],
        exitCode: -1,
        unparsable: 0,
        timedOut: false,
        killed: false,
        stderrTail: `spawn 失败：${e?.message || e}`,
        argv: args
      });
      return;
    }

    const events = [];
    let result = null;
    let unparsable = 0;
    let stdoutBuf = '';
    const stderrLines = [];
    let timedOut = false;
    let killed = false;
    let settled = false;

    const killTree = signal => {
      killed = true;
      try {
        if (process.platform !== 'win32' && child.pid) {
          // 负号 = 整个进程组
          process.kill(-child.pid, signal);
        } else {
          child.kill(signal);
        }
      } catch {
        try {
          child.kill(signal);
        } catch {
          /* 已经没了 */
        }
      }
    };

    const timer = setTimeout(() => {
      timedOut = true;
      log(`回合超时（${Math.round(timeoutMs / 1000)} 秒），终止进程树`);
      killTree('SIGTERM');
      // 给它一点时间优雅退出；不走就强杀（工具子进程不会自己收手）。
      //
      // 这几个定时器**刻意不 unref**：在途的回合应该让进程活着。
      // unref 过的定时器会让「没有其它活跃句柄」的进程直接退出，
      // 于是收尾这段代码根本没机会跑（测试里就是这样暴露的）。
      setTimeout(() => killTree('SIGKILL'), killGraceMs);
      // ★ 最后兵底：**必须**让这一轮结束。
      //
      // 没有这一步时，一个既不退也不报错的孩子会让 Promise 永不 settle ——
      // 驱动会对那封信永远挂住，而且队列是串行的，后面的信全都不再被处理。
      // 杀掉进程本来就应该算「这一轮结束了」，而不是「再等等看」。
      setTimeout(() => {
        if (settled) return;
        log('强杀后仍未退出，按超时收尾（不再等）');
        finish(-1);
      }, killGraceMs + settleGraceMs);
    }, timeoutMs);

    const finish = exitCode => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      resolve({
        sessionId: result?.sessionId || '',
        response: result?.response || '',
        events,
        exitCode,
        unparsable,
        timedOut,
        killed,
        stderrTail: stderrLines.slice(-8).join('\n'),
        argv: args
      });
    };

    // 把「怎么杀」交给调用方：驱动在收到 SIGTERM 时要能终止在途的回合。
    // 不交出去的话，systemd 杀掉驱动之后那个 ZCode 进程还在跑工具，
    // 而没有任何人（也没有任何界面）看着它。
    if (typeof opts.onChild === 'function') {
      try {
        opts.onChild(killTree);
      } catch {
        /* 调用方自己的问题，不影响这一轮 */
      }
    }

    child.stdout?.on('data', chunk => {
      stdoutBuf += chunk.toString('utf8');
      let nl;
      while ((nl = stdoutBuf.indexOf('\n')) >= 0) {
        const line = stdoutBuf.slice(0, nl);
        stdoutBuf = stdoutBuf.slice(nl + 1);
        const parsed = parseStreamLine(line);
        if (!parsed) {
          // 空行是正常的（CLI 用空格分隔事件），只统计真的不像 JSON 的行。
          if (line.trim()) unparsable++;
          continue;
        }
        if (parsed.kind === 'result') {
          result = parsed;
        } else {
          events.push(parsed.event);
          // 逐个事件交给调用方：邮件驱动的会话没有界面，
          // 「模型正在干什么」只能靠日志，否则一个五分钟的回合在外部看起来
          // 与一个卡死的回合完全一样。
          if (typeof opts.onEvent === 'function') {
            try {
              opts.onEvent(parsed.event);
            } catch {
              /* 调用方的日志出错不该影响这一轮 */
            }
          }
        }
      }
    });

    child.stderr?.on('data', chunk => {
      for (const line of chunk.toString('utf8').split('\n')) {
        const t = line.trim();
        if (!t) continue;
        stderrLines.push(t);
        // 实时转出去 —— 邮件驱动没有界面，日志是唯一能看见它在干什么的地方。
        log(t);
      }
    });

    child.on('error', e => {
      stderrLines.push(`进程错误：${e?.message || e}`);
      finish(-1);
    });

    child.on('close', code => {
      // 收尾：最后一行可能没有换行
      if (stdoutBuf.trim()) {
        const parsed = parseStreamLine(stdoutBuf);
        if (parsed?.kind === 'result') result = parsed;
        else if (parsed?.kind === 'event') events.push(parsed.event);
        else unparsable++;
      }
      log(
        `回合结束：退出码 ${code}，事件 ${events.length} 条，` +
          `会话 ${result?.sessionId || '(无)'}，耗时 ${Math.round((Date.now() - startedAt) / 1000)} 秒`
      );
      finish(code ?? -1);
    });
  });
}