zsjz-ai/ai-electron/electron/service/codex/index.ts

import { join } from 'node:path';
import { mkdir } from 'node:fs/promises';
import { getMainWindow } from 'ee-core/electron';
import { logger } from 'ee-core/log';
import { ApprovalBroker } from './approvalBroker';
import { chatBridge } from './chatBridgeService';
import { ensureCodexHome, getCodexDataDir, getCodexHome, getEventsDir } from './codexHome';
import { CodexRuntime, type RuntimeStatus } from './codexRuntime';
import { EventLog } from './eventLog';
import { notificationToEvent } from './eventMapper';
import { McpService } from './mcpService';
import { readAppliedProvider } from './providerService';
import { SkillService } from './skillService';
import {
  asMessage,
  CODEX_APPROVAL_CHANNEL,
  CODEX_APPROVAL_CLOSED_CHANNEL,
  CODEX_DIAGNOSTIC_CHANNEL,
  CODEX_EVENT_CHANNEL,
  CODEX_STATUS_CHANNEL,
  type AppliedProviderInfo,
  type CodexStatusResult,
} from './types';
import type { JsonRpcServerRequest } from './jsonRpcPeer';

/**
 * Codex 能力的装配层：懒启动单例、事件转发到渲染进程、诊断环形缓冲。
 * 刻意不在启动时自动带上 provider —— apiKey 只能由渲染进程从后端取到后传入，
 * 所以重启后需要页面重新「应用模型」，未应用时 MCP/Skill 管理照样可用。
 */

const MAX_DIAGNOSTICS = 500;

export interface CodexContainer {
  runtime: CodexRuntime;
  broker: ApprovalBroker;
  eventLog: EventLog;
  mcp: McpService;
  skills: SkillService;
}

let container: CodexContainer | null = null;
let creating: Promise<CodexContainer> | null = null;
let appliedCache: AppliedProviderInfo | null = null;
let appliedLoaded = false;
const diagnostics: Array<{ ts: string; line: string }> = [];

function send(channel: string, payload: unknown): void {
  try {
    const win = getMainWindow();
    if (win && !win.isDestroyed()) win.webContents.send(channel, payload);
  } catch {
    // 窗口尚未就绪或已销毁：丢事件，不影响主流程
  }
}

function pushDiagnostic(line: string): void {
  const entry = { ts: new Date().toISOString(), line };
  diagnostics.push(entry);
  if (diagnostics.length > MAX_DIAGNOSTICS) {
    diagnostics.splice(0, diagnostics.length - MAX_DIAGNOSTICS);
  }
  // 环形缓冲只在内存里，进程一崩就没了：同步落 ee.log，否则启动失败无从复盘
  logger.warn(`[codex] ${line}`);
  send(CODEX_DIAGNOSTIC_CHANNEL, entry);
}

export async function getCodex(): Promise<CodexContainer> {
  if (container) return container;
  if (creating) return creating;
  creating = (async () => {
    await ensureCodexHome();
    // 桥接的诊断出口统一进环形缓冲（providerService 启动桥接时不再单独传）
    chatBridge.setDiagnostic(pushDiagnostic);
    const runtime = new CodexRuntime({ codexHome: getCodexHome() });
    const broker = new ApprovalBroker(runtime);
    const eventLog = new EventLog(getEventsDir());
    await eventLog.init();

    runtime.on('notification', (notification: { method: string; params?: unknown }) => {
      if (notification.method === 'serverRequest/resolved') {
        const requestId = (notification.params as { requestId?: string | number } | undefined)?.requestId;
        if (requestId !== undefined) broker.resolveFromServer(requestId);
      }
      const event = notificationToEvent(notification);
      if (!event) return;
      send(CODEX_EVENT_CHANNEL, event);
      eventLog.append(event).catch((error: unknown) => {
        logger.error('[codex] 事件落盘失败:', asMessage(error));
      });
    });
    runtime.on('status', (status: RuntimeStatus) => send(CODEX_STATUS_CHANNEL, toStatusResult(status)));
    runtime.on('diagnostic', (line: string) => pushDiagnostic(line));
    runtime.on('serverRequest', (request: JsonRpcServerRequest) => broker.handle(request));

    broker.on('approval', (approval: unknown) => send(CODEX_APPROVAL_CHANNEL, approval));
    broker.on('closed', (token: string) => send(CODEX_APPROVAL_CLOSED_CHANNEL, { token }));
    broker.on('expired', (approval: unknown) => {
      pushDiagnostic(`审批超时已自动拒绝：${JSON.stringify(approval)}`);
    });
    broker.on('diagnostic', (line: string) => pushDiagnostic(line));

    container = {
      runtime,
      broker,
      eventLog,
      mcp: new McpService(runtime),
      skills: new SkillService(runtime, { onDiagnostic: pushDiagnostic }),
    };
    return container;
  })();
  try {
    return await creating;
  } finally {
    creating = null;
  }
}

export async function getAppliedProvider(): Promise<AppliedProviderInfo | null> {
  if (!appliedLoaded) {
    appliedCache = await readAppliedProvider();
    appliedLoaded = true;
  }
  return appliedCache;
}

export function setAppliedProvider(value: AppliedProviderInfo | null): void {
  appliedCache = value;
  appliedLoaded = true;
}

export function toStatusResult(status: RuntimeStatus): CodexStatusResult {
  const bridgeInfo = chatBridge.info;
  return {
    state: status.state,
    running: status.state === 'ready',
    version: status.version,
    binaryPath: status.binaryPath,
    codexHome: getCodexHome(),
    defaultModel: status.defaultModel,
    providerId: status.providerId,
    degraded: status.degraded,
    error: status.error,
    applied: appliedCache,
    bridge: bridgeInfo ? { running: true, upstream: bridgeInfo.upstreamBaseUrl } : null,
  };
}

export function tailDiagnostics(limit = 200): Array<{ ts: string; line: string }> {
  const size = Math.max(1, Math.min(MAX_DIAGNOSTICS, Math.trunc(limit)));
  return diagnostics.slice(-size);
}

/** 会话默认工作目录：用户没指定 cwd 时用，避免 Codex 直接在安装目录里动手 */
export async function defaultWorkspaceDir(): Promise<string> {
  const dir = join(getCodexDataDir(), 'workspace');
  await mkdir(dir, { recursive: true });
  return dir;
}

export async function disposeCodex(): Promise<void> {
  // 桥接不挂在 container 上，有没有 container 都要兜底停掉
  await chatBridge.stop().catch(() => undefined);
  const current = container;
  container = null;
  if (!current) return;
  current.broker.closeAll();
  await current.eventLog.flush().catch(() => undefined);
  await Promise.race([
    current.runtime.dispose(),
    new Promise<void>((resolve) => setTimeout(resolve, 3_000)),
  ]);
}