diff --git a/tests/parse-workers.test.ts b/tests/parse-workers.test.ts index 7abb0e63..fd44c98d 100644 --- a/tests/parse-workers.test.ts +++ b/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 { + 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 { + 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 /// `.` instead of by file name. async function shardBodies(cacheDir: string): Promise> { @@ -140,12 +230,19 @@ async function shardBodies(cacheDir: string): Promise> { 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 { + return readFile(join(cacheDir, 'codex-results.json'), 'utf-8').catch(() => null) +} + function runCli(args: string[], home: string, extraEnv: Record) { 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(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({ 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()) @@ -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({ 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() + 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 () => {