From 72ed163db08189de4cb0b434de3a320b56458d02 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sun, 16 Aug 2026 18:58:35 -0700 Subject: [PATCH] perf(codex): resume an appended rollout from its last task boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex rollout files are append-only and the active ones run to hundreds of MB, but any growth re-read the file from byte 0 because the cache keyed only on mtime+size. The parser now records a restart point at every task_started boundary — the byte offset plus the state the single-pass decode carries across it — and a grown file with the same dev/ino picks up from there. The boundary sits at the task_started line itself, so the task it opens is re-decoded from the tail; the entry stores how many calls were decoded before that point so the resumed run starts from exactly those and cannot double-count the open task. An unusable or absent snapshot falls back to a full re-parse. CODEX_CACHE_VERSION is deliberately not bumped: the new fields are additive and absence-safe both ways, so a bump would discard a warm cache for nothing. --- src/codex-cache.ts | 50 ++++++- src/providers/codex.ts | 171 +++++++++++++++++++---- tests/cache-directory-switch.test.ts | 6 +- tests/providers/codex-resume.test.ts | 200 +++++++++++++++++++++++++++ 4 files changed, 390 insertions(+), 37 deletions(-) create mode 100644 tests/providers/codex-resume.test.ts diff --git a/src/codex-cache.ts b/src/codex-cache.ts index 10d6b864..c227844c 100644 --- a/src/codex-cache.ts +++ b/src/codex-cache.ts @@ -15,18 +15,36 @@ import type { ParsedProviderCall } from './providers/types.js' // v6/v7: rich-session-capture — per-call locAdded/locRemoved/editFailed from // patch_apply_end. Sessions cached under v5 lack these fields; re-parse to add. // v8: persist native MCP timing and compact invocation attribution. +// Deliberately NOT bumped for the resume fields (dev/ino + resumeOffset/ +// resumeState): they are additive and absence-safe in both directions, so a +// bump would only throw away a warm multi-hundred-MB cache to gain nothing. An +// entry without them simply re-parses in full once and gains them. const CODEX_CACHE_VERSION = 8 const CACHE_FILE = 'codex-results.json' -type FileFingerprint = { mtimeMs: number; sizeBytes: number } +type FileFingerprint = { dev: number; ino: number; mtimeMs: number; sizeBytes: number } type FileEntry = { + // Absent on entries written before the resume support landed. + dev?: number + ino?: number mtimeMs: number sizeBytes: number project: string calls: ParsedProviderCall[] + /** Byte offset of a complete-line boundary the parser can restart from. */ + resumeOffset?: number + /** Opaque parser state captured at `resumeOffset` (shape owned by the Codex parser). */ + resumeState?: unknown + /** How many of `calls` were decoded before `resumeOffset`. */ + resumeCallCount?: number } +/** An exact fingerprint match, or an append the parser can resume into. */ +export type CodexCacheHit = + | { kind: 'exact'; calls: ParsedProviderCall[] } + | { kind: 'resume'; calls: ParsedProviderCall[]; offset: number; state: unknown; callCount: number } + type ResultCache = { version: number files: Record @@ -88,12 +106,28 @@ function getEntry(cache: ResultCache, filePath: string, fp: FileFingerprint): Fi export async function readCachedCodexResults( filePath: string, -): Promise { +): Promise { try { const s = await stat(filePath) const cache = await loadCache(currentCacheDir()) - const entry = getEntry(cache, filePath, { mtimeMs: s.mtimeMs, sizeBytes: s.size }) - return entry?.calls ?? null + const fp = { dev: s.dev, ino: s.ino, mtimeMs: s.mtimeMs, sizeBytes: s.size } + const entry = getEntry(cache, filePath, fp) + if (entry) return { kind: 'exact', calls: entry.calls } + // Rollouts are append-only: the same inode, grown past a boundary we + // recorded, can be picked up from that boundary instead of re-read whole. + const stale = cache.files[filePath] + if ( + stale + && stale.dev === fp.dev + && stale.ino === fp.ino + && stale.resumeOffset !== undefined + && stale.resumeState !== undefined + && stale.resumeCallCount !== undefined + && fp.sizeBytes > stale.sizeBytes + && stale.resumeOffset <= fp.sizeBytes + ) { + return { kind: 'resume', calls: stale.calls, offset: stale.resumeOffset, state: stale.resumeState, callCount: stale.resumeCallCount } + } } catch {} return null } @@ -104,7 +138,7 @@ export async function getCachedCodexProject( try { const s = await stat(filePath) const cache = await loadCache(currentCacheDir()) - const entry = getEntry(cache, filePath, { mtimeMs: s.mtimeMs, sizeBytes: s.size }) + const entry = getEntry(cache, filePath, { dev: s.dev, ino: s.ino, mtimeMs: s.mtimeMs, sizeBytes: s.size }) return entry?.project ?? null } catch {} return null @@ -115,7 +149,7 @@ export async function fingerprintFile( ): Promise { try { const s = await stat(filePath) - return { mtimeMs: s.mtimeMs, sizeBytes: s.size } + return { dev: s.dev, ino: s.ino, mtimeMs: s.mtimeMs, sizeBytes: s.size } } catch { return null } @@ -126,14 +160,18 @@ export async function writeCachedCodexResults( project: string, calls: ParsedProviderCall[], fingerprint: FileFingerprint, + resume?: { offset: number; state: unknown; callCount: number }, ): Promise { try { const cache = await loadCache(currentCacheDir()) cache.files[filePath] = { + dev: fingerprint.dev, + ino: fingerprint.ino, mtimeMs: fingerprint.mtimeMs, sizeBytes: fingerprint.sizeBytes, project, calls, + ...(resume ? { resumeOffset: resume.offset, resumeState: resume.state, resumeCallCount: resume.callCount } : {}), } } catch {} } diff --git a/src/providers/codex.ts b/src/providers/codex.ts index 7191c6fe..41d2a4df 100644 --- a/src/providers/codex.ts +++ b/src/providers/codex.ts @@ -569,52 +569,119 @@ function resolveModel(info: CodexEntry['payload'], sessionModel?: string): strin return firstModelString(info?.model, info?.info?.model, info?.info?.model_name, sessionModel) ?? 'gpt-5' } +// Everything the single-pass decode carries across a `task_started` boundary. +// A rollout is append-only, so recording this at such a boundary (where the +// previous task has been flushed to `results` and the per-task accumulators are +// empty) lets a later run restart there and produce byte-identical output +// instead of re-reading the whole file. Every field the loop below reads from a +// PRIOR line must appear here, or a resumed decode silently diverges. +type CodexResumeState = { + sessionModel?: string + sessionId: string + sessionCwd?: string + forkedFromId: string + forkCutoff: string + prevCumulativeTotal: number | null + prevInput: number + prevCached: number + prevOutput: number + prevReasoning: number + pendingTools: string[] + pendingToolSequence: ToolCall[][] + pendingUserMessage: string + pendingOutputChars: number + pendingLocAdded: number + pendingLocRemoved: number + pendingEditFailed: number + estCounter: number + turnCounter: number + currentTurnId: string + taskStartedAt?: number +} + +// The state comes back off our own JSON cache; a truncated or hand-edited file +// must fall back to a full re-parse rather than decode against nonsense. +function isResumeState(value: unknown): value is CodexResumeState { + if (!value || typeof value !== 'object') return false + const v = value as Record + return typeof v['sessionId'] === 'string' + && typeof v['forkedFromId'] === 'string' + && typeof v['forkCutoff'] === 'string' + && (v['prevCumulativeTotal'] === null || typeof v['prevCumulativeTotal'] === 'number') + && typeof v['prevInput'] === 'number' + && typeof v['prevCached'] === 'number' + && typeof v['prevOutput'] === 'number' + && typeof v['prevReasoning'] === 'number' + && Array.isArray(v['pendingTools']) + && Array.isArray(v['pendingToolSequence']) + && typeof v['pendingUserMessage'] === 'string' + && typeof v['pendingOutputChars'] === 'number' + && typeof v['pendingLocAdded'] === 'number' + && typeof v['pendingLocRemoved'] === 'number' + && typeof v['pendingEditFailed'] === 'number' + && typeof v['estCounter'] === 'number' + && typeof v['turnCounter'] === 'number' + && typeof v['currentTurnId'] === 'string' +} + function createParser(source: SessionSource, seenKeys: Set): SessionParser { return { async *parse(): AsyncGenerator { - const cached = await readCachedCodexResults(source.path) - if (cached) { - for (const call of cached) { + const hit = await readCachedCodexResults(source.path) + if (hit?.kind === 'exact') { + for (const call of hit.calls) { if (seenKeys.has(call.deduplicationKey)) continue seenKeys.add(call.deduplicationKey) yield call } return } + const resume = hit && isResumeState(hit.state) + ? { offset: hit.offset, state: hit.state, calls: hit.calls.slice(0, hit.callCount) } + : null const fp = await fingerprintFile(source.path) if (!fp) return - let sessionModel: string | undefined - let sessionId = '' - let sessionCwd: string | undefined - let forkedFromId = '' - let forkCutoff = '' + let sessionModel: string | undefined = resume?.state.sessionModel + let sessionId = resume?.state.sessionId ?? '' + let sessionCwd: string | undefined = resume?.state.sessionCwd + let forkedFromId = resume?.state.forkedFromId ?? '' + let forkCutoff = resume?.state.forkCutoff ?? '' // Null sentinel rather than `0` so the FIRST event is never confused // with a duplicate. A session that only emits last_token_usage (no // total_token_usage) reports cumulativeTotal=0 on every event; with a // 0-initialized prev, the first event would have matched and been // dropped. Once we've observed any event, we record its cumulative // total and dedup on equality regardless of whether it is zero. - let prevCumulativeTotal: number | null = null - let prevInput = 0 - let prevCached = 0 - let prevOutput = 0 - let prevReasoning = 0 - let pendingTools: string[] = [] - let pendingToolSequence: ToolCall[][] = [] - let pendingUserMessage = '' - let pendingOutputChars = 0 + let prevCumulativeTotal: number | null = resume?.state.prevCumulativeTotal ?? null + let prevInput = resume?.state.prevInput ?? 0 + let prevCached = resume?.state.prevCached ?? 0 + let prevOutput = resume?.state.prevOutput ?? 0 + let prevReasoning = resume?.state.prevReasoning ?? 0 + let pendingTools: string[] = resume ? [...resume.state.pendingTools] : [] + let pendingToolSequence: ToolCall[][] = resume ? [...resume.state.pendingToolSequence] : [] + let pendingUserMessage = resume?.state.pendingUserMessage ?? '' + let pendingOutputChars = resume?.state.pendingOutputChars ?? 0 // Rich-session-capture: edit LOC deltas and failed-patch count accumulated // across a turn's patch_apply_end events, flushed onto the turn's call. - let pendingLocAdded = 0 - let pendingLocRemoved = 0 - let pendingEditFailed = 0 - let estCounter = 0 - let turnCounter = 0 - let currentTurnId = `${sessionId}:t0` + let pendingLocAdded = resume?.state.pendingLocAdded ?? 0 + let pendingLocRemoved = resume?.state.pendingLocRemoved ?? 0 + let pendingEditFailed = resume?.state.pendingEditFailed ?? 0 + let estCounter = resume?.state.estCounter ?? 0 + let turnCounter = resume?.state.turnCounter ?? 0 + let currentTurnId = resume?.state.currentTurnId ?? `${sessionId}:t0` let sawAnyLine = false const results: ParsedProviderCall[] = [] + // Calls already decoded before the resume boundary. They pass through the + // same cross-provider dedup a full decode would have applied to them. + if (resume) { + for (const call of resume.calls) { + if (seenKeys.has(call.deduplicationKey)) continue + seenKeys.add(call.deduplicationKey) + results.push(call) + } + } // Calls decoded since the last task_started, held back so task_complete can // stamp active/toolWait timing before they are appended to results. Emitting // a task only once its timing is known keeps single-pass and split/resume @@ -623,15 +690,25 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars let pendingTaskCalls: ParsedProviderCall[] = [] let taskGeneratedTokens = 0 let taskToolIntervals: Array<[number, number]> = [] - let taskStartedAt: number | undefined + let taskStartedAt: number | undefined = resume?.state.taskStartedAt const openToolStarts = new Map() + // Resume point for the NEXT run, refreshed at every task boundary. + const tracker = { lastCompleteLineOffset: resume?.offset ?? 0 } + let resumeOffset = resume?.offset ?? 0 + let resumeState: CodexResumeState | null = resume?.state ?? null + let resumeCallCount = results.length + // Stream the session file line by line. Heavy Codex sessions can exceed // 250 MB on disk; reading the entire file into a string would either hit // the readSessionFile cap or push V8 toward its 512 MB string limit // after split('\n'). readSessionLines streams raw buffers and hands // huge lines to the compact parser without full string conversion. - for await (const rawLine of readSessionLines(source.path, undefined, { largeLineAsBuffer: true })) { + for await (const rawLine of readSessionLines(source.path, undefined, { + largeLineAsBuffer: true, + byteOffsetTracker: tracker, + ...(resume ? { startByteOffset: resume.offset } : {}), + })) { sawAnyLine = true const entry = parseCodexLine(rawLine) if (!entry) continue @@ -684,6 +761,33 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars const startedAt = entry.timestamp ? Date.parse(entry.timestamp) : NaN taskStartedAt = Number.isFinite(startedAt) ? startedAt : undefined openToolStarts.clear() + // Everything decoded so far is now in `results` and the per-task + // accumulators are empty: a clean restart point for an appended tail. + resumeOffset = tracker.lastCompleteLineOffset + resumeCallCount = results.length + resumeState = { + ...(sessionModel !== undefined ? { sessionModel } : {}), + sessionId, + ...(sessionCwd !== undefined ? { sessionCwd } : {}), + forkedFromId, + forkCutoff, + prevCumulativeTotal, + prevInput, + prevCached, + prevOutput, + prevReasoning, + pendingTools: [...pendingTools], + pendingToolSequence: [...pendingToolSequence], + pendingUserMessage, + pendingOutputChars, + pendingLocAdded, + pendingLocRemoved, + pendingEditFailed, + estCounter, + turnCounter, + currentTurnId, + ...(taskStartedAt !== undefined ? { taskStartedAt } : {}), + } continue } @@ -997,13 +1101,24 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars // If the stream yielded nothing the file was unreadable, oversized, or // empty. Skip cache write so a transient failure can't pin an empty - // result set against a fingerprint that would otherwise be re-parsed. - if (!sawAnyLine) return + // result set against a fingerprint that would otherwise be re-parsed. On a + // resume the earlier calls are still valid output, so serve them - but + // still leave the cache entry alone. + if (!sawAnyLine) { + if (resume) for (const call of results) yield call + return + } // Flush the final task, which has no following task_started to trigger it. results.push(...pendingTaskCalls) - await writeCachedCodexResults(source.path, source.project, results, fp) + await writeCachedCodexResults( + source.path, + source.project, + results, + fp, + resumeState ? { offset: resumeOffset, state: resumeState, callCount: resumeCallCount } : undefined, + ) for (const call of results) { yield call diff --git a/tests/cache-directory-switch.test.ts b/tests/cache-directory-switch.test.ts index 9dfc9703..175cd828 100644 --- a/tests/cache-directory-switch.test.ts +++ b/tests/cache-directory-switch.test.ts @@ -111,7 +111,7 @@ describe('call-time CODEBURN_CACHE_DIR isolation', () => { expect(diskB.files[sourcePath].calls.map((entry: ParsedProviderCall) => entry.model)).toEqual(['from-b']) process.env['CODEBURN_CACHE_DIR'] = cacheA - expect((await readCachedCodexResults(sourcePath))?.map(entry => entry.model)).toEqual(['from-a']) + expect((await readCachedCodexResults(sourcePath))?.calls.map(entry => entry.model)).toEqual(['from-a']) }) it('does not flush dirty Codex state from A into B', async () => { @@ -265,13 +265,13 @@ describe('call-time CODEBURN_CACHE_DIR isolation', () => { codexDisk.files[codexSource].calls[0].model = 'after' await writeFile(join(cacheDir, 'codex-results.json'), JSON.stringify(codexDisk)) - expect((await readCachedCodexResults(codexSource))?.map(entry => entry.model)).toEqual(['before']) + expect((await readCachedCodexResults(codexSource))?.calls.map(entry => entry.model)).toEqual(['before']) expect(await readAntigravityModel(antigravitySource)).toBe('before') clearCodexMemCaches() clearAntigravityCacheStates() - expect((await readCachedCodexResults(codexSource))?.map(entry => entry.model)).toEqual(['after']) + expect((await readCachedCodexResults(codexSource))?.calls.map(entry => entry.model)).toEqual(['after']) expect(await readAntigravityModel(antigravitySource)).toBe('after') }) }) diff --git a/tests/providers/codex-resume.test.ts b/tests/providers/codex-resume.test.ts new file mode 100644 index 00000000..35f0d45a --- /dev/null +++ b/tests/providers/codex-resume.test.ts @@ -0,0 +1,200 @@ +// Codex rollouts are append-only and the active ones are huge, so a run that +// re-read a grown session from byte 0 paid for the whole file to pick up a few +// KB. The parser now restarts from the last task boundary it recorded. What has +// to hold: the resumed decode is byte-identical to a full re-parse, and it +// really does start at an offset rather than quietly re-reading everything. +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import { appendFile, mkdtemp, mkdir, rm, writeFile } from 'fs/promises' +import { join } from 'path' +import { tmpdir } from 'os' + +const readLineCalls: Array<{ filePath: string; startByteOffset?: number }> = [] +vi.mock('../../src/fs-utils.js', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + readSessionLines: (filePath: string, skip?: unknown, options?: { startByteOffset?: number }) => { + readLineCalls.push({ filePath, startByteOffset: options?.startByteOffset }) + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (actual.readSessionLines as any)(filePath, skip, options) + }, + } +}) + +import { flushCodexCache, withCodexCacheDirectory } from '../../src/codex-cache.js' +import { createCodexProvider } from '../../src/providers/codex.js' +import type { ParsedProviderCall } from '../../src/providers/types.js' + +let tmpDir: string +let sessionPath: string + +beforeEach(async () => { + tmpDir = await mkdtemp(join(tmpdir(), 'codex-resume-')) + readLineCalls.length = 0 +}) + +afterEach(async () => { + await rm(tmpDir, { recursive: true, force: true }) +}) + +function meta(): string { + return JSON.stringify({ + type: 'session_meta', + timestamp: '2026-04-14T10:00:00Z', + payload: { cwd: '/Users/test/proj', originator: 'codex-cli', session_id: 'sess-1', model: 'gpt-5.3-codex' }, + }) +} + +// One complete task: user turn, tools, an edit, an MCP call, usage, completion. +function task(n: number, cumulative: { input: number; cached: number; output: number; reasoning: number }): string[] { + const at = (s: number) => `2026-04-14T10:${String(n).padStart(2, '0')}:${String(s).padStart(2, '0')}Z` + return [ + JSON.stringify({ type: 'event_msg', timestamp: at(0), payload: { type: 'task_started' } }), + JSON.stringify({ + type: 'response_item', timestamp: at(1), + payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `task ${n}` }] }, + }), + JSON.stringify({ + type: 'response_item', timestamp: at(2), + payload: { type: 'function_call', name: 'shell', call_id: `c${n}`, arguments: JSON.stringify({ command: `ls ${n}` }) }, + }), + JSON.stringify({ + type: 'response_item', timestamp: at(3), + payload: { type: 'function_call_output', call_id: `c${n}` }, + }), + JSON.stringify({ + type: 'event_msg', timestamp: at(4), + payload: { + type: 'patch_apply_end', success: n % 2 === 0, + changes: { [`/Users/test/proj/f${n}.ts`]: { unified_diff: '@@ -1 +1,2 @@\n-old\n+new\n+extra\n' } }, + }, + }), + JSON.stringify({ + type: 'event_msg', timestamp: at(5), + payload: { type: 'mcp_tool_call_end', call_id: `m${n}`, invocation: { server: 'github', tool: 'list' }, duration_ms: 120, result: { Ok: {} } }, + }), + JSON.stringify({ + type: 'response_item', timestamp: at(6), + payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'x'.repeat(40) }] }, + }), + JSON.stringify({ + type: 'event_msg', timestamp: at(7), + 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: cumulative.input, cached_input_tokens: cumulative.cached, + output_tokens: cumulative.output, reasoning_output_tokens: cumulative.reasoning, + total_tokens: cumulative.input + cumulative.output + cumulative.reasoning, + }, + }, + }, + }), + JSON.stringify({ type: 'event_msg', timestamp: at(8), payload: { type: 'task_complete', duration_ms: 5000 } }), + ] +} + +function tasks(from: number, to: number): string[] { + const lines: string[] = [] + for (let n = from; n <= to; n++) { + lines.push(...task(n, { input: 100 * n, cached: 20 * n, output: 50 * n, reasoning: 10 * n })) + } + return lines +} + +async function writeRollout(lines: string[]): Promise { + const dir = join(tmpDir, 'sessions', '2026', '04', '14') + await mkdir(dir, { recursive: true }) + const path = join(dir, 'rollout-sess-1.jsonl') + await writeFile(path, lines.join('\n') + '\n') + return path +} + +async function parse(cacheDir: string): Promise { + return withCodexCacheDirectory(cacheDir, async () => { + const provider = createCodexProvider(tmpDir) + const sources = await provider.discoverSessions() + const seenKeys = new Set() + const calls: ParsedProviderCall[] = [] + for (const source of sources) { + for await (const call of provider.createSessionParser!(source, seenKeys).parse()) calls.push(call) + } + await flushCodexCache() + return calls + }) +} + +describe('codex incremental resume', () => { + it('resumes at a task boundary and matches a full re-parse exactly', async () => { + const warmCache = join(tmpDir, 'cache-warm') + const coldCache = join(tmpDir, 'cache-cold') + + sessionPath = await writeRollout([meta(), ...tasks(1, 3)]) + const first = await parse(warmCache) + expect(first.length).toBe(3) + + await appendFile(sessionPath, tasks(4, 6).join('\n') + '\n') + + readLineCalls.length = 0 + const resumed = await parse(warmCache) + const resumeReads = readLineCalls.filter(c => c.filePath === sessionPath) + // The parse re-entered the file at a boundary rather than at byte 0. + expect(resumeReads.some(c => (c.startByteOffset ?? 0) > 0)).toBe(true) + expect(resumeReads.every(c => (c.startByteOffset ?? 0) > 0)).toBe(true) + + // Byte-for-byte agreement with a decode that never saw a cache. + const full = await parse(coldCache) + expect(resumed.length).toBe(6) + expect(JSON.stringify(resumed)).toBe(JSON.stringify(full)) + }) + + it('stays exact across successive appends, resuming from a resumed state', async () => { + const warmCache = join(tmpDir, 'cache-warm') + + sessionPath = await writeRollout([meta(), ...tasks(1, 2)]) + await parse(warmCache) + await appendFile(sessionPath, tasks(3, 4).join('\n') + '\n') + await parse(warmCache) + // A tail with no task boundary at all: the next run restarts from the same + // boundary and re-decodes the open task. + await appendFile(sessionPath, tasks(5, 5).slice(1).join('\n') + '\n') + const resumed = await parse(warmCache) + + const full = await parse(join(tmpDir, 'cache-cold')) + expect(JSON.stringify(resumed)).toBe(JSON.stringify(full)) + }) + + it('serves an unchanged file from the cache without reading it', async () => { + const cacheDir = join(tmpDir, 'cache') + sessionPath = await writeRollout([meta(), ...tasks(1, 2)]) + const first = await parse(cacheDir) + + readLineCalls.length = 0 + const second = await parse(cacheDir) + expect(readLineCalls.filter(c => c.filePath === sessionPath)).toHaveLength(0) + expect(JSON.stringify(second)).toBe(JSON.stringify(first)) + }) + + it('falls back to a full re-parse when the stored resume state is unusable', async () => { + const cacheDir = join(tmpDir, 'cache') + sessionPath = await writeRollout([meta(), ...tasks(1, 2)]) + await parse(cacheDir) + + const cachePath = join(cacheDir, 'codex-results.json') + const { readFile } = await import('fs/promises') + const raw = JSON.parse(await readFile(cachePath, 'utf-8')) + raw.files[sessionPath].resumeState = { garbage: true } + await writeFile(cachePath, JSON.stringify(raw)) + const { clearCodexMemCaches } = await import('../../src/codex-cache.js') + clearCodexMemCaches() + + await appendFile(sessionPath, tasks(3, 3).join('\n') + '\n') + readLineCalls.length = 0 + const resumed = await parse(cacheDir) + expect(readLineCalls.filter(c => c.filePath === sessionPath).every(c => (c.startByteOffset ?? 0) === 0)).toBe(true) + + const full = await parse(join(tmpDir, 'cache-cold')) + expect(JSON.stringify(resumed)).toBe(JSON.stringify(full)) + }) +})