codex.ts

// src/executor/codex.ts — Codex CLI-subprocess executor + pure event mapper.
//
// Drives the installed `codex` CLI (v0.139.0) via `codex exec --json` and normalizes
// its line-delimited JSON (JSONL) ThreadEvent stream into the canonical ActionEvent
// union (src/shared/events.ts). The mapper is PURE so it can be unit-tested against a
// captured fixture without spawning a process.
//
// Spawn invariants (see BUILD_GUIDE Executor Adapter, steps 3 + gotchas):
//   - stdin MUST be /dev/null (stdio[0]='ignore') or `codex exec` can block on a TTY.
//   - Parse STDOUT ONLY as JSONL; stderr carries logs/skill-load errors/telemetry — drain it.
//   - Tolerant parse: skip lines not starting with '{', wrap JSON.parse in try/catch.
//   - Unknown event/item types -> map to null (skip), never throw.
import { spawn } from 'node:child_process';
import { finalTerminalEvent } from './exitAccounting';
import { createInterface } from 'node:readline';
import {
  ActionEvent,
  ExecutorRunOptions,
  Executor,
  newRunId,
} from '../../shared/events';
import { executorPathEnv, resolveBin } from './resolveBin';
import { armSigkillEscalation } from './abortKill';

// Resolve the codex binary portably: RORO_CODEX_BIN override -> PATH -> common install dirs ->
// bare 'codex' (spawn ENOENTs loud). Handles packaged Electron stripping PATH without a hardcoded path.
const CODEX_BIN = resolveBin('codex', process.env.RORO_CODEX_BIN);
/** Grace after abort's SIGTERM before SIGKILL, so a hung child can't hold the executor slot. */
const SIGKILL_GRACE_MS = 1000;

/**
 * Pure mapper: one Codex ThreadEvent object -> at most one canonical ActionEvent.
 *
 * Maps the CURRENT codex v0.139.0 JSONL shapes (verified live, see __fixtures__):
 *   thread.started{thread_id}                       -> run.started(threadId)
 *   turn.started                                    -> turn.started
 *   item.started|completed / type=reasoning         -> reasoning (only on completed; carries text)
 *   item.started|completed / type=command_execution -> command(status, command, output, exitCode)
 *   item.started|completed / type=file_change       -> file_change(status, files[])
 *   item.started|completed / type=agent_message     -> message (only on completed; final text)
 *   item.started|completed / type=mcp_tool_call     -> tool(status, server, tool)
 *   item.started|completed / type=web_search        -> tool(status, tool='web_search', summary=query)
 *   item.* / type=error                             -> run.failed
 *   turn.completed{usage}                           -> run.completed
 *   turn.failed{error}                              -> run.failed
 *   error{message}                                  -> run.failed
 * Anything else (todo_list, unknown item/event types) -> null (skip; forward-compat).
 *
 * Item status note (v0.139.0): item.started carries status:'in_progress'; item.completed
 * carries 'completed' | 'failed'. We derive the canonical ActionEvent status from the
 * envelope event type (started -> 'started') plus the item's own failure signal on completion.
 */
export function mapCodexThreadEvent(
  obj: unknown,
  runId: string,
): ActionEvent | null {
  const ts = Date.now();
  if (!obj || typeof obj !== 'object') return null;
  const ev = obj as Record<string, unknown>;
  if (typeof ev.type !== 'string') return null;

  switch (ev.type) {
    case 'thread.started': {
      const threadId =
        typeof ev.thread_id === 'string' ? ev.thread_id : undefined;
      return { kind: 'run.started', agent: 'codex', runId, threadId, ts };
    }

    case 'turn.started':
      return { kind: 'turn.started', runId, ts };

    case 'item.started':
    case 'item.completed': {
      const completed = ev.type === 'item.completed';
      const it = ev.item;
      if (!it || typeof it !== 'object') return null;
      const item = it as Record<string, unknown>;
      const itemType =
        typeof item.type === 'string' ? item.type : undefined;
      const itemId = typeof item.id === 'string' ? item.id : '';

      switch (itemType) {
        case 'reasoning': {
          // Reasoning carries meaningful text only on completion.
          if (!completed) return null;
          const text = typeof item.text === 'string' ? item.text : '';
          return { kind: 'reasoning', runId, itemId, text, ts };
        }

        case 'command_execution': {
          const command =
            typeof item.command === 'string' ? item.command : '';
          const output =
            typeof item.aggregated_output === 'string'
              ? item.aggregated_output
              : undefined;
          const exitCode =
            typeof item.exit_code === 'number' ? item.exit_code : undefined;
          // Failure derivation: trust the item's own status, and treat any
          // non-zero exit code as a failure. exit_code is null/absent on start.
          const itemStatus =
            typeof item.status === 'string' ? item.status : undefined;
          const failed =
            itemStatus === 'failed' || (exitCode != null && exitCode !== 0);
          const status: 'started' | 'completed' | 'failed' = !completed
            ? 'started'
            : failed
              ? 'failed'
              : 'completed';
          return {
            kind: 'command',
            runId,
            itemId,
            status,
            command,
            output,
            exitCode,
            ts,
          };
        }

        case 'file_change': {
          const rawChanges = Array.isArray(item.changes) ? item.changes : [];
          const files = rawChanges
            .filter(
              (c): c is Record<string, unknown> =>
                !!c && typeof c === 'object',
            )
            .map((c) => ({
              path: typeof c.path === 'string' ? c.path : '',
              op: normalizeFileOp(c.kind),
            }));
          const itemStatus =
            typeof item.status === 'string' ? item.status : undefined;
          const status: 'started' | 'completed' | 'failed' = !completed
            ? 'started'
            : itemStatus === 'failed'
              ? 'failed'
              : 'completed';
          return { kind: 'file_change', runId, itemId, status, files, ts };
        }

        case 'mcp_tool_call': {
          const status: 'started' | 'completed' | 'failed' = !completed
            ? 'started'
            : item.status === 'failed' || item.error != null
              ? 'failed'
              : 'completed';
          return {
            kind: 'tool',
            runId,
            itemId,
            status,
            server: typeof item.server === 'string' ? item.server : undefined,
            tool: typeof item.tool === 'string' ? item.tool : 'mcp',
            ts,
          };
        }

        case 'web_search': {
          const status: 'started' | 'completed' | 'failed' = completed
            ? 'completed'
            : 'started';
          return {
            kind: 'tool',
            runId,
            itemId,
            status,
            tool: 'web_search',
            summary: typeof item.query === 'string' ? item.query : undefined,
            ts,
          };
        }

        case 'agent_message': {
          // Final assistant text only arrives on completion.
          if (!completed) return null;
          const text = typeof item.text === 'string' ? item.text : '';
          return { kind: 'message', runId, text, ts };
        }

        case 'error': {
          const error =
            typeof item.message === 'string' ? item.message : 'item error';
          return { kind: 'run.failed', runId, ok: false, error, ts };
        }

        // todo_list and any unknown item type -> skip (forward-compat).
        default:
          return null;
      }
    }

    case 'turn.completed':
      return { kind: 'run.completed', runId, ok: true, usage: ev.usage, ts };

    case 'turn.failed': {
      const err = ev.error;
      const error =
        err && typeof err === 'object' && typeof (err as Record<string, unknown>).message === 'string'
          ? ((err as Record<string, unknown>).message as string)
          : 'turn.failed';
      return { kind: 'run.failed', runId, ok: false, error, ts };
    }

    case 'error': {
      const error =
        typeof ev.message === 'string' ? ev.message : 'codex error';
      return { kind: 'run.failed', runId, ok: false, error, ts };
    }

    // Unknown top-level event type -> skip.
    default:
      return null;
  }
}

function normalizeFileOp(kind: unknown): 'add' | 'update' | 'delete' {
  return kind === 'add' || kind === 'delete' ? kind : 'update';
}

/** Pure arg builder (exported for tests). `readOnly` maps to codex's read-only sandbox — used by
 *  the fact-proposal ask, which must never carry write capability. */
export function codexExecArgs(opts: Pick<ExecutorRunOptions, 'repo' | 'prompt' | 'readOnly'>): string[] {
  return ['exec', '--json', '--skip-git-repo-check', '-s', opts.readOnly ? 'read-only' : 'workspace-write', '-C', opts.repo, opts.prompt];
}

/**
 * Spawn the codex CLI and yield normalized ActionEvents.
 *
 * `codex exec --json --skip-git-repo-check -s workspace-write -C <repo> "<prompt>" </dev/null`
 * stdin = 'ignore' (== /dev/null), stdout = JSONL parsed via readline, stderr drained.
 */
/** codex has NO single authoritative final-result string (claude uses the SDK's `m.result` — see claude.ts).
 *  Its LAST assistant `agent_message` is NORMALLY the final answer — a best-effort heuristic, not a guarantee
 *  (a run ending on a terse/mid-work message would surface that instead) — so stitch it onto run.completed as
 *  `finalText`: the truthful memory receipt (speechBubble.ts), the answer caption, and the "job done"
 *  notification read it. The receipt claims RECALL not answer content, so the heuristic never makes it lie.
 *  Pass-through when it's not run.completed, when finalText is already set (NEVER clobber the SDK/claude value),
 *  or when there was no message (leaving finalText empty correctly keeps the receipt suppressed). */
export function attachCodexFinalText(ev: ActionEvent, lastMessageText: string | undefined): ActionEvent {
  return ev.kind === 'run.completed' && !ev.finalText && lastMessageText
    ? { ...ev, finalText: lastMessageText }
    : ev;
}

/** Batch form for replay contexts (the proposal-eval fixture): track the last message, attach it. runCodex's
 *  streaming loop mirrors this exact rule inline (tracking `lastMessageText` + attachCodexFinalText) so the
 *  two paths never diverge — the proposal eval stays representative of production. */
export function stitchCodexFinalText(events: readonly ActionEvent[]): ActionEvent[] {
  let last: string | undefined;
  return events.map((ev) => {
    if (ev.kind === 'message' && ev.text) last = ev.text;
    return attachCodexFinalText(ev, last);
  });
}

export async function* runCodex(
  opts: ExecutorRunOptions,
): AsyncIterable<ActionEvent> {
  const runId = newRunId();

  if (opts.signal?.aborted) {
    yield { kind: 'run.failed', runId, ok: false, error: 'aborted', ts: Date.now() };
    return;
  }

  const args = codexExecArgs(opts);

  const child = spawn(CODEX_BIN, args, {
    stdio: ['ignore', 'pipe', 'pipe'], // stdin=/dev/null; no TTY hang
    env: { ...process.env, PATH: executorPathEnv(CODEX_BIN, process.env) },
    signal: opts.signal,
  });
  // {signal} sends SIGTERM on abort; escalate to SIGKILL if the child ignores it.
  armSigkillEscalation(child, opts.signal, SIGKILL_GRACE_MS);

  // Drain stderr (plain logs / telemetry, NOT JSONL) — but keep a TAIL to diagnose a crash-without-a-result.
  let stderrTail = '';
  child.stderr?.on('data', (d) => { stderrTail = (stderrTail + String(d)).slice(-1000); });

  // Surface a clean run.failed if the binary cannot be spawned (ENOENT) or the
  // AbortSignal kills it — instead of letting the process error escape the loop.
  let spawnError: Error | null = null;
  child.on('error', (err) => {
    spawnError = err;
  });

  // Capture the child's EXIT so a nonzero/killed termination with NO JSON terminal event fails loud (c5),
  // instead of the orchestrator synthesizing a fabricated run.completed (and persisting a false success).
  const closed = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => {
    child.on('close', (code, signal) => resolve({ code, signal }));
  });
  let emittedTerminal = false;
  let lastMessageText: string | undefined; // the last agent_message → stitched onto run.completed as finalText

  const rl = createInterface({ input: child.stdout!, crlfDelay: Infinity });
  try {
    for await (const line of rl) {
      if (opts.signal?.aborted) break;
      const s = line.trim();
      if (!s || s[0] !== '{') continue; // skip blanks / non-JSON banner lines
      let obj: unknown;
      try {
        obj = JSON.parse(s);
      } catch {
        continue; // tolerate partial/garbage lines
      }
      const mapped = mapCodexThreadEvent(obj, runId);
      if (mapped) {
        if (mapped.kind === 'message' && mapped.text) lastMessageText = mapped.text;
        if (mapped.kind === 'run.completed' || mapped.kind === 'run.failed') emittedTerminal = true;
        // Streaming mirror of stitchCodexFinalText: attach the last agent_message as run.completed.finalText.
        yield attachCodexFinalText(mapped, lastMessageText);
      }
    }
  } catch (e) {
    yield {
      kind: 'run.failed',
      runId,
      ok: false,
      error: messageOf(e),
      ts: Date.now(),
    };
    return;
  }

  if (opts.signal?.aborted) {
    yield { kind: 'run.failed', runId, ok: false, error: 'aborted', ts: Date.now() };
    return;
  }
  if (spawnError) {
    yield {
      kind: 'run.failed',
      runId,
      ok: false,
      error: messageOf(spawnError),
      ts: Date.now(),
    };
    return;
  }

  // The stream ended with NO terminal event, not aborted, no spawn error → account for the child's EXIT.
  // A nonzero/killed exit must fail loud (never read as success). The 5s race bounds a child that closed
  // stdout but lingers; an unknown exit (code null) is treated as failure (the safe direction).
  if (!emittedTerminal) {
    const exit = await Promise.race([
      closed,
      new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((r) =>
        setTimeout(() => r({ code: null, signal: null }), 5000),
      ),
    ]);
    const terminal = finalTerminalEvent(
      { runId, bin: 'codex', emittedTerminal, aborted: Boolean(opts.signal?.aborted), spawnError: false, code: exit.code, signal: exit.signal, stderrTail },
      Date.now(),
    );
    if (terminal) yield terminal;
  }
}

function messageOf(e: unknown): string {
  if (e instanceof Error) return e.message;
  return String(e);
}

export const CodexExecutor: Executor = {
  run(opts: ExecutorRunOptions): AsyncIterable<ActionEvent> {
    return runCodex(opts);
  },
};