feat: add Codex executor with session resume

2026-07-22 22:15:15 +08:00
parent d38151ad78
commit 214b7c9a30
19 changed files with 1855 additions and 33 deletions
@@ -16,6 +16,7 @@ import { spawnStreamingChild } from "./child-process.js";
 import type {
  AgySessionHandle,
  ClaudeSessionHandle,
+  CodexSessionHandle,
  ExecutorAdapter,
  ExecutorConfig,
  ExecutorKind,
@@ -72,7 +73,7 @@ function getSafetyConstraints(vcsKind?: VcsKind): string {

 function startProgressSidecar(
  dir: string,
-  executor: "reasonix" | "claude" | "agy",
+  executor: ExecutorKind,
  cycleIndex: number,
  attemptIndex: number,
  logFilePath: string | undefined,
@@ -1172,6 +1173,360 @@ class AgyAdapter implements ExecutorAdapter {
 }

 // ---------------------------------------------------------------------------
+// ---------------------------------------------------------------------------
+// Codex adapter
+// ---------------------------------------------------------------------------
+
+function extractCodexSessionId(output: string): string | undefined {
+  try {
+    for (const line of output.split("\n")) {
+      const trimmed = line.trim();
+      if (!trimmed.startsWith("{")) continue;
+      const parsed = JSON.parse(trimmed) as Record<string, unknown>;
+      // Real Codex thread.started carries thread_id at event top level
+      if (parsed.type === "thread.started") {
+        if (typeof parsed.thread_id === "string") return parsed.thread_id;
+        const thread = parsed.thread as Record<string, unknown> | undefined;
+        if (thread && typeof thread.id === "string") return thread.id;
+      }
+      // Legacy fallback
+      if (typeof parsed.session_id === "string") return parsed.session_id;
+      if (parsed.type === "session_meta" && typeof parsed.id === "string") return parsed.id;
+    }
+  } catch {
+    // fall through
+  }
+  const m = output.match(/"session_id"\s*:\s*"([^"]+)"/);
+  return m?.[1];
+}
+
+export function validateCodexSessionForWorkflow(
+  sessionId: string,
+  repoRoot: string,
+  planTitle: string,
+): string {
+  const { sessionFile, content } = readCodexSessionForRepository(sessionId, repoRoot);
+  if (!content.includes(planTitle)) {
+    throw new Error(`Codex session ${sessionId} does not contain frozen plan title: ${planTitle}`);
+  }
+  return sessionFile;
+}
+
+export function validateCodexSessionForRepository(sessionId: string, repoRoot: string): string {
+  return readCodexSessionForRepository(sessionId, repoRoot).sessionFile;
+}
+
+function readCodexSessionForRepository(
+  sessionId: string,
+  repoRoot: string,
+): { sessionFile: string; content: string } {
+  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(sessionId)) {
+    throw new Error(`Invalid Codex session ID: ${sessionId}`);
+  }
+
+  const homeDir = process.env.HOME || os.homedir();
+  const searchDirs = [
+    path.join(homeDir, ".codex", "sessions"),
+    path.join(homeDir, ".codex", "archived_sessions"),
+  ];
+
+  const candidates: string[] = [];
+  const findSessionFile = (dir: string) => {
+    if (!fs.existsSync(dir)) return;
+    for (const entry of fs.readdirSync(dir, { withFileTypes: true })) {
+      if (entry.isDirectory()) {
+        findSessionFile(path.join(dir, entry.name));
+      } else if (entry.isFile()) {
+        // Real rollout filenames: rollout-<timestamp>-<uuid>.jsonl
+        const name = entry.name;
+        if (name.endsWith(".jsonl") && (name.includes(sessionId) || name === `${sessionId}.jsonl`)) {
+          candidates.push(path.join(dir, entry.name));
+        }
+      }
+    }
+  };
+  for (const searchDir of searchDirs) {
+    findSessionFile(searchDir);
+  }
+
+  if (candidates.length !== 1) {
+    throw new Error(
+      candidates.length === 0
+        ? `Codex session file not found for ID: ${sessionId}`
+        : `Multiple Codex session files found for ID: ${sessionId}`,
+    );
+  }
+
+  const sessionFile = candidates[0];
+  const fd = fs.openSync(sessionFile, "r");
+  const maxBytes = 4 * 1024 * 1024;
+  const buffer = Buffer.alloc(maxBytes);
+  let content: string;
+  try {
+    const bytesRead = fs.readSync(fd, buffer, 0, maxBytes, 0);
+    content = buffer.toString("utf8", 0, bytesRead);
+  } finally {
+    fs.closeSync(fd);
+  }
+
+  let repoMatches = false;
+  let sessionMatches = false;
+  for (const line of content.split("\n")) {
+    if (!line.trim()) continue;
+    try {
+      const entry = JSON.parse(line) as Record<string, unknown>;
+      if (entry.type === "session_meta") {
+        const payload = entry.payload as Record<string, unknown> | undefined;
+        // Parse session ID from payload.id
+        const entryId = (payload?.id as string) ?? (entry.id as string) ?? (entry.session_id as string);
+        if (entryId === sessionId) sessionMatches = true;
+        // Parse working directory from payload.cwd
+        const cwd = (payload?.cwd as string) ?? (entry.cwd as string);
+        if (cwd && (path.resolve(cwd) === path.resolve(repoRoot) || path.normalize(cwd) === path.normalize(repoRoot))) {
+          repoMatches = true;
+        }
+      }
+      if (entry.session_id === sessionId) sessionMatches = true;
+      if (entry.cwd && path.resolve(entry.cwd as string) === path.resolve(repoRoot)) repoMatches = true;
+    } catch {
+      // Ignore a trailing partial line
+    }
+  }
+
+  if (!sessionMatches || !repoMatches) {
+    throw new Error(`Codex session ${sessionId} does not belong to repository ${repoRoot}`);
+  }
+  return { sessionFile, content };
+}
+
+function deriveCodexActivity(line: string): string | undefined {
+  const trimmed = line.trim();
+  if (!trimmed.startsWith("{")) return undefined;
+  try {
+    const ev = JSON.parse(trimmed) as Record<string, unknown>;
+    if (ev.type === "item.started" && typeof ev.item === "object" && ev.item) {
+      const item = ev.item as Record<string, unknown>;
+      // Real Codex item events use item.tool_name or item.name
+      if (typeof item.tool_name === "string") return `Tool: ${item.tool_name}`;
+      if (typeof item.name === "string") return `Tool: ${item.name}`;
+      if (typeof item.type === "string") return `Item: ${item.type}`;
+    }
+    if (ev.type === "item.completed") return "Item completed";
+    if (ev.type === "turn.completed") return "Turn completed";
+    if (ev.type === "thread.started") return "Thread started";
+    if (ev.type === "tool" && typeof ev.name === "string") return `Tool: ${ev.name}`;
+    if (ev.type === "message") return "Message...";
+    if (ev.type === "completion") return "Completion...";
+    if (ev.type === "initialization") return "Initializing...";
+    // Real Codex error may be in message or error field
+    if (ev.type === "error") {
+      const msg = typeof ev.message === "string" ? ev.message : typeof ev.error === "string" ? ev.error : undefined;
+      if (msg) return `Error: ${msg}`;
+    }
+  } catch {
+    // fall through
+  }
+  return undefined;
+}
+
+function extractCodexFailureReason(output: string): string | undefined {
+  for (const line of output.split("\n")) {
+    const trimmed = line.trim();
+    if (!trimmed.startsWith("{")) continue;
+    try {
+      const parsed = JSON.parse(trimmed) as Record<string, unknown>;
+      // Real Codex error includes message field
+      if (parsed.type === "error") {
+        const msg = typeof parsed.message === "string" && parsed.message.trim() ? parsed.message :
+                     typeof parsed.error === "string" && parsed.error.trim() ? parsed.error :
+                     typeof parsed.message === "object" && parsed.message !== null && typeof (parsed.message as Record<string, unknown>).content === "string" ? (parsed.message as Record<string, unknown>).content as string :
+                     typeof parsed.error === "object" && parsed.error !== null && typeof (parsed.error as Record<string, unknown>).content === "string" ? (parsed.error as Record<string, unknown>).content as string : undefined;
+        if (msg) return msg.trim();
+      }
+    } catch {
+      // skip
+    }
+  }
+  return undefined;
+}
+
+function codexExecutorEnv(): NodeJS.ProcessEnv {
+  const env = executorEnv();
+  delete env.CODEX_THREAD_ID;
+  return env;
+}
+
+class CodexAdapter implements ExecutorAdapter {
+  readonly kind: ExecutorKind = "codex";
+
+  async probe(config: ExecutorConfig = {}): Promise<boolean> {
+    try {
+      const res = spawnSync(resolveExecutable(config, "codex"), ["--version"], {
+        stdio: "ignore",
+        env: codexExecutorEnv(),
+      });
+      return res.status === 0 && !res.error;
+    } catch {
+      return false;
+    }
+  }
+
+  async start(opts: ExecutorStartOpts): Promise<ExecutorResult> {
+    const bin = resolveExecutable(opts.config, "codex");
+    const prompt = buildStartPrompt(opts);
+
+    // Codex CLI accepts prompt via stdin using '-' argument
+    const args: string[] = [
+      "exec",
+      "--json",
+      "--sandbox", "workspace-write",
+      "--cd", opts.repoRoot,
+      ...(opts.config.model ? ["--model", opts.config.model] : []),
+      "-",
+    ];
+
+    const startMs = Date.now();
+    const logFile = opts.logFilePath;
+    const attemptIndex = deriveAttemptIndex(logFile);
+    const sidecarTimer = startProgressSidecar(opts.runDir, "codex", opts.cycleIndex, attemptIndex, logFile);
+
+    let lineBuffer = "";
+
+    const child = await spawnStreamingChild(bin, args, {
+      cwd: opts.repoRoot,
+      env: codexExecutorEnv(),
+      input: prompt,
+      signal: opts.signal,
+      processGroup: true,
+      logFilePath: logFile,
+      onProgress: ({ pid, bytesLogged }) => {
+        opts.onActivity?.();
+        updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() });
+      },
+      onStdoutData: (chunk: string) => {
+        lineBuffer += chunk;
+        while (true) {
+          const nl = lineBuffer.indexOf("\n");
+          if (nl < 0) break;
+          const line = lineBuffer.slice(0, nl);
+          lineBuffer = lineBuffer.slice(nl + 1);
+          const activity = deriveCodexActivity(line);
+          if (activity) {
+            const isToolActivity = !activity.startsWith("Error:");
+            if (isToolActivity) {
+              updateSidecar(opts.runDir, { activitySummary: activity, lastActivityAt: nowIso() });
+            } else {
+              updateSidecar(opts.runDir, { lastActivityAt: nowIso() });
+            }
+          }
+        }
+      },
+    });
+
+    let failed = child.status !== 0 || !!child.error;
+    const capturedId = extractCodexSessionId(child.stdout) ?? extractCodexSessionId(child.stderr);
+    let failureReason = child.error?.message
+      ?? extractCodexFailureReason(child.stdout)
+      ?? extractCodexFailureReason(child.stderr);
+
+    if (!failed && !capturedId) {
+      failed = true;
+      failureReason = "Codex execution completed successfully but yielded no resumable session ID.";
+    }
+
+    stopProgressSidecar(opts.runDir, sidecarTimer, failed);
+    const durationMs = Date.now() - startMs;
+    const handle: CodexSessionHandle | undefined = failed
+      ? undefined
+      : { kind: "codex", sessionId: capturedId! };
+    const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : "");
+
+    return {
+      exitCode: child.status,
+      report,
+      durationMs,
+      ...(handle ? { sessionHandle: handle } : {}),
+      failed,
+      ...(failureReason ? { failureReason } : {}),
+    };
+  }
+
+  async resume(opts: ExecutorResumeOpts): Promise<ExecutorResult> {
+    const handle = opts.handle as CodexSessionHandle;
+    const bin = resolveExecutable(opts.config, "codex");
+    const prompt = buildFixPrompt(opts);
+
+    // resume uses -c sandbox_mode= override (--sandbox is rejected by codex exec resume)
+    const args: string[] = [
+      "exec", "resume",
+      "-c", 'sandbox_mode="workspace-write"',
+      "--json",
+      ...(opts.config.model ? ["--model", opts.config.model] : []),
+      handle.sessionId,
+      "-",
+    ];
+
+    const startMs = Date.now();
+    const logFile = opts.logFilePath;
+    const attemptIndex = deriveAttemptIndex(logFile);
+    const sidecarTimer = startProgressSidecar(opts.runDir, "codex", opts.cycleIndex, attemptIndex, logFile);
+
+    let lineBuffer = "";
+
+    const child = await spawnStreamingChild(bin, args, {
+      cwd: opts.repoRoot,
+      env: codexExecutorEnv(),
+      input: prompt,
+      signal: opts.signal,
+      processGroup: true,
+      logFilePath: logFile,
+      onProgress: ({ pid, bytesLogged }) => {
+        opts.onActivity?.();
+        updateSidecar(opts.runDir, { pid, logBytes: bytesLogged, lastActivityAt: nowIso() });
+      },
+      onStdoutData: (chunk: string) => {
+        lineBuffer += chunk;
+        while (true) {
+          const nl = lineBuffer.indexOf("\n");
+          if (nl < 0) break;
+          const line = lineBuffer.slice(0, nl);
+          lineBuffer = lineBuffer.slice(nl + 1);
+          const activity = deriveCodexActivity(line);
+          if (activity) {
+            const isToolActivity = !activity.startsWith("Error:");
+            if (isToolActivity) {
+              updateSidecar(opts.runDir, { activitySummary: activity, lastActivityAt: nowIso() });
+            } else {
+              updateSidecar(opts.runDir, { lastActivityAt: nowIso() });
+            }
+          }
+        }
+      },
+    });
+
+    const failed = child.status !== 0 || !!child.error;
+    stopProgressSidecar(opts.runDir, sidecarTimer, failed);
+    const durationMs = Date.now() - startMs;
+
+    const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : "");
+    const failureReason = child.error?.message
+      ?? extractCodexFailureReason(child.stdout)
+      ?? extractCodexFailureReason(child.stderr);
+    return {
+      exitCode: child.status,
+      report,
+      durationMs,
+      sessionHandle: handle,
+      failed,
+      ...(failureReason ? { failureReason } : {}),
+    };
+  }
+
+  async cancel(_handle: ExecutorSessionHandle): Promise<void> {
+  }
+}
+
+
 // Factory
 // ---------------------------------------------------------------------------

@@ -1183,6 +1538,8 @@ export function createExecutorAdapter(kind: ExecutorKind): ExecutorAdapter {
      return new ClaudeAdapter();
    case "agy":
      return new AgyAdapter();
+    case "codex":
+      return new CodexAdapter();
    default: {
      const _exhaustive: never = kind;
      throw new Error(`Unknown executor kind: ${_exhaustive}`);