cc-web/lib/codex-app-server-client.js

'use strict';

const readline = require('readline');
const { spawn } = require('child_process');

function createCodexAppServerClient(options = {}) {
  const command = options.command || 'codex';
  const args = Array.isArray(options.args) && options.args.length > 0
    ? options.args.slice()
    : ['app-server', '--stdio'];
  const env = options.env || process.env;
  const cwd = options.cwd || process.cwd();
  const clientInfo = options.clientInfo || {
    name: 'ccweb_codexapp',
    title: 'CC-Web Codex App',
    version: '0.1.0',
  };
  const onNotification = typeof options.onNotification === 'function' ? options.onNotification : () => {};
  const onServerRequest = typeof options.onServerRequest === 'function' ? options.onServerRequest : null;
  const onExit = typeof options.onExit === 'function' ? options.onExit : () => {};
  const onLog = typeof options.onLog === 'function' ? options.onLog : () => {};
  const postInitialize = typeof options.postInitialize === 'function' ? options.postInitialize : null;

  let proc = null;
  let rl = null;
  let nextId = 1;
  let initPromise = null;
  let exited = false;
  const pending = new Map();

  function rejectAllPending(err) {
    for (const [, pendingRequest] of pending) {
      clearTimeout(pendingRequest.timer);
      pendingRequest.reject(err);
    }
    pending.clear();
  }

  function sendRaw(message) {
    if (!proc || !proc.stdin || proc.stdin.destroyed) {
      throw new Error('Codex app-server 未启动。');
    }
    proc.stdin.write(`${JSON.stringify(message)}\n`);
  }

  function respondToServerRequest(id, result, error) {
    try {
      if (error) {
        sendRaw({ id, error });
      } else {
        sendRaw({ id, result: result || {} });
      }
    } catch (err) {
      onLog('WARN', 'codex_app_server_response_failed', { error: err.message });
    }
  }

  function handleServerRequest(message) {
    const id = message.id;
    const method = message.method;
    const params = message.params || {};
    if (onServerRequest) {
      Promise.resolve()
        .then(() => onServerRequest({ method, params, id }))
        .then((result) => respondToServerRequest(id, result || {}))
        .catch((err) => respondToServerRequest(id, null, {
          code: -32603,
          message: err?.message || 'cc-web 无法处理 Codex app-server 请求。',
        }));
      return;
    }
    respondToServerRequest(id, null, {
      code: -32601,
      message: `cc-web 暂不支持 Codex app-server 请求: ${method}`,
    });
  }

  function handleMessage(line) {
    let message;
    try {
      message = JSON.parse(line);
    } catch {
      onLog('WARN', 'codex_app_server_invalid_json', { line: String(line || '').slice(0, 200) });
      return;
    }

    if (Object.prototype.hasOwnProperty.call(message, 'id')) {
      const pendingRequest = pending.get(message.id);
      if (pendingRequest) {
        pending.delete(message.id);
        clearTimeout(pendingRequest.timer);
        if (message.error) {
          const err = new Error(message.error.message || 'Codex app-server 请求失败。');
          err.code = message.error.code;
          err.data = message.error.data;
          pendingRequest.reject(err);
        } else {
          pendingRequest.resolve(message.result || {});
        }
        return;
      }
      if (message.method) {
        handleServerRequest(message);
        return;
      }
    }

    if (message.method) {
      onNotification(message);
    }
  }

  function request(method, params = {}, timeoutMs = 300000) {
    const id = nextId++;
    const message = { id, method, params };
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        pending.delete(id);
        reject(new Error(`Codex app-server 请求超时: ${method}`));
      }, timeoutMs);
      pending.set(id, { resolve, reject, timer, method });
      try {
        sendRaw(message);
      } catch (err) {
        clearTimeout(timer);
        pending.delete(id);
        reject(err);
      }
    });
  }

  function notification(method, params = {}) {
    sendRaw({ method, params });
  }

  function start() {
    if (initPromise) return initPromise;
    exited = false;
    proc = spawn(command, args, {
      env,
      cwd,
      stdio: ['pipe', 'pipe', 'pipe'],
      windowsHide: true,
    });

    let stderr = '';
    proc.stderr.on('data', (chunk) => {
      stderr += chunk.toString();
      if (stderr.length > 4000) stderr = stderr.slice(-4000);
    });

    rl = readline.createInterface({ input: proc.stdout });
    rl.on('line', handleMessage);

    proc.on('exit', (code, signal) => {
      exited = true;
      if (rl) rl.close();
      const err = new Error(`Codex app-server 已退出: code=${code ?? 'null'} signal=${signal || 'null'}`);
      err.exitCode = code;
      err.signal = signal;
      err.stderr = stderr;
      rejectAllPending(err);
      onExit({ code, signal, stderr });
    });

    proc.on('error', (err) => {
      rejectAllPending(err);
      onExit({ code: null, signal: null, stderr: err.message });
    });

    initPromise = request('initialize', {
      clientInfo,
      capabilities: { experimentalApi: true },
    }, 30000)
      .then(async (result) => {
        notification('initialized', {});
        if (postInitialize) await postInitialize({ request, notification, onLog });
        return result;
      })
      .catch((err) => {
        stop();
        throw err;
      });

    return initPromise;
  }

  function stop() {
    initPromise = null;
    if (rl) {
      try { rl.close(); } catch {}
      rl = null;
    }
    if (proc && !exited) {
      try { proc.kill('SIGTERM'); } catch {}
      setTimeout(() => {
        try {
          if (proc && !proc.killed) proc.kill('SIGKILL');
        } catch {}
      }, 3000);
    }
    proc = null;
    rejectAllPending(new Error('Codex app-server 已停止。'));
  }

  function isRunning() {
    return !!proc && !exited;
  }

  return {
    start,
    stop,
    request,
    notification,
    isRunning,
    pid: () => proc?.pid || null,
  };
}

module.exports = { createCodexAppServerClient };