gstack/test/helpers/codex-session-runner.ts

/**
 * Codex CLI subprocess runner for skill E2E testing.
 *
 * Spawns `codex exec` as a completely independent process, parses its JSONL
 * output, and returns structured results. Follows the same pattern as
 * session-runner.ts but adapted for the Codex CLI.
 *
 * Key differences from Claude session-runner:
 * - Uses `codex exec` instead of `claude -p`
 * - Output is JSONL with different event types (item.completed, turn.completed, thread.started)
 * - Uses `--json` flag instead of `--output-format stream-json`
 * - Needs temp HOME with skill installed at ~/.codex/skills/{skillName}/SKILL.md
 */

import * as fs from 'fs';
import * as path from 'path';
import * as os from 'os';
import { spawn } from 'child_process';
import { StringDecoder } from 'node:string_decoder';
import { hermeticChildEnv } from './hermetic-env';
import { extractSkillSections } from './skill-fixture';
import { killProcessGroup } from '../../scripts/test-strict-output';
import { CODEX_FRONTIER_MODEL } from '../../scripts/resolvers/constants';
import { estimateCostUsd, hasPricing } from '../../lib/pricing';

// --- Interfaces ---

export interface CodexResult {
  output: string;           // Full agent message text
  reasoning: string[];      // [codex thinking] blocks
  toolCalls: string[];      // [codex ran] commands
  tokens: number;           // Total tokens used
  exitCode: number;         // Process exit code
  durationMs: number;       // Wall clock time
  sessionId: string | null; // Thread ID for session continuity
  rawLines: string[];       // Raw JSONL lines for debugging
  stderr: string;           // Stderr output (skill loading errors, auth failures)
  usage?: CodexTokenUsage | null; // Summed turn.completed usage; null when no turn completed
  model?: string;           // The --model this session ran
}

/**
 * Token classes of Codex `turn.completed` usage. `cached_input` (cache reads)
 * and `cache_write_input` are subsets of `input`; `reasoning_output` is a
 * subset of `output`.
 */
export interface CodexTokenUsage {
  input: number;
  cached_input: number;
  cache_write_input: number;
  output: number;
  reasoning_output: number;
}

export const NO_CODEX_USAGE: CodexTokenUsage = Object.freeze({ input: 0, cached_input: 0, cache_write_input: 0, output: 0, reasoning_output: 0 });

const count = (value: unknown) => (typeof value === 'number' && Number.isFinite(value) && value > 0 ? value : 0);

/** Add one `turn.completed` usage object to a running total. */
export function addCodexTurnUsage(total: CodexTokenUsage, usage: Record<string, unknown> | undefined): CodexTokenUsage {
  return {
    input: total.input + count(usage?.input_tokens),
    cached_input: total.cached_input + count(usage?.cached_input_tokens),
    cache_write_input: total.cache_write_input + count(usage?.cache_write_input_tokens),
    output: total.output + count(usage?.output_tokens),
    reasoning_output: total.reasoning_output + count(usage?.reasoning_output_tokens),
  };
}

/**
 * A session's cost at lib/pricing.ts rates: cache reads at the cached-input
 * rate, all other input (cache writes included) at the input rate, output at
 * the output rate. Null when the model has no price row.
 */
export function codexUsageCostUsd(usage: CodexTokenUsage, model: string | undefined): number | null {
  if (!hasPricing(model)) return null;
  const cached = Math.min(usage.cached_input, usage.input);
  return estimateCostUsd({ input: usage.input - cached, cached, output: usage.output }, model);
}

/** Existing pipe-drain allowance, separate from the model's work budget. */
export const CODEX_DRAIN_GRACE_MS = 5_000;

export class CodexHarnessError extends Error {
  constructor(message: string, readonly result?: CodexResult) {
    super(message);
    this.name = 'CodexHarnessError';
  }
}

// --- JSONL parser (ported from Python in codex/SKILL.md.tmpl) ---

export interface ParsedCodexJSONL {
  output: string;
  reasoning: string[];
  toolCalls: string[];
  tokens: number;
  sessionId: string | null;
  usage: CodexTokenUsage | null;
}

/**
 * Parse an array of JSONL lines from `codex exec --json` into structured data.
 * Pure function — no I/O, no side effects.
 *
 * Handles these Codex event types:
 * - thread.started → extract thread_id (session ID)
 * - item.completed → extract reasoning, agent_message, command_execution
 * - turn.completed → extract token usage
 */
export function parseCodexJSONL(lines: string[]): ParsedCodexJSONL {
  const outputParts: string[] = [];
  const reasoning: string[] = [];
  const toolCalls: string[] = [];
  let tokens = 0;
  let sessionId: string | null = null;
  let usage: CodexTokenUsage | null = null;

  for (const line of lines) {
    if (!line.trim()) continue;
    try {
      const obj = JSON.parse(line);
      const t = obj.type || '';

      if (t === 'thread.started') {
        const tid = obj.thread_id || '';
        if (tid) sessionId = tid;
      } else if (t === 'item.completed' && obj.item) {
        const item = obj.item;
        const itype = item.type || '';
        const text = item.text || '';

        if (itype === 'reasoning' && text) {
          reasoning.push(text);
        } else if (itype === 'agent_message' && text) {
          outputParts.push(text);
        } else if (itype === 'command_execution') {
          const cmd = item.command || '';
          if (cmd) toolCalls.push(cmd);
        }
      } else if (t === 'turn.completed') {
        const turn = obj.usage || {};
        tokens += (turn.input_tokens || 0) + (turn.output_tokens || 0);
        usage = addCodexTurnUsage(usage ?? NO_CODEX_USAGE, turn);
      }
    } catch { /* skip malformed lines */ }
  }

  return {
    output: outputParts.join('\n'),
    reasoning,
    toolCalls,
    tokens,
    sessionId,
    usage,
  };
}

// --- Skill installation helper ---

/**
 * Install a SKILL.md into a temp HOME directory for Codex to discover.
 * Creates ~/.codex/skills/{skillName}/SKILL.md in the temp HOME and copies
 * agents/openai.yaml when present so Codex sees the same metadata as a real install.
 *
 * When `sections` is provided, the installed SKILL.md is an EXTRACTION
 * (frontmatter + the named `## <section>` blocks via
 * test/helpers/skill-fixture.ts) instead of the full 1000-1900-line file —
 * CLAUDE.md: "E2E test fixtures: extract, don't copy". Omit `sections` only
 * when the test's purpose is to validate the real generated artifact itself
 * (e.g., codex-discover-skill asserts the full SKILL.md loads without
 * "invalid" / "Skipped loading" stderr from Codex).
 *
 * Returns the temp HOME path. Caller is responsible for cleanup.
 */
export function installSkillToTempHome(
  skillDir: string,
  skillName: string,
  tempHome?: string,
  sections?: string[],
  runtimeRoot?: string,
): string {
  const home = tempHome || fs.mkdtempSync(path.join(os.tmpdir(), 'codex-e2e-'));
  const destDir = path.join(home, '.codex', 'skills', skillName);
  fs.mkdirSync(destDir, { recursive: true });

  const srcSkill = path.join(skillDir, 'SKILL.md');
  if (sections && sections.length > 0) {
    // extractSkillSections throws loudly on a missing file or renamed
    // section — a fixture is never silently written empty.
    fs.writeFileSync(path.join(destDir, 'SKILL.md'), extractSkillSections(skillDir, sections));
  } else {
    // A missing/unreadable full fixture must fail just like an extraction.
    // Preserve copyFileSync's filesystem diagnostic; an agent can mention a
    // nonexistent skill in its response and otherwise pass discovery checks.
    fs.copyFileSync(srcSkill, path.join(destDir, 'SKILL.md'));
  }
  if (runtimeRoot) {
    // The temp HOME has no installed gstack runtime; point runtime helpers at the one under test.
    const installed = path.join(destDir, 'SKILL.md');
    fs.writeFileSync(installed, fs.readFileSync(installed, 'utf8').replaceAll('~/.codex/skills/gstack', runtimeRoot));
  }

  const srcOpenAIYaml = path.join(skillDir, 'agents', 'openai.yaml');
  if (fs.existsSync(srcOpenAIYaml)) {
    const destAgentsDir = path.join(destDir, 'agents');
    fs.mkdirSync(destAgentsDir, { recursive: true });
    fs.copyFileSync(srcOpenAIYaml, path.join(destAgentsDir, 'openai.yaml'));
  }

  return home;
}

// --- Spawn core ---

/**
 * Spawn `codex <args>` in its own process group and collect its JSONL until
 * exit, the deadline or an abort, with the existing drain grace. Shared by
 * runCodexSkill and the Codex host (codex-host-runner.ts). `onLine` sees each
 * JSONL line as it arrives with the elapsed milliseconds since `startTime`.
 */
export async function execCodex(args: string[], opts: {
  cwd: string;
  env: Record<string, string>;
  startTime: number;
  deadline: number;
  signal?: AbortSignal;
  onLine?: (line: string, elapsedMs: number) => void;
  /** Called with the group-owning pid right after spawn; a throw kills the group. */
  onSpawn?: (pid: number) => void;
}): Promise<CodexResult> {
  const { cwd, env, startTime, deadline, signal, onLine } = opts;
  const proc = spawn('codex', args, {
    cwd,
    stdio: ['ignore', 'pipe', 'pipe'],
    detached: process.platform !== 'win32',
    env,
  });
  if (opts.onSpawn) {
    try {
      if (proc.pid === undefined) throw new Error('Codex did not start');
      opts.onSpawn(proc.pid);
    } catch (error) {
      killProcessGroup(proc, 'SIGKILL');
      proc.stdout!.destroy();
      proc.stderr!.destroy();
      throw error;
    }
  }
  const collectedLines: string[] = [];
  // work --exit--> drain (<=5s) --> result
  //   \--timeout/abort--> kill group + close pipes --> timeout result
  // Exit status comes from 'exit'; a forced drain after exit 0 is an error.
  let stdoutBuffer = '';
  let stderr = '';
  let exitCode: number | undefined;
  let timedOut = false;
  let drainExpired = false;
  let streamError: { stream: 'stdout' | 'stderr'; error: Error } | undefined;
  let spawnError: Error | undefined;
  let stdoutDone = false;
  let stderrDone = false;
  let stdoutEnded = false;
  let stderrEnded = false;
  let finalized = false;
  const stdoutDecoder = new StringDecoder('utf8');
  const stderrDecoder = new StringDecoder('utf8');
  let workTimer: ReturnType<typeof setTimeout> | undefined;
  let drainTimer: ReturnType<typeof setTimeout> | undefined;
  let finish!: () => void;
  const finished = new Promise<void>((resolve) => { finish = resolve; });

  const maybeFinish = () => {
    if (exitCode !== undefined && stdoutDone && stderrDone) finish();
  };
  const closePipes = () => {
    // Destroy both streams: canceling stdout alone cannot release an
    // inherited stderr pipe held by a descendant in another process group.
    proc.stdout!.destroy();
    proc.stderr!.destroy();
  };
  const expireDrain = () => {
    drainExpired = !stdoutDone || !stderrDone;
    killProcessGroup(proc, 'SIGKILL');
    closePipes();
    // Also bound the rare case where a failed group kill produces no exit.
    finish();
  };
  const stopRun = () => {
    // A real auth/crash exit that preceded a pipe stall stays a real exit.
    if (exitCode === undefined) timedOut = true;
    else drainExpired ||= !stdoutDone || !stderrDone;
    killProcessGroup(proc, 'SIGKILL');
    closePipes();
    clearTimeout(drainTimer);
    drainTimer = setTimeout(expireDrain, CODEX_DRAIN_GRACE_MS);
    maybeFinish();
  };
  const onExit = (code: number | null, exitSignal: NodeJS.Signals | null) => {
    exitCode = code ?? (exitSignal ? 128 + (os.constants.signals[exitSignal] ?? 0) : 1);
    clearTimeout(workTimer);
    clearTimeout(drainTimer);
    // 'exit' means the child is gone; 'close' also waits for descendant
    // pipes. Start the drain allowance at actual exit, never at close.
    drainTimer = setTimeout(expireDrain, CODEX_DRAIN_GRACE_MS);
    maybeFinish();
  };
  const onSpawnError = (error: Error) => {
    if (finalized) return;
    spawnError = error;
    exitCode = 1;
    closePipes();
    finish();
  };
  // Closure releases lifecycle waits, but only 'end' proves all bytes were
  // drained. A destroyed pipe can emit 'close' without either EOF or error.
  const onStdoutDone = () => { if (!finalized) { stdoutDone = true; maybeFinish(); } };
  const onStderrDone = () => { if (!finalized) { stderrDone = true; maybeFinish(); } };
  const onStdoutEnd = () => { if (!finalized) { stdoutBuffer += stdoutDecoder.end(); stdoutEnded = true; onStdoutDone(); } };
  const onStderrEnd = () => { if (!finalized) { stderr += stderrDecoder.end(); stderrEnded = true; onStderrDone(); } };
  const onStreamError = (stream: 'stdout' | 'stderr', error: Error) => {
    if (!finalized) streamError ??= { stream, error };
  };
  const onStdout = (chunk: Buffer) => {
    if (finalized) return;
    stdoutBuffer += stdoutDecoder.write(chunk);
    const lines = stdoutBuffer.split('\n');
    stdoutBuffer = lines.pop() || '';
    for (const line of lines) {
      if (!line.trim()) continue;
      collectedLines.push(line);
      onLine?.(line, Date.now() - startTime);
      try {
        const event = JSON.parse(line);
        if (event.type === 'item.completed' && event.item) {
          const item = event.item;
          const elapsed = Math.round((Date.now() - startTime) / 1000);
          if (item.type === 'command_execution' && item.command) {
            process.stderr.write(`  [codex ${elapsed}s] ran: ${item.command.slice(0, 100)}\n`);
          } else if (item.type === 'agent_message' && item.text) {
            process.stderr.write(`  [codex ${elapsed}s] message: ${item.text.slice(0, 100)}\n`);
          }
        }
      } catch { /* malformed JSONL is ignored by parseCodexJSONL too */ }
    }
  };
  const onStderr = (chunk: Buffer) => { if (!finalized) stderr += stderrDecoder.write(chunk); };

  proc.on('exit', onExit);
  proc.on('error', onSpawnError);
  proc.stdout!.on('data', onStdout);
  proc.stderr!.on('data', onStderr);
  proc.stdout!.on('end', onStdoutEnd).on('close', onStdoutDone).on('error', error => onStreamError('stdout', error));
  proc.stderr!.on('end', onStderrEnd).on('close', onStderrDone).on('error', error => onStreamError('stderr', error));
  signal?.addEventListener('abort', stopRun, { once: true });
  workTimer = setTimeout(stopRun, Math.max(0, deadline - Date.now()));
  if (signal?.aborted || Date.now() >= deadline) stopRun();

  try {
    await finished;
    if (stdoutBuffer.trim()) collectedLines.push(stdoutBuffer);
    const parsed = parseCodexJSONL(collectedLines);
    const result: CodexResult = {
      ...parsed,
      exitCode: timedOut ? 124 : exitCode ?? 1,
      durationMs: Date.now() - startTime,
      rawLines: collectedLines,
      stderr,
    };
    if (stderr.trim()) process.stderr.write(`  [codex stderr] ${stderr.trim().slice(0, 200)}\n`);
    if (spawnError) throw new CodexHarnessError(`Could not start Codex: ${spawnError.message}`, result);
    const incompleteStreams = [!stdoutEnded && 'stdout', !stderrEnded && 'stderr'].filter(Boolean).join(' and ');
    if (result.exitCode === 0 && (drainExpired || streamError || incompleteStreams)) {
      throw new CodexHarnessError(
        drainExpired ? `Codex output drain exceeded ${CODEX_DRAIN_GRACE_MS}ms after exit 0`
          : streamError ? `Codex ${streamError.stream} stream failed: ${streamError.error.message}`
          : `Codex ${incompleteStreams} closed before EOF after exit 0`,
        result,
      );
    }
    return result;
  } finally {
    finalized = true;
    clearTimeout(workTimer);
    clearTimeout(drainTimer);
    signal?.removeEventListener('abort', stopRun);
    // Child exit and closed pipes do not prove its descendants exited:
    // background tools may redirect both streams. Reap the owned group on
    // every result, including ordinary success and genuine process failure.
    killProcessGroup(proc, 'SIGKILL');
    closePipes();
    proc.removeListener('exit', onExit);
    // Retain the harmless error listener through stream teardown so a late
    // OS spawn error cannot turn into an unhandled event after cleanup.
    proc.stdout!.removeListener('data', onStdout);
    proc.stderr!.removeListener('data', onStderr);
  }
}

// --- Main runner ---

/**
 * Run a Codex skill via `codex exec` and return structured results.
 *
 * Spawns codex in a temp HOME with the skill installed, parses JSONL output,
 * and returns a CodexResult. Skips gracefully if codex binary is not found.
 */
export async function runCodexSkill(opts: {
  skillDir: string;         // Path to skill directory containing SKILL.md
  prompt: string;           // What to ask Codex to do with the skill
  timeoutMs?: number;       // Default 300000 (5 min)
  cwd?: string;             // Working directory
  skillName?: string;       // Skill name for installation (default: dirname)
  sandbox?: string;         // Sandbox mode (default: 'read-only')
  sections?: string[];      // Install only these `## <section>` blocks (extract, don't copy)
  model?: string;           // Exact Codex model ID (passed with --model)
  configOverrides?: string[]; // TOML key=value overrides (passed with -c)
  ignoreUserConfig?: boolean; // Add --ignore-user-config; auth still comes from CODEX_HOME
  signal?: AbortSignal;     // Abort the process group when an enclosing eval expires
  runtimeRoot?: string;     // gstack runtime that ~/.codex/skills/gstack helper paths resolve to
}): Promise<CodexResult> {
  const {
    skillDir,
    prompt,
    timeoutMs = 300_000,
    cwd,
    skillName,
    sandbox = 'read-only',
    sections,
    model,
    configOverrides = [],
    ignoreUserConfig = false,
    signal,
    runtimeRoot,
  } = opts;

  const startTime = Date.now();
  const deadline = startTime + timeoutMs;
  const name = skillName || path.basename(skillDir) || 'gstack';

  const emptyResult = (exitCode: number, output = ''): CodexResult => ({
    output, reasoning: [], toolCalls: [], tokens: 0, exitCode,
    durationMs: Date.now() - startTime, sessionId: null, rawLines: [], stderr: '', usage: null,
  });
  if (signal?.aborted || Date.now() >= deadline) return emptyResult(124);

  // Preflight and setup are part of the budget, not extra time before it.
  // Bun's implicit child environment retains the launch-time PATH. Use the
  // current environment so preflight and the actual spawn see the same shims.
  const whichResult = Bun.spawnSync(['which', 'codex'], {
    env: process.env,
    timeout: Math.max(1, Math.min(30_000, deadline - Date.now())),
  });
  if (signal?.aborted || Date.now() >= deadline) return emptyResult(124);
  if (whichResult.exitCode !== 0) {
    if (whichResult.signalCode) throw new CodexHarnessError('Codex binary lookup did not complete');
    return emptyResult(-1, 'SKIP: codex binary not found');
  }

  // Set up temp HOME with skill installed
  const tempHome = fs.mkdtempSync(path.join(os.tmpdir(), 'codex-e2e-'));
  const realHome = os.homedir();

  try {
    installSkillToTempHome(skillDir, name, tempHome, sections, runtimeRoot);

    // Copy authentication only. Copying the whole operator ~/.codex tree leaks
    // plugins, MCP servers, rules, memories, and skills into a supposedly
    // hermetic E2E; required private MCPs can then fail before the model starts.
    const realCodexConfig = process.env.CODEX_HOME || path.join(realHome, '.codex');
    const tempCodexDir = path.join(tempHome, '.codex');
    if (fs.existsSync(realCodexConfig)) {
      for (const entry of ['auth.json']) {
        const src = path.join(realCodexConfig, entry);
        const dst = path.join(tempCodexDir, entry);
        if (fs.existsSync(src) && !fs.existsSync(dst)) {
          fs.cpSync(src, dst, { recursive: true });
        }
      }
    }

    if (signal?.aborted || Date.now() >= deadline) return emptyResult(124);

    // Build codex exec command.
    // --skip-git-repo-check: newer codex CLIs refuse exec in an untrusted
    // non-git directory ("Not inside a trusted directory and
    // --skip-git-repo-check was not specified") — our temp skill dirs are
    // exactly that. Empirically verified against codex on this machine.
    const args = ['exec', '--json', '-s', sandbox, '--skip-git-repo-check'];
    if (ignoreUserConfig) args.push('--ignore-user-config');
    const sessionModel = model ?? process.env.GSTACK_CODEX_MODEL ?? CODEX_FRONTIER_MODEL;
    args.push('--model', sessionModel);
    for (const override of configOverrides) args.push('-c', override);
    args.push(prompt);

    // Spawn codex with temp HOME so it discovers our installed skill.
    // Hermetic scrub (test/helpers/hermetic-env.ts) with codex's auth surface
    // re-admitted: codex auths from $HOME/.codex (copied into tempHome above)
    // plus OPENAI_API_KEY/CODEX_* when present. HOME override merges last.
    // node:child_process spawn with `detached` (own process group) — mirrors
    // session-runner.ts. Bun.spawn's bare proc.kill() signalled only codex
    // itself; command subprocesses codex spawned survived as orphans holding
    // our pipes open (the same blocked-drain hang the claude runner fixed —
    // this copy never inherited that fix until now).
    // Codex refuses to create its PATH aliases (codex-linux-sandbox among
    // them) when CODEX_HOME sits under its temp dir, so the child gets a temp
    // dir beside CODEX_HOME instead of the shard TMPDIR that holds tempHome.
    const tempTmpDir = path.join(tempHome, 'tmp');
    fs.mkdirSync(tempTmpDir, { recursive: true });
    const result = await execCodex(args, {
      cwd: cwd || skillDir, startTime, deadline, signal,
      env: hermeticChildEnv(
        { HOME: tempHome, CODEX_HOME: tempCodexDir, TMPDIR: tempTmpDir },
        { extraAllow: ['OPENAI_API_KEY', 'CODEX_*'] },
      ),
    }).catch((error: unknown) => {
      if (error instanceof CodexHarnessError && error.result) error.result.model = sessionModel;
      throw error;
    });
    return { ...result, model: sessionModel };

  } finally {
    // Clean up temp HOME
    try { fs.rmSync(tempHome, { recursive: true, force: true }); } catch { /* non-fatal */ }
  }
}