test(parse-workers): cover the Codex worker path end to end

A mixed Claude + Codex corpus with forked rollouts in both creation orders,
asserting an identical payload, byte-identical cache shards and a byte-identical
codex-results.json between CODEBURN_PARSE_WORKERS=0 and =3, with the codex
discard count pinned above zero so the overlap path is really exercised.

Plus: a resumable rollout never reaching a worker (the decision line reports no
full parses pending after an append), the off-thread decode matching
parseCodexFileFull exactly including the cache entry it hands back, no cache file
written by the decode itself, no leaked threads, and the files-OR-bytes gate.

tests/parse-workers.test.ts

@@ -1,13 +1,15 @@
 import { spawnSync } from 'node:child_process'
-import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
+import { appendFile, mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { createHash } from 'node:crypto'
 
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
-import { decideParseWorkers, ParseWorkerPool, parseFilesInOrder } from '../src/parse-workers.js'
+import { decideParseWorkers, ParseWorkerPool, parseFilesInOrder, type ClaudeWorkerParse } from '../src/parse-workers.js'
 import { clearSessionCache, parseAllSessions, parseClaudeFileFull } from '../src/parser.js'
+import { parseCodexFileFull, type CodexFullParse } from '../src/providers/codex.js'
+import type { SessionSource } from '../src/providers/types.js'
 
 // Two full cold CLI parses of a multi-hundred-file corpus, plus in-process parses
 // that spawn real threads.
@@ -29,18 +31,30 @@ describe('decideParseWorkers', () => {
     expect(decideParseWorkers(BIG_PENDING, { cores: 16, availableBytes: 6 * 1024 ** 3 }, NO_ENV).workers).toBe(6)
     // The smallest machine that clears every gate still only earns 2 threads
     expect(decideParseWorkers({ files: 200, bytes: 300 * 1024 ** 2 }, { cores: 3, availableBytes: 4 * 1024 ** 3 }, NO_ENV).workers).toBe(2)
-    // Few enough files that MIN_FILES_PER_WORKER is the binding constraint
-    expect(decideParseWorkers({ files: 300, bytes: 6 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(6)
+    // 300 files earn 6, but the 6 GB behind them earn 30 — bytes win, then the
+    // memory budget caps it. A Codex corpus is exactly this shape.
+    expect(decideParseWorkers({ files: 300, bytes: 6 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(8)
+    // Few enough files AND bytes that MIN_FILES_PER_WORKER is the binding constraint
+    expect(decideParseWorkers({ files: 300, bytes: 700 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(6)
   })
 
   it('stays serial on low-spec machines and on warm/small parses', () => {
     expect(decideParseWorkers(BIG_PENDING, { cores: 2, availableBytes: 32 * 1024 ** 3 }, NO_ENV).workers).toBe(0)
     // A 4 GB box: availableMemory() always reads a little under the nominal size
     expect(decideParseWorkers(BIG_PENDING, { cores: 16, availableBytes: 3.9 * 1024 ** 3 }, NO_ENV).workers).toBe(0)
-    // Warm/incremental: a handful of appended files
-    expect(decideParseWorkers({ files: 12, bytes: 6 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
-    // Many files but almost no bytes behind them
-    expect(decideParseWorkers({ files: 5000, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
+    // Warm/incremental: neither gate reached
+    expect(decideParseWorkers({ files: 12, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
+  })
+
+  it('takes files OR bytes, so a few huge rollouts still parallelise', () => {
+    // 150 rollouts of 4 GB: under the file gate, far over the byte gate
+    expect(decideParseWorkers({ files: 150, bytes: 4 * 1024 ** 3 }, BIG_SYSTEM, NO_ENV).workers).toBe(8)
+    // Many small files: under the byte gate, over the file gate
+    expect(decideParseWorkers({ files: 5000, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(8)
+    // Neither: still serial
+    expect(decideParseWorkers({ files: 150, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).workers).toBe(0)
+    expect(decideParseWorkers({ files: 150, bytes: 10 * 1024 ** 2 }, BIG_SYSTEM, NO_ENV).reason)
+      .toContain('below 200 pending files and 210 MB pending')
   })
 
   it('honours CODEBURN_PARSE_WORKERS, which also bypasses the auto gates', () => {
@@ -127,6 +141,82 @@ async function writeResumedPair(claudeDir: string, tag: string, originalName: st
   )
 }
 
+type CodexTask = { n: number; at: string }
+
+/// One Codex rollout: session_meta plus a complete task cycle per entry. A
+/// token_count dedup key is namespaced by the FORK PARENT when there is one and
+/// keyed on the cumulative token breakdown, so a fork restating a parent's tasks
+/// emits exactly the parent's keys — the cross-file overlap that only the
+/// install-order check can resolve. The replayed tasks are timestamped well past
+/// `metaTs + 5s` on purpose: inside that window the parser drops replays outright
+/// and the dedup path would never be reached.
+function codexRollout(sessionId: string, cwd: string, tasks: CodexTask[], forkedFrom?: string, metaTs = '2026-05-04T09:00:00.000Z'): string {
+  const lines = [JSON.stringify({
+    type: 'session_meta',
+    timestamp: metaTs,
+    payload: {
+      cwd, originator: 'codex-cli', session_id: sessionId, model: 'gpt-5.3-codex',
+      ...(forkedFrom ? { forked_from_id: forkedFrom } : {}),
+    },
+  })]
+  lines.push(...codexTaskLines(tasks))
+  return lines.join('\n') + '\n'
+}
+
+function codexTaskLines(tasks: CodexTask[]): string[] {
+  const lines: string[] = []
+  for (const { n, at } of tasks) {
+    const ts = (s: number) => new Date(Date.parse(at) + s * 1000).toISOString()
+    lines.push(
+      JSON.stringify({ type: 'event_msg', timestamp: ts(0), payload: { type: 'task_started' } }),
+      JSON.stringify({ type: 'response_item', timestamp: ts(1), payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `codex task ${n}` }] } }),
+      JSON.stringify({ type: 'response_item', timestamp: ts(2), payload: { type: 'function_call', name: 'shell', call_id: `c${n}`, arguments: JSON.stringify({ command: `ls ${n}` }) } }),
+      JSON.stringify({ type: 'response_item', timestamp: ts(3), payload: { type: 'function_call_output', call_id: `c${n}` } }),
+      JSON.stringify({ type: 'event_msg', timestamp: ts(4), payload: { type: 'patch_apply_end', success: true, changes: { [`/tmp/cx/f${n}.ts`]: { unified_diff: '@@ -1 +1,2 @@\n-old\n+new\n+extra\n' } } } }),
+      JSON.stringify({ type: 'response_item', timestamp: ts(5), payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'y'.repeat(120) }] } }),
+      JSON.stringify({
+        type: 'event_msg', timestamp: ts(6),
+        payload: {
+          type: 'token_count',
+          info: {
+            last_token_usage: { input_tokens: 100, cached_input_tokens: 20, output_tokens: 50, reasoning_output_tokens: 10, total_tokens: 180 },
+            total_token_usage: { input_tokens: 100 * n, cached_input_tokens: 20 * n, output_tokens: 50 * n, reasoning_output_tokens: 10 * n, total_tokens: 160 * n },
+          },
+        },
+      }),
+      JSON.stringify({ type: 'event_msg', timestamp: ts(7), payload: { type: 'task_complete', duration_ms: 4000 } }),
+    )
+  }
+  return lines
+}
+
+async function writeCodexRollout(codexHome: string, day: string, name: string, body: string): Promise<string> {
+  const dir = join(codexHome, 'sessions', '2026', '05', day)
+  await mkdir(dir, { recursive: true })
+  const path = join(dir, `rollout-${name}.jsonl`)
+  await writeFile(path, body)
+  return path
+}
+
+/// A parent rollout and a fork that replays its tasks before adding its own.
+/// `parentFirst` flips which file is created first, since discovery follows
+/// directory order: the shared keys must land on whichever file the SERIAL loop
+/// reaches first, in either order.
+async function writeForkedCodexPair(codexHome: string, day: string, tag: string, parentFirst: boolean): Promise<void> {
+  const shared = [1, 2, 3].map(n => ({ n, at: `2026-05-04T09:${String(10 + n).padStart(2, '0')}:00.000Z` }))
+  const parent = codexRollout(`${tag}-parent`, `/tmp/cx${tag}`, shared)
+  const fork = codexRollout(
+    `${tag}-fork`,
+    `/tmp/cx${tag}`,
+    [...shared.map(t => ({ ...t, at: `2026-05-04T10:${String(10 + t.n).padStart(2, '0')}:00.000Z` })), { n: 4, at: '2026-05-04T10:30:00.000Z' }],
+    `${tag}-parent`,
+  )
+  const order: Array<[string, string]> = parentFirst
+    ? [[`${tag}-a-parent`, parent], [`${tag}-b-fork`, fork]]
+    : [[`${tag}-a-fork`, fork], [`${tag}-b-parent`, parent]]
+  for (const [name, body] of order) await writeCodexRollout(codexHome, day, name, body)
+}
+
 /// Cache shard file names carry a random nonce, so compare bodies keyed by
 /// `<provider>.<month>` instead of by file name.
 async function shardBodies(cacheDir: string): Promise<Record<string, string>> {
@@ -140,12 +230,19 @@ async function shardBodies(cacheDir: string): Promise<Record<string, string>> {
   return out
 }
 
+/// The Codex incremental cache is a single JSON file; both runs read the same
+/// rollouts, so it must come out identical byte for byte.
+async function codexResults(cacheDir: string): Promise<string | null> {
+  return readFile(join(cacheDir, 'codex-results.json'), 'utf-8').catch(() => null)
+}
+
 function runCli(args: string[], home: string, extraEnv: Record<string, string>) {
   return spawnSync(process.execPath, ['--import', 'tsx', 'src/cli.ts', ...args], {
     cwd: process.cwd(),
     env: {
       ...process.env,
       CLAUDE_CONFIG_DIR: join(home, '.claude'),
+      CODEX_HOME: join(home, '.codex'),
       CODEBURN_CACHE_DIR: join(home, '.cache', 'codeburn'),
       HOME: home,
       TZ: 'UTC',
@@ -196,6 +293,9 @@ describe('parallel cold parse', () => {
     const serialShards = await shardBodies(serialCache)
     expect(Object.keys(serialShards).length).toBeGreaterThan(0)
     expect(await shardBodies(parallelCache)).toEqual(serialShards)
+    // Byte-compared, not deep-equalled: the Codex cache is written entry by entry
+    // in install order, so the key order is itself a claim about that order.
+    expect(await codexResults(parallelCache)).toEqual(await codexResults(serialCache))
     return parallel
   }
 
@@ -222,23 +322,78 @@ describe('parallel cold parse', () => {
       .reduce((n, m) => n + Number(m[1]), 0)
     expect(overlaps).toBeGreaterThan(0)
   })
+
+  // Codex is the bigger half of a real cold parse, and its cross-file dedup is
+  // stronger than Claude's: a forked rollout replays its parent's token_count
+  // history under the PARENT's key namespace, so two files claim the same keys
+  // outright. Both fork orders are present, so install order decides which file
+  // keeps the shared tasks either way.
+  it('matches the serial parse for a mixed Claude + Codex corpus with forked rollouts', async () => {
+    await writeCorpus(join(home, '.claude'), 2, 6)
+    const codex = join(home, '.codex')
+    await writeForkedCodexPair(codex, '04', 'fwd', true)
+    await writeForkedCodexPair(codex, '05', 'rev', false)
+    for (const n of range(4)) {
+      await writeCodexRollout(codex, '06', `plain-${n}`, codexRollout(`plain-${n}`, `/tmp/cx${n}`, [1, 2].map(t => ({ n: t, at: `2026-05-06T0${n}:${t}0:00.000Z` }))))
+    }
+
+    const parallel = await bothWays({ CODEBURN_VERBOSE: '1' })
+
+    expect(parallel.stderr).toContain('codeburn: codex parse workers=3')
+    // Pin that the codex-cache comparison in bothWays was not vacuous.
+    expect(await codexResults(join(home, 'cache-parallel'))).toContain('rollout-')
+    // Pin that the codex discard path actually ran rather than passing by luck.
+    const codexDiscards = [...parallel.stderr.matchAll(/codex parse workers done, (\d+)\/\d+ results/g)]
+      .reduce((n, m) => n + Number(m[1]), 0)
+    expect(codexDiscards).toBeGreaterThan(0)
+  })
+
+  // Workers only ever run WHOLE-file decodes. A rollout that grew by a few KB is
+  // resumed from its last task boundary in-process: a thread hop would cost more
+  // than it saves, and the resume state lives in the parent's codex cache.
+  it('never hands a resumable rollout to a worker', async () => {
+    const codex = join(home, '.codex')
+    const path = await writeCodexRollout(codex, '04', 'grow', codexRollout('grow', '/tmp/cxg', [1, 2, 3].map(n => ({ n, at: `2026-05-04T09:${n}0:00.000Z` }))))
+    const cache = join(home, 'cache-inc')
+    const args = ['status', '--format', 'menubar-json']
+
+    const cold = runCli(args, home, { CODEBURN_PARSE_WORKERS: '3', CODEBURN_CACHE_DIR: cache, CODEBURN_VERBOSE: '1' })
+    expect(cold.status, cold.stderr).toBe(0)
+    expect(cold.stderr).toContain('codeburn: codex parse workers=3')
+
+    await appendFile(path, codexTaskLines([{ n: 4, at: '2026-05-04T09:40:00.000Z' }]).join('\n') + '\n')
+    const warm = runCli(args, home, { CODEBURN_PARSE_WORKERS: '3', CODEBURN_CACHE_DIR: cache, CODEBURN_VERBOSE: '1' })
+    expect(warm.status, warm.stderr).toBe(0)
+    expect(warm.stderr).toContain('codeburn: codex parse workers=0 (no full parses pending)')
+  })
 })
 
 describe('ParseWorkerPool', () => {
   let home: string
   let files: string[]
+  let codexPath: string
+  let codexSource: SessionSource
 
   beforeEach(async () => {
     clearSessionCache()
     home = await mkdtemp(join(tmpdir(), 'cb-pool-'))
     files = await writeCorpus(join(home, '.claude'), 2, 4)
+    codexPath = await writeCodexRollout(
+      join(home, '.codex'), '04', 'pool',
+      codexRollout('pool-1', '/tmp/cx', [1, 2].map(n => ({ n, at: `2026-05-04T09:${n}0:00.000Z` }))),
+    )
+    codexSource = { provider: 'codex', path: codexPath, project: 'tmp-cx' }
     process.env['CLAUDE_CONFIG_DIR'] = join(home, '.claude')
+    // Isolated so a parse in this process can never walk the developer's own
+    // ~/.codex, and so the pool the Codex path opens is covered by the leak check.
+    process.env['CODEX_HOME'] = join(home, '.codex')
     process.env['CODEBURN_CACHE_DIR'] = join(home, '.cache', 'codeburn')
   })
 
   afterEach(async () => {
     clearSessionCache()
     delete process.env['CODEBURN_PARSE_WORKERS']
+    delete process.env['CODEX_HOME']
     await rm(home, { recursive: true, force: true })
   })
 
@@ -250,7 +405,7 @@ describe('ParseWorkerPool', () => {
     const before = liveWorkers()
     const pool = new ParseWorkerPool(3)
     const results = []
-    for await (const r of parseFilesInOrder(pool, files)) results.push(r)
+    for await (const r of parseFilesInOrder<ClaudeWorkerParse>(pool, files.map(filePath => ({ kind: 'claude' as const, filePath })))) results.push(r)
     await pool.close()
 
     expect(results).toHaveLength(files.length)
@@ -267,10 +422,10 @@ describe('ParseWorkerPool', () => {
   // result the worker would have produced.
   it('reports failures instead of throwing, and the serial fallback matches', async () => {
     const pool = new ParseWorkerPool(1)
-    const fromWorker = await pool.submit(files[0]!)
+    const fromWorker = await pool.submit<ClaudeWorkerParse>({ kind: 'claude', filePath: files[0]! })
     await pool.close()
 
-    const afterClose = await pool.submit(files[1]!)
+    const afterClose = await pool.submit({ kind: 'claude', filePath: files[1]! })
     expect(afterClose.ok).toBe(false)
 
     const serial = await parseClaudeFileFull(files[0]!, new Set<string>())
@@ -281,6 +436,31 @@ describe('ParseWorkerPool', () => {
     expect(worker).toEqual(JSON.parse(JSON.stringify(serial)))
   })
 
+
+  // Same contract for a Codex rollout: the off-thread decode is the serial
+  // decode, including the cache entry the parent has to install, and a worker
+  // that cannot answer hands the file back for an in-process parse.
+  it('decodes a codex rollout off-thread exactly as the serial path does', async () => {
+    const before = liveWorkers()
+    const pool = new ParseWorkerPool(1)
+    const fromWorker = await pool.submit<CodexFullParse & { keys: string[] }>({ kind: 'codex', source: codexSource })
+    await pool.close()
+    expect(liveWorkers()).toBe(before)
+
+    const afterClose = await pool.submit({ kind: 'codex', source: codexSource })
+    expect(afterClose.ok).toBe(false)
+
+    const seen = new Set<string>()
+    const serial = await parseCodexFileFull(codexSource, seen)
+    if (!fromWorker.ok || !fromWorker.parsed) throw new Error('expected a parsed result')
+    const { keys, ...worker } = fromWorker.parsed
+    expect(keys.length).toBeGreaterThan(0)
+    expect(new Set(keys)).toEqual(seen)
+    expect(worker).toEqual(JSON.parse(JSON.stringify(serial)))
+    // The decode itself must never have touched the codex cache file.
+    expect(await codexResults(join(home, '.cache', 'codeburn'))).toBeNull()
+  })
+
   // The resident `serve` child parses over and over in one process; a thread
   // that outlives its parse would accumulate across requests.
   it('leaves no live worker behind after back-to-back parses', async () => {