cc-web/lib/codex-app-worker.js

'use strict';

const { createCodexAppServerClient } = require('./codex-app-server-client');

let client = null;
let currentSpec = null;
let nextParentRequestId = 1;
const pendingParentRequests = new Map();

function send(message) {
  if (typeof process.send === 'function') {
    process.send(message);
  }
}

function serializeError(err) {
  return {
    code: err?.code || -32603,
    message: err?.message || String(err || 'Codex App worker error'),
    data: err?.data || null,
  };
}

function reply(id, result, error) {
  if (!id) return;
  if (error) {
    send({ id, error: serializeError(error) });
  } else {
    send({ id, result: result || {} });
  }
}

function requestParent(request, timeoutMs = 300000) {
  const requestId = String(nextParentRequestId++);
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pendingParentRequests.delete(requestId);
      reject(new Error(`Codex App worker 等待主进程处理请求超时: ${request.method || ''}`));
    }, timeoutMs);
    pendingParentRequests.set(requestId, { resolve, reject, timer });
    send({ type: 'serverRequest', requestId, request });
  });
}

async function postInitialize({ request, onLog } = {}) {
  if (typeof request !== 'function') return;
  try {
    await request('experimentalFeature/enablement/set', { enablement: { goals: true } }, 30000);
    if (typeof onLog === 'function') onLog('INFO', 'codex_app_worker_goals_feature_enabled', {});
  } catch (err) {
    if (typeof onLog === 'function') {
      onLog('INFO', 'codex_app_worker_goals_feature_enable_failed', { error: err?.message || String(err || '') });
    }
  }

  try {
    const result = await request('collaborationMode/list', {}, 30000);
    if (typeof onLog === 'function') onLog('INFO', 'codex_app_worker_collaboration_modes', { result });
  } catch (err) {
    if (typeof onLog === 'function') {
      onLog('INFO', 'codex_app_worker_collaboration_mode_list_failed', { error: err?.message || String(err || '') });
    }
  }
}

function configure(spec = {}) {
  const nextSpec = {
    command: spec.command || 'codex',
    args: Array.isArray(spec.args) && spec.args.length > 0 ? spec.args : ['app-server', '--stdio'],
    env: spec.env || process.env,
    cwd: spec.cwd || process.cwd(),
    clientInfo: spec.clientInfo || undefined,
  };

  const nextSignature = JSON.stringify({
    command: nextSpec.command,
    args: nextSpec.args,
    cwd: nextSpec.cwd,
    envCodeHome: nextSpec.env.CODEX_HOME || '',
  });
  const currentSignature = currentSpec ? JSON.stringify({
    command: currentSpec.command,
    args: currentSpec.args,
    cwd: currentSpec.cwd,
    envCodeHome: currentSpec.env.CODEX_HOME || '',
  }) : '';

  if (client && nextSignature === currentSignature) {
    currentSpec = nextSpec;
    return;
  }

  if (client) {
    try { client.stop(); } catch {}
    client = null;
  }
  currentSpec = nextSpec;
  client = createCodexAppServerClient({
    command: nextSpec.command,
    args: nextSpec.args,
    env: nextSpec.env,
    cwd: nextSpec.cwd,
    clientInfo: nextSpec.clientInfo,
    onNotification: (notification) => send({ type: 'notification', notification }),
    onServerRequest: (request) => requestParent(request),
    onExit: (info) => send({ type: 'exit', info }),
    onLog: (level, event, data) => send({ type: 'log', level, event, data }),
    postInitialize,
  });
}

process.on('message', (message = {}) => {
  if (message.type === 'serverRequestResult') {
    const item = pendingParentRequests.get(String(message.requestId || ''));
    if (!item) return;
    pendingParentRequests.delete(String(message.requestId || ''));
    clearTimeout(item.timer);
    if (message.error) {
      const err = new Error(message.error.message || '主进程处理 Codex App 请求失败。');
      err.code = message.error.code;
      err.data = message.error.data;
      item.reject(err);
    } else {
      item.resolve(message.result || {});
    }
    return;
  }

  Promise.resolve()
    .then(async () => {
      switch (message.type) {
        case 'configure':
          configure(message.spec || {});
          return {};
        case 'start':
          if (!client) configure(currentSpec || {});
          return client.start();
        case 'request':
          if (!client) configure(currentSpec || {});
          return client.request(message.method, message.params || {}, message.timeoutMs || 300000);
        case 'notification':
          if (!client) configure(currentSpec || {});
          client.notification(message.method, message.params || {});
          return {};
        case 'reloadMcpServers':
          if (!client) configure(currentSpec || {});
          return client.reloadMcpServers();
        case 'stop':
          if (client) client.stop();
          process.exit(0);
          return {};
        default:
          throw new Error(`未知 Codex App worker 消息: ${message.type}`);
      }
    })
    .then((result) => reply(message.id, result))
    .catch((err) => reply(message.id, null, err));
});

process.on('disconnect', () => {
  if (client) {
    try { client.stop(); } catch {}
  }
  process.exit(0);
});