From 0bfdfa372af35ee9ffef472697a699a6d3c3c90c Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sun, 16 Aug 2026 18:51:16 -0700 Subject: [PATCH 1/4] perf(cache): shard the session cache per provider A warm launch rewrote the entire session cache whenever any provider appended a few KB: on a 6 GB corpus that is a 155 MB stringify + fsync every run. The on-disk cache is now a version-suffixed directory holding one shard per provider plus a small envelope, and a save rewrites only the providers marked dirty. - Dirtiness is tracked per provider (markCacheDirty) instead of one global flag, so an appended Claude session no longer republishes Codex, Copilot and the rest. - Shards carry a nonce in their filename and the envelope is renamed last, so a save is published at a single point: readers never see a half-updated set, and a writer that loses the refresh ownership fence leaves the canonical shards untouched. - A shard that fails validation is treated as an absent provider rather than rejecting the whole cache, so one malformed turn costs one provider's re-parse instead of every provider's history. - v7 migrates losslessly: the blob is re-laid-out into shards and removed only once that save publishes. Nothing re-parses. - Cold-parse progress saves now trigger every N files parsed rather than every 5s, so a slow cold parse no longer rewrites the growing cache on a wall clock. --- src/parser.ts | 47 +-- src/session-cache.ts | 302 +++++++++++++----- tests/cache-persistence-coldstart.test.ts | 5 +- tests/cache-refresh-lock.test.ts | 4 +- tests/codex-cache-invalidation.test.ts | 7 +- tests/fixtures/cache-refresh-worker.ts | 4 +- tests/fixtures/session-cache-io.ts | 35 ++ tests/hydration-interrupt-convergence.test.ts | 8 +- tests/kiro-cache-invalidation.test.ts | 4 +- tests/kiro-projectpath.test.ts | 4 +- tests/parser-antigravity-timestamp.test.ts | 6 +- tests/parser-cache-refresh-timeout.test.ts | 6 +- tests/parser-gemini-cache.test.ts | 9 +- tests/parser-hydration-lock.test.ts | 16 +- tests/parser-incremental-append.test.ts | 13 +- tests/parser.test.ts | 9 +- tests/providers/codewhale.test.ts | 8 +- tests/session-cache-rich-capture.test.ts | 4 +- tests/session-cache-shards.test.ts | 208 ++++++++++++ tests/session-cache.test.ts | 39 +-- 20 files changed, 563 insertions(+), 175 deletions(-) create mode 100644 tests/fixtures/session-cache-io.ts create mode 100644 tests/session-cache-shards.test.ts diff --git a/src/parser.ts b/src/parser.ts index 6aac4eec..af115106 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -23,7 +23,9 @@ import { DURABLE_PROVIDER_NAMES, fingerprintFile, isCacheComplete, + isCacheDirty, loadCache, + markCacheDirty, reconcileFile, saveCache, } from './session-cache.js' @@ -2115,7 +2117,7 @@ async function scanProjectDirs( ...(Object.keys(mergedSpawnLinks).length > 0 ? { agentSpawnLinks: mergedSpawnLinks } : {}), ...(mergedAmbiguousIds.length > 0 ? { ambiguousSpawnAgentIds: mergedAmbiguousIds } : {}), } - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, 'claude') filesDone++ await parseProgress.tick(filesDone) if (filesDone % 50 === 0 || filesDone === progressTotal) { @@ -2152,7 +2154,7 @@ async function scanProjectDirs( ...(Object.keys(sessionMeta.agentSpawnLinks).length > 0 ? { agentSpawnLinks: sessionMeta.agentSpawnLinks } : {}), ...(sessionMeta.ambiguousSpawnAgentIds.length > 0 ? { ambiguousSpawnAgentIds: sessionMeta.ambiguousSpawnAgentIds } : {}), } - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, 'claude') } catch (err) { // A single malformed Claude session file must not abort the whole run — that // would empty the daily-cache backfill and wipe the trend/history (issue #441, @@ -2160,7 +2162,7 @@ async function scanProjectDirs( // by the current fingerprint so it isn't re-read and re-thrown every run; it // re-parses only if the file changes. section.files[filePath] = { fingerprint: info.fp, mcpInventory: [], turns: [], failed: true } - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, 'claude') warnProviderParseFailure('claude', filePath, err) } filesDone++ @@ -2181,7 +2183,7 @@ async function scanProjectDirs( // but they carry attributable PR spend (surfaced above as a legacy split). if (section.files[cachedPath]?.prLinks?.length) continue delete section.files[cachedPath] - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, 'claude') } } @@ -2686,6 +2688,7 @@ function getOrCreateProviderSection(cache: SessionCache, provider: string): Prov } } cache.providers[provider] = section + markCacheDirty(cache, provider) return section } @@ -2803,10 +2806,10 @@ export function emitScanProgress(event: ScanProgressEvent): void { try { process.stderr.write(`${PROGRESS_LINE_PREFIX}${JSON.stringify(event)}\n`) } catch { /* stderr closed */ } } -// Minimum spacing between partial-progress saves during a cold parse. Low enough +// Files parsed between partial-progress saves during a cold parse. Low enough // that an interrupted long run loses little work, high enough that repeated -// full-cache writes never dominate a fast warm run. -const PROGRESS_SAVE_THROTTLE_MS = 5000 +// cache writes never dominate the parse. +const PROGRESS_SAVE_FILE_INTERVAL = 2000 export function createScanProgress(label: string, total: number) { const show = !interactiveScanUI && total > 20 && process.stderr.isTTY === true @@ -3022,7 +3025,7 @@ async function parseProviderSources( } } didParse = true - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, providerName) } catch (err) { if (isSqliteBusyError(err)) { warnProviderReadFailureOnce(providerName, err) @@ -3035,7 +3038,7 @@ async function parseProviderSources( // on every refresh; it re-parses only if it changes. Empty turns => no // usage contributed. section.files[source.path] = { fingerprint: fp, mcpInventory: [], turns: [], failed: true } - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, providerName) warnProviderParseFailure(providerName, source.path, err) continue } @@ -3052,14 +3055,14 @@ async function parseProviderSources( // parseAllSessions can fast-check without a getProvider() round-trip. if (!readOnly && provider.durableSources && !section.durable) { section.durable = true - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, providerName) } if (!readOnly && sources.length > 0 && !provider.durableSources) { for (const cachedPath of Object.keys(section.files)) { if (!allDiscoveredFiles.has(cachedPath)) { delete section.files[cachedPath] - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, providerName) } } } @@ -3076,7 +3079,7 @@ async function parseProviderSources( .reduce((max, ts) => Math.max(max, ts), 0) if (newestTs > 0 && newestTs < cutoffMs) { delete section.files[cachedPath] - ;(diskCache as { _dirty?: boolean })._dirty = true + markCacheDirty(diskCache, providerName) } } } @@ -3860,16 +3863,20 @@ async function runParse( providerGroups.set(source.provider, existing) } - // Cold-run robustness: persist partial progress during a long parse (throttled) - // so a run interrupted before the single end-of-parse save still leaves a warm - // cache behind. saveCache is atomic (temp + rename) and clears `_dirty`, so this + // Cold-run robustness: persist partial progress during a long parse so a run + // interrupted before the single end-of-parse save still leaves a warm cache + // behind. Triggered by files parsed rather than elapsed time: the cost of a + // save scales with the corpus, not the clock, so a wall-clock throttle made a + // slow cold parse rewrite the whole (growing) cache every few seconds. At this + // interval a ~18k-file cold parse saves under a dozen times. saveCache is + // atomic (temp + rename) and writes only the dirty provider shards, so this // never races the final save below. - let lastSaveAt = Date.now() + let filesSinceSave = 0 const saveProgress = async (): Promise => { if (!isCold || readOnly) return - if (!(diskCache as { _dirty?: boolean })._dirty) return - if (Date.now() - lastSaveAt < PROGRESS_SAVE_THROTTLE_MS) return - lastSaveAt = Date.now() + if (!isCacheDirty(diskCache)) return + if (++filesSinceSave < PROGRESS_SAVE_FILE_INTERVAL) return + filesSinceSave = 0 try { await saveCache(diskCache) } catch { /* best-effort partial save */ } } @@ -3952,7 +3959,7 @@ async function runParse( // partial saves keep `complete: false` and the next launch resumes cold. const wasComplete = isCacheComplete(diskCache) if (!readOnly && !wasComplete) diskCache.complete = true - if (!readOnly && ((diskCache as { _dirty?: boolean })._dirty || !wasComplete)) { + if (!readOnly && (isCacheDirty(diskCache) || !wasComplete)) { try { const published = await saveCache(diskCache, refreshLock?.verifyStillOwner) if (!published) throw new RefreshFenceLostError() diff --git a/src/session-cache.ts b/src/session-cache.ts index 75ce0a86..9a0010a1 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -157,14 +157,19 @@ export type SessionCache = { // INVARIANT: a version bump must extend `PRIOR_CACHE_VERSIONS` (the adoption path // below) to EVERY prior version that can still exist on disk, or expired-PR // history from the immediately preceding build silently vanishes. -export const CACHE_VERSION = 7 +// v8: on-disk layout only - the single blob became a directory of per-provider +// shards plus a small envelope, so a launch that only touched one provider +// rewrites just that provider's file. The turn shape is unchanged, so a v7 file +// migrates losslessly (migrateSingleFileCache) rather than re-parsing. +export const CACHE_VERSION = 8 -// The cache filename is version-suffixed so different binaries (e.g. an old -// launchd menubar on a prior release and a newer desktop app) each own a -// distinct file and can never clobber each other's incompatible schema. Bumping -// CACHE_VERSION automatically mints a fresh filename, superseding the migration -// dance the legacy unversioned file used to need. -const CACHE_FILE = `session-cache.v${CACHE_VERSION}.json` +// The cache directory is version-suffixed for the same reason the file used to +// be: different binaries (an old launchd menubar, a newer desktop app) each own +// a distinct layout and can never clobber each other's incompatible schema. +const CACHE_DIR_NAME = `session-cache.v${CACHE_VERSION}` +// Written LAST on every save: it names the shard file of every provider, so the +// rename that publishes it is the single point at which a save becomes visible. +const ENVELOPE_FILE = 'envelope.json' // The pre-versioning filename. Never written or deleted anymore — old binaries // still own it. On first load we adopt-copy it once (see loadCache) when the // versioned file is absent and the legacy file's version matches ours. @@ -284,17 +289,48 @@ export const PROVIDER_PARSE_VERSIONS: Record = { antigravity: 'worktree-project-grouping-v5', } -function getCachePath(): string { - return join(getCodeburnCacheDir(), CACHE_FILE) -} - function getLegacyCachePath(): string { return join(getCodeburnCacheDir(), LEGACY_CACHE_FILE) } -/** Absolute path of the active (version-suffixed) session cache file. */ -export function sessionCachePath(): string { - return getCachePath() +/** Absolute path of the active (version-suffixed) session cache directory. */ +export function sessionCacheDir(): string { + return join(getCodeburnCacheDir(), CACHE_DIR_NAME) +} + +type CacheEnvelope = { + version: number + complete?: boolean + nonce: string + shards: Record +} + +// Save bookkeeping, held beside the cache rather than on it so it never lands in +// a shard's JSON or in a caller's deep-equality. `shards` is the +// provider -> shard filename map the last load/save published; a provider that +// is neither dirty nor already sharded is written on the next save. +type CacheState = { dirty: boolean; dirtyProviders: Set; shards: Record } +const cacheStates = new WeakMap() + +function stateOf(cache: SessionCache): CacheState { + let state = cacheStates.get(cache) + if (!state) { + state = { dirty: false, dirtyProviders: new Set(), shards: {} } + cacheStates.set(cache, state) + } + return state +} + +/** Record that `provider`'s section changed, so the next save rewrites its shard. */ +export function markCacheDirty(cache: SessionCache, provider: string): void { + const state = stateOf(cache) + state.dirty = true + state.dirtyProviders.add(provider) +} + +/** True when any provider section changed since the last save. */ +export function isCacheDirty(cache: SessionCache): boolean { + return stateOf(cache).dirty } // ── Env Fingerprint ──────────────────────────────────────────────────── @@ -447,10 +483,11 @@ function validateProviderSection(s: unknown): s is ProviderSection { return Object.values(o['files'] as Record).every(validateCachedFile) } -function validateCache(raw: unknown): raw is SessionCache { +// Full validation of a single-file (pre-v8) cache blob at `version`. +function validateCache(raw: unknown, version: number): raw is SessionCache { if (!raw || typeof raw !== 'object') return false const o = raw as Record - if (o['version'] !== CACHE_VERSION) return false + if (o['version'] !== version) return false if (!o['providers'] || typeof o['providers'] !== 'object' || Array.isArray(o['providers'])) return false return Object.values(o['providers'] as Record).every(validateProviderSection) } @@ -463,7 +500,7 @@ function validateCache(raw: unknown): raw is SessionCache { // CACHE_VERSION bump MUST extend this list to every prior version that can still // exist on disk, or that history silently vanishes. (v5 was missed on the 5->6 // bump; v6 on the 6->7 bump; both are listed here.) -const PRIOR_CACHE_VERSIONS = [6, 5] as const +const PRIOR_CACHE_VERSIONS = [7, 6, 5] as const function priorCacheFile(version: number): string { return `session-cache.v${version}.json` @@ -539,54 +576,110 @@ async function adoptNewestPriorCache(): Promise { return merged } -// In-process memo of the parsed cache, keyed by the file identity that last -// produced it. On a 100MB+ corpus the JSON.parse of the session cache is -// seconds of work per load; a resident process (codeburn serve) pays it once -// and revalidates with a stat() per request. A rewrite by ANOTHER process -// moves mtime/size and forces a reload, so cross-process freshness is -// preserved; saveCache updates the memo write-through so the object handed -// out stays the canonical one after a refresh. -let cacheMemo: { path: string; mtimeMs: number; size: number; cache: SessionCache } | null = null +// In-process memo of the parsed cache, keyed by the envelope nonce that last +// produced it. On a 100MB+ corpus the JSON.parse of the shards is seconds of +// work per load; a resident process (codeburn serve) pays it once and +// revalidates by re-reading the (tiny) envelope per request. A save by ANOTHER +// process mints a new nonce and forces a reload, so cross-process freshness is +// preserved; saveCache updates the memo write-through so the object handed out +// stays the canonical one after a refresh. +let cacheMemo: { dir: string; nonce: string; cache: SessionCache } | null = null export function clearLoadCacheMemo(): void { cacheMemo = null } -export async function loadCache(): Promise { - const path = getCachePath() +function isEnvelope(raw: unknown): raw is CacheEnvelope { + if (!raw || typeof raw !== 'object') return false + const o = raw as Record + return o['version'] === CACHE_VERSION + && typeof o['nonce'] === 'string' + && !!o['shards'] && typeof o['shards'] === 'object' && !Array.isArray(o['shards']) + && Object.values(o['shards'] as Record).every(v => typeof v === 'string') +} + +// A shard that is missing or malformed is treated as an ABSENT provider, not as +// a corrupt cache: only that provider re-parses, instead of the old all-or- +// nothing where one bad turn discarded every provider's history. +async function loadShard(path: string): Promise { try { - const info = await stat(path) - if (cacheMemo && cacheMemo.path === path && cacheMemo.mtimeMs === info.mtimeMs && cacheMemo.size === info.size) { - return cacheMemo.cache - } - const raw = await readFile(path, 'utf-8') - const parsed = JSON.parse(raw) - if (!validateCache(parsed)) return afterMissingVersionedCache() - cacheMemo = { path, mtimeMs: info.mtimeMs, size: info.size, cache: parsed } - return parsed + const parsed = JSON.parse(await readFile(path, 'utf-8')) + return validateProviderSection(parsed) ? parsed : null } catch { - return afterMissingVersionedCache() + return null } } -// The current versioned file is absent/unreadable. Prefer adopting the newest -// prior versioned file's expired-source PR orphans (v6 before v5); failing that, -// fall back to the legacy unversioned file. Either way the versioned file is -// minted on the next save. -async function afterMissingVersionedCache(): Promise { +export async function loadCache(): Promise { + const dir = sessionCacheDir() + let envelope: CacheEnvelope + try { + const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) + if (!isEnvelope(parsed)) return afterMissingShardCache() + envelope = parsed + } catch { + return afterMissingShardCache() + } + if (cacheMemo && cacheMemo.dir === dir && cacheMemo.nonce === envelope.nonce) return cacheMemo.cache + + const cache: SessionCache = { version: CACHE_VERSION, providers: {}, complete: envelope.complete === true } + const shards: Record = {} + for (const [provider, file] of Object.entries(envelope.shards)) { + const section = await loadShard(join(dir, file)) + if (!section) continue + cache.providers[provider] = section + shards[provider] = file + } + stateOf(cache).shards = shards + cacheMemo = { dir, nonce: envelope.nonce, cache } + return cache +} + +// The shard directory is absent/unreadable. Prefer the LOSSLESS re-layout of the +// v7 single-file cache (same turn shape, so nothing re-parses); failing that, +// adopt the prior versions' expired-source PR orphans, then the legacy +// unversioned file. Either way the shard directory is minted on the next save. +async function afterMissingShardCache(): Promise { + const migrated = await migrateSingleFileCache() + if (migrated) return migrated const prior = await adoptNewestPriorCache() if (prior) return prior - // validateCache requires version === CACHE_VERSION, so a different-version - // legacy file is ignored (left intact). We copy it into the versioned file once - // via saveCache; the legacy file is never modified. + // validateCache requires the version to match, so a different-version legacy + // file is ignored (left intact). We copy it into the shard layout once via + // saveCache; the legacy file is never modified. return adoptLegacyCache() } +// One-time, lossless migration of the v7 single-file cache: v8 changed the +// on-disk LAYOUT only, so every section moves across verbatim and nothing +// re-parses. Every section is marked dirty so the save below writes each shard; +// the v7 file is removed only once that save has published. +async function migrateSingleFileCache(): Promise { + const v7Path = join(getCodeburnCacheDir(), priorCacheFile(7)) + let parsed: unknown + try { + parsed = JSON.parse(await readFile(v7Path, 'utf-8')) + } catch { + return null + } + if (!validateCache(parsed, 7)) return null + const cache: SessionCache = { + version: CACHE_VERSION, + providers: parsed.providers, + complete: parsed.complete === true, + } + for (const provider of Object.keys(cache.providers)) markCacheDirty(cache, provider) + const published = await saveCache(cache).catch(() => false) + if (published) await retryCacheFileMutation(() => unlink(v7Path)) + return cache +} + async function adoptLegacyCache(): Promise { try { const raw = await readFile(getLegacyCachePath(), 'utf-8') const parsed = JSON.parse(raw) - if (!validateCache(parsed)) return emptyCache() + if (!validateCache(parsed, CACHE_VERSION)) return emptyCache() + for (const provider of Object.keys(parsed.providers)) markCacheDirty(parsed, provider) await saveCache(parsed).catch(() => {}) return parsed } catch { @@ -594,15 +687,16 @@ async function adoptLegacyCache(): Promise { } } -export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise): Promise { - const dir = getCodeburnCacheDir() - if (!existsSync(dir)) await mkdir(dir, { recursive: true }) - - const finalPath = getCachePath() - const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp` - delete (cache as { _dirty?: boolean })._dirty - const payload = JSON.stringify(cache) +// Shard filenames carry a fresh nonce on every write, so a save never overwrites +// the file the currently-published envelope points at: readers keep seeing a +// consistent set until the envelope rename publishes the new one, and a writer +// that loses the ownership fence leaves the canonical shards untouched. +function shardFileName(provider: string): string { + return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${randomBytes(8).toString('hex')}.json` +} +async function writeFileAtomic(finalPath: string, payload: string): Promise { + const tempPath = `${finalPath}.tmp` const handle = await open(tempPath, 'w', 0o600) try { await handle.writeFile(payload, { encoding: 'utf-8' }) @@ -610,40 +704,76 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr } finally { await handle.close() } - try { - // The warm refresh transaction passes an ownership fence. It must be the - // final operation before publication so a displaced writer cannot replace - // the canonical cache with its stale snapshot. - if (verifyStillOwner && !await verifyStillOwner()) { - await retryCacheFileMutation(() => unlink(tempPath)) - return false - } - let renamed = false for (let attempt = 0; attempt < 3; attempt++) { try { await rename(tempPath, finalPath) - renamed = true - break + return } catch (err) { const code = (err as NodeJS.ErrnoException).code if ((code !== 'EPERM' && code !== 'EBUSY') || attempt === 2) throw err await new Promise(resolve => { setTimeout(resolve, 10 * (attempt + 1)) }) } } - if (!renamed) throw new Error('session cache rename failed') - // Write-through: the object just published IS the freshest state; capture - // the post-rename file identity so the next loadCache in this process - // reuses it instead of re-parsing what it just wrote. - try { - const info = await stat(finalPath) - cacheMemo = { path: finalPath, mtimeMs: info.mtimeMs, size: info.size, cache } - } catch { - cacheMemo = null + } catch (err) { + await retryCacheFileMutation(() => unlink(tempPath)) + throw err + } +} + +export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Promise): Promise { + const dir = sessionCacheDir() + if (!existsSync(dir)) await mkdir(dir, { recursive: true }) + + const state = stateOf(cache) + const priorShards = state.shards + const shards: Record = {} + const written: string[] = [] + + try { + for (const [provider, section] of Object.entries(cache.providers)) { + const prior = priorShards[provider] + if (prior && !state.dirtyProviders.has(provider)) { shards[provider] = prior; continue } + const name = shardFileName(provider) + await writeFileAtomic(join(dir, name), JSON.stringify(section)) + written.push(name) + shards[provider] = name + } + + // The warm refresh transaction passes an ownership fence. It must be the + // final operation before publication so a displaced writer cannot replace + // the canonical cache with its stale snapshot. Shards written above are + // unreferenced until the envelope names them, so a lost fence publishes + // nothing. + if (verifyStillOwner && !await verifyStillOwner()) { + for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) + return false + } + + const envelope: CacheEnvelope = { + version: CACHE_VERSION, + complete: cache.complete === true, + nonce: randomBytes(8).toString('hex'), + shards, + } + await writeFileAtomic(join(dir, ENVELOPE_FILE), JSON.stringify(envelope)) + + state.dirty = false + state.dirtyProviders.clear() + state.shards = shards + // Write-through: the object just published IS the freshest state, so the + // next loadCache in this process reuses it instead of re-parsing. + cacheMemo = { dir, nonce: envelope.nonce, cache } + // Shards the new envelope no longer references are garbage; a reader that + // already opened one keeps reading it, and any failure here is swept later + // by cleanupOrphanedTempFiles. + for (const [provider, name] of Object.entries(priorShards)) { + if (shards[provider] === name) continue + await retryCacheFileMutation(() => unlink(join(dir, name))) } return true } catch (err) { - await retryCacheFileMutation(() => unlink(tempPath)) + for (const name of written) await retryCacheFileMutation(() => unlink(join(dir, name))) throw err } } @@ -798,19 +928,29 @@ export function mergeCallByDedupKey( // ── Temp Cleanup ─────────────────────────────────────────────────────── +// Sweeps our own shard directory: interrupted temp writes, plus shards the +// published envelope no longer references (a save supersedes a provider's shard +// rather than overwriting it). Temps from OTHER cache versions belong to old +// binaries mid-write and are never touched. export async function cleanupOrphanedTempFiles(): Promise { - const dir = getCodeburnCacheDir() + const dir = sessionCacheDir() if (!existsSync(dir)) return + const referenced = new Set([ENVELOPE_FILE]) + let envelopeRead = false + try { + const parsed = JSON.parse(await readFile(join(dir, ENVELOPE_FILE), 'utf-8')) + if (isEnvelope(parsed)) { + for (const name of Object.values(parsed.shards)) referenced.add(name) + envelopeRead = true + } + } catch {} + try { const entries = await readdir(dir) const now = Date.now() - - // Only our own (versioned) temp files. Legacy `session-cache.json.*.tmp` - // temps belong to old binaries mid-write and must not be touched. - const prefix = `${CACHE_FILE}.` for (const entry of entries) { - if (!entry.startsWith(prefix) || !entry.endsWith('.tmp')) continue + if (!entry.endsWith('.tmp') && (!envelopeRead || referenced.has(entry))) continue try { const fullPath = join(dir, entry) const s = await stat(fullPath) diff --git a/tests/cache-persistence-coldstart.test.ts b/tests/cache-persistence-coldstart.test.ts index c18b677f..7ab1d893 100644 --- a/tests/cache-persistence-coldstart.test.ts +++ b/tests/cache-persistence-coldstart.test.ts @@ -4,7 +4,8 @@ import { join } from 'path' import { tmpdir } from 'os' import { parseAllSessions, clearSessionCache } from '../src/parser.js' -import { CACHE_VERSION, sessionCachePath } from '../src/session-cache.js' +import { CACHE_VERSION } from '../src/session-cache.js' +import { readCacheOnDisk } from './fixtures/session-cache-io.js' let tmpDir: string let cacheDir: string @@ -48,7 +49,7 @@ describe('cold-start cache persistence', () => { const projects = await parseAllSessions() expect(projects.length).toBeGreaterThan(0) - const raw = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) + const raw = await readCacheOnDisk() expect(raw.version).toBe(CACHE_VERSION) const claudeFiles = Object.keys(raw.providers?.claude?.files ?? {}) expect(claudeFiles.length).toBeGreaterThan(0) diff --git a/tests/cache-refresh-lock.test.ts b/tests/cache-refresh-lock.test.ts index cdabaeec..b59c6834 100644 --- a/tests/cache-refresh-lock.test.ts +++ b/tests/cache-refresh-lock.test.ts @@ -8,7 +8,7 @@ import { type RefreshLockClock, } from '../src/cache-refresh-lock.js' import { clearSessionCache, parseAllSessions } from '../src/parser.js' -import { emptyCache, loadCache, saveCache, sessionCachePath } from '../src/session-cache.js' +import { emptyCache, loadCache, saveCache, sessionCacheDir } from '../src/session-cache.js' const dirs: string[] = [] @@ -202,7 +202,7 @@ describe('warm session-cache refresh lock', () => { await result.handle.release() expect(JSON.parse(await readFile(lockPath(dir), 'utf-8')).token).toBe('successor') - expect(sessionCachePath()).toContain(dir) + expect(sessionCacheDir()).toContain(dir) }) // retry shields environmental fd/CPU starvation in a saturated full-suite diff --git a/tests/codex-cache-invalidation.test.ts b/tests/codex-cache-invalidation.test.ts index e4e53111..35b80f7f 100644 --- a/tests/codex-cache-invalidation.test.ts +++ b/tests/codex-cache-invalidation.test.ts @@ -13,7 +13,7 @@ import { createHash } from 'crypto' import { join } from 'path' import { clearSessionCache, parseAllSessions } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' const testRoot = vi.hoisted(() => { const root = `${process.env['TMPDIR'] || '/tmp'}/codex-stale-repro-${process.pid}-${Date.now()}` @@ -75,8 +75,7 @@ describe('codex parser change invalidates stale session-cache (#478/#513)', () = // release: pre-fix envFingerprint, unchanged file fingerprint, cached // turns lack the mcp__ tool. Also reset codex-results.json to v4 so the // provider (if it runs at all) must genuinely re-parse. - const cachePath = sessionCachePath() - const cache = JSON.parse(await readFile(cachePath, 'utf8')) + const cache = await readCacheOnDisk() as any cache.providers.codex.envFingerprint = preFixFingerprint() for (const f of Object.values(cache.providers.codex.files) as any[]) { for (const turn of f.turns) { @@ -88,7 +87,7 @@ describe('codex parser change invalidates stale session-cache (#478/#513)', () = } } } - await writeFile(cachePath, JSON.stringify(cache)) + await writeCacheOnDisk(cache) const codexCachePath = join(CACHE_DIR, 'codex-results.json') const codexCache = JSON.parse(await readFile(codexCachePath, 'utf8')) codexCache.version = 4 diff --git a/tests/fixtures/cache-refresh-worker.ts b/tests/fixtures/cache-refresh-worker.ts index a2807df1..1a24c0c4 100644 --- a/tests/fixtures/cache-refresh-worker.ts +++ b/tests/fixtures/cache-refresh-worker.ts @@ -3,7 +3,7 @@ import { mkdir, readFile, writeFile } from 'fs/promises' import { join } from 'path' import { acquireCacheRefreshLock } from '../../src/cache-refresh-lock.js' -import { loadCache, saveCache } from '../../src/session-cache.js' +import { loadCache, markCacheDirty, saveCache } from '../../src/session-cache.js' const [cacheDir, barrierDir, id, sourcePath, bypass = 'false'] = process.argv.slice(2) if (!cacheDir || !barrierDir || !id || !sourcePath) throw new Error('missing worker argument') @@ -33,7 +33,7 @@ try { mcpInventory: [], turns: [], } - ;(cache as { _dirty?: boolean })._dirty = true + markCacheDirty(cache, 'regression') await writeFile(join(barrierDir, `${id}.parsed`), '') await waitFor(`${id}.save`) const published = await saveCache(cache, refresh?.handle.verifyStillOwner) diff --git a/tests/fixtures/session-cache-io.ts b/tests/fixtures/session-cache-io.ts new file mode 100644 index 00000000..445a769b --- /dev/null +++ b/tests/fixtures/session-cache-io.ts @@ -0,0 +1,35 @@ +// Test-side IO for the sharded session cache: the on-disk form is a directory +// (envelope + one shard per provider), so tests read and write it through the +// real load/save path instead of touching a single JSON file. +import { readFile, readdir } from 'fs/promises' +import { join } from 'path' + +import { + clearLoadCacheMemo, + loadCache, + markCacheDirty, + saveCache, + sessionCacheDir, + type SessionCache, +} from '../../src/session-cache.js' + +/** The cache exactly as it is on disk, bypassing the in-process memo. */ +export async function readCacheOnDisk(): Promise { + clearLoadCacheMemo() + return loadCache() +} + +/** Publish `cache`, rewriting every provider's shard. */ +export async function writeCacheOnDisk(cache: SessionCache): Promise { + for (const provider of Object.keys(cache.providers)) markCacheDirty(cache, provider) + await saveCache(cache) + clearLoadCacheMemo() +} + +/** Byte-level snapshot of the whole cache directory (names + contents). */ +export async function cacheDirSnapshot(): Promise { + const dir = sessionCacheDir() + const names = (await readdir(dir)).sort() + const parts = await Promise.all(names.map(async name => `${name}:${await readFile(join(dir, name), 'utf-8')}`)) + return parts.join('\n') +} diff --git a/tests/hydration-interrupt-convergence.test.ts b/tests/hydration-interrupt-convergence.test.ts index 7ad68a09..e3cb761e 100644 --- a/tests/hydration-interrupt-convergence.test.ts +++ b/tests/hydration-interrupt-convergence.test.ts @@ -6,7 +6,7 @@ import { afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest' import { loadPricing, setLocalModelSavings, setModelAliases } from '../src/models.js' import { buildMenubarPayloadForRange } from '../src/usage-aggregator.js' import { clearSessionCache } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' import { dailyCachePath } from '../src/daily-cache.js' import type { DateRange } from '../src/types.js' @@ -91,9 +91,9 @@ describe('interrupted hydration converges to the uninterrupted result', () => { // (a) Session cache: present but NOT marked complete — an interrupted cold // start's throttled partial save. - const sessionRaw = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) + const sessionRaw = await readCacheOnDisk() sessionRaw.complete = false - await writeFile(sessionCachePath(), JSON.stringify(sessionRaw), 'utf-8') + await writeCacheOnDisk(sessionRaw) // (b) Daily cache: frozen with the older days dropped but `lastComputedDate` // advanced to yesterday and NO completeness marker — the exact freeze that @@ -118,7 +118,7 @@ describe('interrupted hydration converges to the uninterrupted result', () => { expect(healed.current.calls).toBe(reference.current.calls) // And the on-disk markers are now durably complete, so the next launch is warm. - expect(JSON.parse(await readFile(sessionCachePath(), 'utf-8')).complete).toBe(true) + expect((await readCacheOnDisk()).complete).toBe(true) expect(JSON.parse(await readFile(dailyCachePath(), 'utf-8')).complete).toBe(true) }) }) diff --git a/tests/kiro-cache-invalidation.test.ts b/tests/kiro-cache-invalidation.test.ts index 0046dd7a..93bc9c1b 100644 --- a/tests/kiro-cache-invalidation.test.ts +++ b/tests/kiro-cache-invalidation.test.ts @@ -25,9 +25,9 @@ import { CACHE_VERSION, computeEnvFingerprint, fingerprintFile, - sessionCachePath, type SessionCache, } from '../src/session-cache.js' +import { writeCacheOnDisk } from './fixtures/session-cache-io.js' // The kiro provider singleton captures homedir() when its module is first // imported, so HOME must point at the test root before ../src/parser.js is @@ -99,7 +99,7 @@ async function seedCache(execPath: string, envFingerprint: string): Promise { }, } await mkdir(CACHE_DIR, { recursive: true }) - await writeFile(sessionCachePath(), JSON.stringify(cache)) + await writeCacheOnDisk(cache) clearSessionCache() const rows = await kiroCalls() diff --git a/tests/parser-antigravity-timestamp.test.ts b/tests/parser-antigravity-timestamp.test.ts index dbe11550..e394af5e 100644 --- a/tests/parser-antigravity-timestamp.test.ts +++ b/tests/parser-antigravity-timestamp.test.ts @@ -7,7 +7,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { getDateRange } from '../src/cli-date.js' import { clearSessionCache, parseAllSessions } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { readCacheOnDisk } from './fixtures/session-cache-io.js' import { isSqliteAvailable } from '../src/sqlite.js' import type { DateRange } from '../src/types.js' @@ -40,9 +40,7 @@ function createGenMetadataDb(dbPath: string, fixture: Fixture): void { } async function cachedAntigravityTurns(cacheDir: string, dbPath: string): Promise> { - const saved = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) as { - providers: Record }> }> - } + const saved = await readCacheOnDisk() return saved.providers['antigravity']?.files[dbPath]?.turns ?? [] } diff --git a/tests/parser-cache-refresh-timeout.test.ts b/tests/parser-cache-refresh-timeout.test.ts index 8883c5a7..53e53e0e 100644 --- a/tests/parser-cache-refresh-timeout.test.ts +++ b/tests/parser-cache-refresh-timeout.test.ts @@ -8,7 +8,7 @@ vi.mock('../src/cache-refresh-lock.js', () => ({ })) import { clearSessionCache, isSessionHydrationComplete, parseAllSessions } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { cacheDirSnapshot } from './fixtures/session-cache-io.js' let root: string let sessionPath: string @@ -52,12 +52,12 @@ describe('parseAllSessions warm refresh timeout', () => { it('serves the prior complete snapshot and leaves the holder cache untouched', async () => { await writeSession(50) expect(output(await parseAllSessions(undefined, 'claude'))).toBe(50) - const before = await readFile(sessionCachePath(), 'utf-8') + const before = await cacheDirSnapshot() await writeSession(5000) clearSessionCache() expect(output(await parseAllSessions(undefined, 'claude'))).toBe(50) - expect(await readFile(sessionCachePath(), 'utf-8')).toBe(before) + expect(await cacheDirSnapshot()).toBe(before) }) // The snapshot a timed-out refresh serves is only as good as what has changed diff --git a/tests/parser-gemini-cache.test.ts b/tests/parser-gemini-cache.test.ts index b5510274..d2f8c7e1 100644 --- a/tests/parser-gemini-cache.test.ts +++ b/tests/parser-gemini-cache.test.ts @@ -5,7 +5,8 @@ import { join } from 'path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { clearSessionCache, parseAllSessions } from '../src/parser.js' -import { CACHE_VERSION, computeEnvFingerprint, sessionCachePath } from '../src/session-cache.js' +import { CACHE_VERSION, computeEnvFingerprint, type SessionCache } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' import type { DateRange } from '../src/types.js' let home: string @@ -54,7 +55,7 @@ describe('Gemini session cache migration', () => { })) const fileStat = await stat(sessionPath) - await writeFile(sessionCachePath(), JSON.stringify({ + await writeCacheOnDisk({ version: CACHE_VERSION, providers: { gemini: { @@ -97,7 +98,7 @@ describe('Gemini session cache migration', () => { }, }, }, - })) + } as SessionCache) const range: DateRange = { start: new Date('2026-05-16T00:00:00.000Z'), @@ -117,7 +118,7 @@ describe('Gemini session cache migration', () => { 'gemini:gemini-session-1:g2', ]) - const savedCache = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) + const savedCache = await readCacheOnDisk() as any const savedKeys = savedCache.providers.gemini.files[sessionPath].turns.flatMap((turn: { calls: Array<{ deduplicationKey: string }> }) => turn.calls.map(call => call.deduplicationKey), ) diff --git a/tests/parser-hydration-lock.test.ts b/tests/parser-hydration-lock.test.ts index 79e14983..bc637b0b 100644 --- a/tests/parser-hydration-lock.test.ts +++ b/tests/parser-hydration-lock.test.ts @@ -6,12 +6,13 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { existsSync } from 'fs' -import { mkdir, mkdtemp, rm, unlink, writeFile, readFile } from 'fs/promises' +import { mkdir, mkdtemp, rm, unlink, writeFile } from 'fs/promises' import { tmpdir } from 'os' import { join } from 'path' import { clearSessionCache, parseAllSessions } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { sessionCacheDir } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' let tmpHome: string let cacheDir: string @@ -71,17 +72,16 @@ describe('parseAllSessions hydration lock', () => { await writeClaudeSession(50) expect(totalOutput(await parseAllSessions(undefined, 'claude'))).toBe(50) - const warm = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) - for (const section of Object.values(warm.providers) as Array<{ files: Record }> }> }>) { + const tampered = await readCacheOnDisk() + for (const section of Object.values(tampered.providers)) { for (const file of Object.values(section.files)) { for (const turn of file.turns) for (const call of turn.calls) call.usage.outputTokens = 999 } } - const tampered = JSON.stringify(warm) // Go cold: remove the versioned cache and drop the in-memory cache so the // next parse genuinely cold-starts and consults the lock. - await unlink(sessionCachePath()) + await rm(sessionCacheDir(), { recursive: true }) clearSessionCache() // A fresh lock held by another live process (pid 1 is always alive and is @@ -97,7 +97,7 @@ describe('parseAllSessions hydration lock', () => { // The "first process" finishes: it leaves the warm (tampered) cache behind // and releases the lock. The waiter wakes, reloads, and serves the cache. - await writeFile(sessionCachePath(), tampered) + await writeCacheOnDisk(tampered) await unlink(lockPath()) const result = await promise @@ -120,7 +120,7 @@ describe('parseAllSessions hydration lock', () => { // Lock released in the finally. expect(existsSync(lockPath())).toBe(false) // The parse warmed the versioned cache. - expect(existsSync(sessionCachePath())).toBe(true) + expect(existsSync(sessionCacheDir())).toBe(true) }) it('ignores a fresh lock whose pid is dead', async () => { diff --git a/tests/parser-incremental-append.test.ts b/tests/parser-incremental-append.test.ts index 3c192146..364ac663 100644 --- a/tests/parser-incremental-append.test.ts +++ b/tests/parser-incremental-append.test.ts @@ -20,7 +20,7 @@ vi.mock('../src/fs-utils.js', async (importOriginal) => { }) import { parseAllSessions, clearSessionCache } from '../src/parser.js' -import { sessionCachePath } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' import type { ProjectSummary } from '../src/types.js' let tmpDir: string @@ -129,8 +129,8 @@ describe('incremental append parsing', () => { await writeFile(sessionPath, baseLines().join('\n') + '\n') await parseWith(warmCache) - const cachedOffset: number = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) - .providers.claude.files[sessionPath].lastCompleteLineOffset + const cachedOffset = (await readCacheOnDisk()) + .providers['claude']!.files[sessionPath]!.lastCompleteLineOffset! expect(cachedOffset).toBeGreaterThan(0) // 2) append a new complete turn plus a torn (invalid JSON, no newline) tail. @@ -317,10 +317,9 @@ describe('incremental append parsing', () => { await parseWith(warmCache) // Corrupt the persisted offset to point far beyond the file, then grow it. - const cachePath = sessionCachePath() - const cache = JSON.parse(await readFile(cachePath, 'utf-8')) - cache.providers.claude.files[sessionPath].lastCompleteLineOffset = 10_000_000 - await writeFile(cachePath, JSON.stringify(cache)) + const cache = await readCacheOnDisk() + cache.providers['claude']!.files[sessionPath]!.lastCompleteLineOffset = 10_000_000 + await writeCacheOnDisk(cache) await appendFile(sessionPath, userLine('2026-05-01T13:00:00.000Z', 'grow the file') + '\n' + diff --git a/tests/parser.test.ts b/tests/parser.test.ts index ff029bc0..98c45376 100644 --- a/tests/parser.test.ts +++ b/tests/parser.test.ts @@ -14,7 +14,8 @@ import { createRequire } from 'node:module' import { isSqliteAvailable } from '../src/sqlite.js' import { clearSessionCache, parseAllSessions, setParseReuseValidator } from '../src/parser.js' -import { loadCache, saveCache, sessionCachePath } from '../src/session-cache.js' +import { loadCache, saveCache } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' import type { SessionSource, SessionParser, ParsedProviderCall } from '../src/providers/types.js' // ── Synthetic provider state ─────────────────────────────────────────────── @@ -449,12 +450,10 @@ describe('(f) durable orphans survive a parse-version bump', () => { // Simulate the fingerprint a PREVIOUS release computed (any mismatching // value takes the same code path as a real parse-version bump). - const { readFile, writeFile: writeFileFs } = await import('fs/promises') - const cachePath = sessionCachePath() - const disk = JSON.parse(await readFile(cachePath, 'utf-8')) as { providers: Record } + const disk = await readCacheOnDisk() expect(disk.providers['copilot']).toBeDefined() disk.providers['copilot']!.envFingerprint = '0000000000000000' - await writeFileFs(cachePath, JSON.stringify(disk), 'utf-8') + await writeCacheOnDisk(disk) // First parse after the "upgrade": the orphan must still be counted and // must survive in the rewritten cache, not be erased with the section. diff --git a/tests/providers/codewhale.test.ts b/tests/providers/codewhale.test.ts index bdf9fd3a..0b6a29da 100644 --- a/tests/providers/codewhale.test.ts +++ b/tests/providers/codewhale.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'os' import { join } from 'path' import { clearSessionCache, parseAllSessions } from '../../src/parser.js' -import { sessionCachePath } from '../../src/session-cache.js' +import { readCacheOnDisk } from '../fixtures/session-cache-io.js' import { MAX_SESSION_FILE_BYTES } from '../../src/fs-utils.js' import { codewhale, createCodeWhaleProvider } from '../../src/providers/codewhale.js' import type { ParsedProviderCall } from '../../src/providers/types.js' @@ -275,10 +275,8 @@ describe('codewhale provider', () => { expect(first[0]!.totalCostUSD).toBeCloseTo(0.75) expect(second[0]!.totalCostUSD).toBeCloseTo(0.75) - const cache = JSON.parse(await readFile(sessionCachePath(), 'utf-8')) as { - providers: { codewhale: { envFingerprint: string } } - } - expect(cache.providers.codewhale.envFingerprint).toMatch(/^[a-f0-9]{16}$/) + const cache = await readCacheOnDisk() + expect(cache.providers['codewhale']!.envFingerprint).toMatch(/^[a-f0-9]{16}$/) }) it('exposes canonical model and tool display names', () => { diff --git a/tests/session-cache-rich-capture.test.ts b/tests/session-cache-rich-capture.test.ts index 11301308..3ed609d7 100644 --- a/tests/session-cache-rich-capture.test.ts +++ b/tests/session-cache-rich-capture.test.ts @@ -12,8 +12,8 @@ import { emptyCache, loadCache, saveCache, - sessionCachePath, } from '../src/session-cache.js' +import { writeCacheOnDisk } from './fixtures/session-cache-io.js' const TMP_DIR = join(tmpdir(), `codeburn-rich-cache-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`) @@ -113,7 +113,7 @@ describe('session cache round-trip for rich-capture fields', () => { }, } if (!existsSync(TMP_DIR)) await mkdir(TMP_DIR, { recursive: true }) - await writeFile(sessionCachePath(), JSON.stringify(oldCache), 'utf-8') + await writeCacheOnDisk(oldCache) const loaded = await loadCache() const call = loaded.providers['claude']!.files['/x/old.jsonl']!.turns[0]!.calls[0]! diff --git a/tests/session-cache-shards.test.ts b/tests/session-cache-shards.test.ts new file mode 100644 index 00000000..f7708fa1 --- /dev/null +++ b/tests/session-cache-shards.test.ts @@ -0,0 +1,208 @@ +// Per-provider shard layout (CACHE_VERSION 8): the on-disk cache is a directory +// holding one envelope plus one shard per provider. What matters here is that +// the move off the single v7 blob loses nothing, that a save rewrites only the +// providers that changed, and that one unreadable shard costs exactly one +// provider instead of the whole cache. +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { mkdir, readFile, readdir, rm, stat, utimes, writeFile } from 'fs/promises' +import { existsSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' + +import { + CACHE_VERSION, + cleanupOrphanedTempFiles, + clearLoadCacheMemo, + loadCache, + markCacheDirty, + saveCache, + sessionCacheDir, + type CachedFile, + type SessionCache, +} from '../src/session-cache.js' + +let TMP_DIR: string + +beforeEach(async () => { + TMP_DIR = join(tmpdir(), `codeburn-shard-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`) + process.env['CODEBURN_CACHE_DIR'] = TMP_DIR + await mkdir(TMP_DIR, { recursive: true }) + clearLoadCacheMemo() +}) + +afterEach(async () => { + if (existsSync(TMP_DIR)) await rm(TMP_DIR, { recursive: true }) +}) + +function cachedFile(overrides: Partial = {}): CachedFile { + return { + fingerprint: { dev: 1, ino: 2, mtimeMs: 3, sizeBytes: 4 }, + lastCompleteLineOffset: 128, + mcpInventory: ['mcp__github__list'], + turns: [{ + timestamp: '2026-05-15T10:00:00Z', + sessionId: 'sess-1', + userMessage: 'do the thing', + calls: [{ + provider: 'claude', + model: 'claude-sonnet-4-20250514', + usage: { + inputTokens: 1000, + outputTokens: 500, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + cachedInputTokens: 0, + reasoningTokens: 0, + webSearchRequests: 0, + cacheCreationOneHourTokens: 0, + }, + costUSD: 0.01, + speed: 'standard', + timestamp: '2026-05-15T10:00:00Z', + tools: ['Read'], + bashCommands: [], + skills: [], + subagentTypes: [], + deduplicationKey: 'msg-1', + }], + }], + ...overrides, + } +} + +function v7Cache(): SessionCache { + return { + version: 7, + complete: true, + providers: { + claude: { + envFingerprint: 'claude-fp', + files: { + '/live/a.jsonl': cachedFile(), + '/live/b.jsonl': cachedFile({ turns: [] }), + // An orphaned PR-linked entry: its transcript is gone and can never + // re-parse, so the migration has to carry it across verbatim. + '/gone/pruned.jsonl': cachedFile({ prLinks: ['https://github.com/o/r/pull/1'] }), + }, + }, + codex: { + envFingerprint: 'codex-fp', + durable: true, + files: { '/live/rollout.jsonl': cachedFile() }, + }, + }, + } +} + +async function shardNames(): Promise { + return (await readdir(sessionCacheDir())).sort() +} + +describe('v7 -> shard migration', () => { + it('is lossless: every entry survives, shards replace the v7 file, reload matches', async () => { + const v7 = v7Cache() + const v7Path = join(TMP_DIR, 'session-cache.v7.json') + await writeFile(v7Path, JSON.stringify(v7)) + + const loaded = await loadCache() + // Same content, re-stamped at the current version. + expect(loaded).toEqual({ ...v7, version: CACHE_VERSION }) + + // Shards on disk, v7 blob removed. + expect(existsSync(v7Path)).toBe(false) + const names = await shardNames() + expect(names).toContain('envelope.json') + expect(names.filter(n => n.startsWith('claude.'))).toHaveLength(1) + expect(names.filter(n => n.startsWith('codex.'))).toHaveLength(1) + + // A second load reads only the shards and produces the same cache. + clearLoadCacheMemo() + expect(await loadCache()).toEqual(loaded) + }) + + it('leaves a corrupt v7 file alone and starts fresh', async () => { + await writeFile(join(TMP_DIR, 'session-cache.v7.json'), '{broken') + const loaded = await loadCache() + expect(loaded.providers).toEqual({}) + expect(existsSync(join(TMP_DIR, 'session-cache.v7.json'))).toBe(true) + }) +}) + +describe('per-provider dirty tracking', () => { + it('rewrites only the provider that changed', async () => { + await writeFile(join(TMP_DIR, 'session-cache.v7.json'), JSON.stringify(v7Cache())) + const cache = await loadCache() + + const dir = sessionCacheDir() + const before = new Map() + for (const name of await shardNames()) before.set(name, await readFile(join(dir, name), 'utf-8')) + + cache.providers['codex']!.files['/live/rollout.jsonl'] = cachedFile({ mcpInventory: ['changed'] }) + markCacheDirty(cache, 'codex') + await saveCache(cache) + + const after = await shardNames() + const claudeShard = [...before.keys()].find(n => n.startsWith('claude.'))! + // The untouched provider keeps its exact file, byte for byte. + expect(after).toContain(claudeShard) + expect(await readFile(join(dir, claudeShard), 'utf-8')).toBe(before.get(claudeShard)) + // The changed provider is republished under a new name; the old one is gone. + const codexBefore = [...before.keys()].find(n => n.startsWith('codex.'))! + const codexAfter = after.find(n => n.startsWith('codex.'))! + expect(codexAfter).not.toBe(codexBefore) + expect(after).not.toContain(codexBefore) + + clearLoadCacheMemo() + const reloaded = await loadCache() + expect(reloaded.providers['codex']!.files['/live/rollout.jsonl']!.mcpInventory).toEqual(['changed']) + expect(reloaded.providers['claude']).toEqual(cache.providers['claude']) + }) +}) + +describe('corrupt shard isolation', () => { + it('drops only the unreadable provider, keeping the rest intact', async () => { + await writeFile(join(TMP_DIR, 'session-cache.v7.json'), JSON.stringify(v7Cache())) + const cache = await loadCache() + + const dir = sessionCacheDir() + const claudeShard = (await shardNames()).find(n => n.startsWith('claude.'))! + await writeFile(join(dir, claudeShard), '{"envFingerprint":"claude-fp","files":{"/x":{"turns":') + + clearLoadCacheMemo() + const reloaded = await loadCache() + expect(reloaded.providers['claude']).toBeUndefined() + expect(reloaded.providers['codex']).toEqual(cache.providers['codex']) + }) +}) + +describe('cleanupOrphanedTempFiles', () => { + it('sweeps stale shard temps and unreferenced shards, keeping the live ones', async () => { + await saveCache({ version: CACHE_VERSION, complete: true, providers: { + claude: { envFingerprint: 'fp', files: { '/a.jsonl': cachedFile() } }, + } }) + const dir = sessionCacheDir() + const live = (await shardNames()).find(n => n.startsWith('claude.'))! + + const stale = new Date(Date.now() - 10 * 60 * 1000) + const oldTemp = join(dir, 'claude.deadbeef.json.tmp') + await writeFile(oldTemp, 'partial') + await utimes(oldTemp, stale, stale) + const orphanShard = join(dir, 'codex.deadbeef.json') + await writeFile(orphanShard, '{}') + await utimes(orphanShard, stale, stale) + const recentTemp = join(dir, 'claude.feedface.json.tmp') + await writeFile(recentTemp, 'in flight') + + await cleanupOrphanedTempFiles() + + expect(existsSync(oldTemp)).toBe(false) + expect(existsSync(orphanShard)).toBe(false) + expect(existsSync(recentTemp)).toBe(true) + expect(existsSync(join(dir, live))).toBe(true) + expect(existsSync(join(dir, 'envelope.json'))).toBe(true) + // The live shard is untouched by the sweep even though it is older than the + // temp-file age cutoff. + const info = await stat(join(dir, live)) + expect(info.size).toBeGreaterThan(0) + }) +}) diff --git a/tests/session-cache.test.ts b/tests/session-cache.test.ts index de8df9c0..718c0c77 100644 --- a/tests/session-cache.test.ts +++ b/tests/session-cache.test.ts @@ -21,11 +21,12 @@ import { mergeCallByDedupKey, reconcileFile, saveCache, - sessionCachePath, + sessionCacheDir, } from '../src/session-cache.js' +import { readCacheOnDisk, writeCacheOnDisk } from './fixtures/session-cache-io.js' -// Version-suffixed filename (e.g. session-cache.v5.json) the cache now writes to. -const CACHE_FILE = () => basename(sessionCachePath()) +// Version-suffixed directory (e.g. session-cache.v8) the cache now writes to. +const CACHE_DIR = () => basename(sessionCacheDir()) const TMP_DIR = join(tmpdir(), `codeburn-scache-test-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`) @@ -185,8 +186,7 @@ describe('loadCache / saveCache', () => { it('atomic write does not leave partial file on error', async () => { await saveCache(emptyCache()) - const raw = await readFile(sessionCachePath(), 'utf-8') - expect(JSON.parse(raw)).toEqual(emptyCache()) + expect(await readCacheOnDisk()).toEqual(emptyCache()) }) }) @@ -196,14 +196,15 @@ describe('versioned cache file + legacy adoption', () => { function validCache(): SessionCache { return { version: CACHE_VERSION, + complete: false, providers: { claude: { envFingerprint: 'abc123', files: { '/path/to/session.jsonl': makeCachedFile() } } }, } } - it('writes and reads the version-suffixed file, never the legacy name', async () => { - expect(basename(sessionCachePath())).toBe(`session-cache.v${CACHE_VERSION}.json`) + it('writes and reads the version-suffixed directory, never the legacy name', async () => { + expect(basename(sessionCacheDir())).toBe(`session-cache.v${CACHE_VERSION}`) await saveCache(validCache()) - expect(existsSync(sessionCachePath())).toBe(true) + expect(existsSync(sessionCacheDir())).toBe(true) expect(existsSync(join(TMP_DIR, 'session-cache.json'))).toBe(false) expect(await loadCache()).toEqual(validCache()) }) @@ -213,9 +214,9 @@ describe('versioned cache file + legacy adoption', () => { const legacy = join(TMP_DIR, 'session-cache.json') await writeFile(legacy, JSON.stringify(validCache())) - // Versioned file absent → adopt-copy from legacy on first load. + // Versioned directory absent → adopt-copy from legacy on first load. expect(await loadCache()).toEqual(validCache()) - expect(existsSync(sessionCachePath())).toBe(true) + expect(existsSync(sessionCacheDir())).toBe(true) // Legacy left intact (not deleted, not rewritten). expect(existsSync(legacy)).toBe(true) expect(JSON.parse(await readFile(legacy, 'utf-8'))).toEqual(validCache()) @@ -224,7 +225,7 @@ describe('versioned cache file + legacy adoption', () => { // file exists. const mutated: SessionCache = { version: CACHE_VERSION, providers: { codex: { envFingerprint: 'zzz', files: {} } } } await writeFile(legacy, JSON.stringify(mutated)) - expect(await loadCache()).toEqual(validCache()) + expect(await readCacheOnDisk()).toEqual(validCache()) }) it('ignores a different-version legacy file and never touches it', async () => { @@ -234,8 +235,8 @@ describe('versioned cache file + legacy adoption', () => { await writeFile(legacy, JSON.stringify(stale)) expect((await loadCache()).providers).toEqual({}) - // No versioned file adopted; legacy left byte-intact. - expect(existsSync(sessionCachePath())).toBe(false) + // No versioned directory adopted; legacy left byte-intact. + expect(existsSync(sessionCacheDir())).toBe(false) expect(JSON.parse(await readFile(legacy, 'utf-8'))).toEqual(stale) }) @@ -246,9 +247,9 @@ describe('versioned cache file + legacy adoption', () => { await writeFile(legacy, legacyContent) await saveCache(validCache()) - // The versioned file holds the new data; the legacy file is byte-untouched. + // The shards hold the new data; the legacy file is byte-untouched. expect(await readFile(legacy, 'utf-8')).toBe(legacyContent) - expect(JSON.parse(await readFile(sessionCachePath(), 'utf-8'))).toEqual(validCache()) + expect(await readCacheOnDisk()).toEqual(validCache()) }) }) @@ -820,7 +821,7 @@ describe('loadCache validation', () => { } } }, } await writeRawCache(cache) - expect((await loadCache())).toEqual(cache) + expect(await loadCache()).toEqual(cache) }) it('accepts a fully valid cache with all fields populated', async () => { @@ -845,7 +846,8 @@ describe('cleanupOrphanedTempFiles', () => { it('removes .tmp files older than 5 minutes', async () => { await mkdir(TMP_DIR, { recursive: true }) - const oldTmp = join(TMP_DIR, `${CACHE_FILE()}.abc123.tmp`) + await mkdir(join(TMP_DIR, CACHE_DIR()), { recursive: true }) + const oldTmp = join(TMP_DIR, CACHE_DIR(), 'claude.abc123.json.tmp') await writeFile(oldTmp, 'stale') const { utimes } = await import('fs/promises') const oldTime = new Date(Date.now() - 10 * 60 * 1000) @@ -858,7 +860,8 @@ describe('cleanupOrphanedTempFiles', () => { it('preserves recent .tmp files', async () => { await mkdir(TMP_DIR, { recursive: true }) - const recentTmp = join(TMP_DIR, `${CACHE_FILE()}.def456.tmp`) + await mkdir(join(TMP_DIR, CACHE_DIR()), { recursive: true }) + const recentTmp = join(TMP_DIR, CACHE_DIR(), 'claude.def456.json.tmp') await writeFile(recentTmp, 'recent') await cleanupOrphanedTempFiles() From 72ed163db08189de4cb0b434de3a320b56458d02 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sun, 16 Aug 2026 18:58:35 -0700 Subject: [PATCH 2/4] 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)) + }) +}) From e7f9fbc8ce8840a5809b66d84e1d45c199189ee8 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sun, 16 Aug 2026 19:01:52 -0700 Subject: [PATCH 3/4] docs(changelog): per-provider cache shards and codex incremental resume --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f6ffb1d..fad4cf06 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,8 @@ ## Unreleased ### Changed +- **A warm launch rewrites only the provider that changed.** The session cache was a single blob, so any provider appending a few KB republished the whole thing — 147 MB of stringify + fsync on a 6 GB corpus, ~18% of a warm run. It is now a version-suffixed directory holding one shard per provider plus a small envelope, written per provider and published by a single envelope rename. An existing v7 cache is re-laid-out losslessly on first load and the old file removed once the new layout is on disk: nothing re-parses. One unreadable shard now costs that provider a re-parse instead of discarding every provider's history, and partial saves during a cold parse are triggered every 2000 files rather than every 5 seconds, so a slow cold parse no longer rewrites the growing cache on a wall clock. +- **An appended Codex rollout parses only its tail.** Rollout files are append-only and the active ones run to hundreds of MB, but the Codex result cache keyed on mtime + size alone, so any growth re-read the file from byte 0. Each entry now records a restart point at the last task boundary — byte offset plus the state the single-pass decode carries across it — and a grown file with the same inode resumes there, producing output identical to a full re-parse. An entry without a usable restart point simply re-parses in full once and gains one. - **One rule for every cache file.** `CODEBURN_CACHE_DIR` when set, otherwise `~/.cache/codeburn`. `XDG_CACHE_HOME` is no longer consulted; the sync ledger, the only file that ever honored it, is merged into the canonical location on first read and the legacy copy is retired, so nothing is re-uploaded after the move. (#972) ### Fixed (Desktop & Menubar) From 0e11b5151600137d2097eb5b553eb30ee0ccc8d2 Mon Sep 17 00:00:00 2001 From: iamtoruk Date: Sun, 16 Aug 2026 19:23:39 -0700 Subject: [PATCH 4/4] fix(cache): make the shard layout safe against a second live writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two live processes share one cache directory routinely (a one-shot CLI beside the resident serve child, two menubar polls), and the shard layout had two ways to lose data there. - The atomic write used a FIXED temp name, so two writers publishing the envelope — every save does — shared one `envelope.json.tmp` and interleaved into a torn payload, or one deleted the shards the other's envelope named. 39 of 40 rounds ended in a total cache loss. The temp name carries a nonce again, as it did before the shard layout. - A save reused a shard filename from its own load snapshot without checking the file was still there. Another process republishing that provider unlinks the old shard, so the stale writer published an envelope naming a deleted file — read back as a corrupt provider and dropped whole, including PR-linked orphans no re-parse can recover. A reused shard is now existence-checked, and re-verified once more immediately before the envelope is published; a vanished one is rewritten from memory. Also: - Progress saves take a 30s floor beside the file counter. Only the claude scan reports per file; every other provider calls saveProgress once at its own boundary, so the counter alone never fired there. - The unreferenced-shard sweep waits an hour (temps still 5 minutes): an unreferenced shard may belong to a concurrent save whose envelope has not landed yet. - The sweep also retires the pre-v8 single-file temps in the parent directory, which nothing writes anymore. - The shard directory is created 0o700. - The claude and provider paths mark the cache dirty where they DELETE a stale entry, not only where they replace it: an unreadable file skips the replace, and the deletion would otherwise live only in memory. - Codex only treats a grown file as an append when the recorded boundary still lands just after a newline, so a same-inode rewrite that happens to end up larger re-parses instead of resuming mid-line. --- src/codex-cache.ts | 21 +++++ src/parser.ts | 14 ++- src/session-cache.ts | 87 ++++++++++++++----- tests/parser-hydration-lock.test.ts | 5 +- tests/providers/codex-resume.test.ts | 93 +++++++++++++++++++- tests/session-cache-shards.test.ts | 122 +++++++++++++++++++++++++-- 6 files changed, 307 insertions(+), 35 deletions(-) diff --git a/src/codex-cache.ts b/src/codex-cache.ts index c227844c..a7f2f7c1 100644 --- a/src/codex-cache.ts +++ b/src/codex-cache.ts @@ -104,6 +104,26 @@ function getEntry(cache: ResultCache, filePath: string, fp: FileFingerprint): Fi return null } +// A grown file is only assumed to be an APPEND if the recorded boundary still +// falls right after a newline. A same-inode rewrite (truncate + refill, or an +// in-place edit) that happens to end up larger would otherwise resume into the +// middle of an unrelated line. Reading one byte is cheaper than being wrong. +async function endsLineAt(filePath: string, offset: number): Promise { + if (offset === 0) return true + try { + const handle = await open(filePath, 'r') + try { + const buf = Buffer.alloc(1) + const { bytesRead } = await handle.read(buf, 0, 1, offset - 1) + return bytesRead === 1 && buf[0] === 0x0a + } finally { + await handle.close() + } + } catch { + return false + } +} + export async function readCachedCodexResults( filePath: string, ): Promise { @@ -125,6 +145,7 @@ export async function readCachedCodexResults( && stale.resumeCallCount !== undefined && fp.sizeBytes > stale.sizeBytes && stale.resumeOffset <= fp.sizeBytes + && await endsLineAt(filePath, stale.resumeOffset) ) { return { kind: 'resume', calls: stale.calls, offset: stale.resumeOffset, state: stale.resumeState, callCount: stale.resumeCallCount } } diff --git a/src/parser.ts b/src/parser.ts index af115106..03ea22f9 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -2012,7 +2012,10 @@ async function scanProjectDirs( let filesDone = 0 emitScanProgress({ kind: 'tick', provider: 'claude', done: 0, total: progressTotal }) for (const { filePath, info, append } of changedFiles) { + // Marked here, not after the re-parse: an unreadable file `continue`s out + // below, and the deletion would otherwise live only in memory. delete section.files[filePath] + markCacheDirty(diskCache, 'claude') try { if (append) { @@ -2810,6 +2813,11 @@ export function emitScanProgress(event: ScanProgressEvent): void { // that an interrupted long run loses little work, high enough that repeated // cache writes never dominate the parse. const PROGRESS_SAVE_FILE_INTERVAL = 2000 +// Only the claude scan reports per file; every other provider calls saveProgress +// once, at its own boundary. Without a time floor the counter would never reach +// the interval during a long non-claude phase and progress saves would simply +// stop happening there. +const PROGRESS_SAVE_MAX_INTERVAL_MS = 30_000 export function createScanProgress(label: string, total: number) { const show = !interactiveScanUI && total > 20 && process.stderr.isTTY === true @@ -2980,6 +2988,7 @@ async function parseProviderSources( // that pruned-away data is preserved for monotonic monthly totals. if (!provider.durableSources && !clearedPaths.has(source.path)) { delete section.files[source.path] + markCacheDirty(diskCache, providerName) clearedPaths.add(source.path) } @@ -3872,11 +3881,14 @@ async function runParse( // atomic (temp + rename) and writes only the dirty provider shards, so this // never races the final save below. let filesSinceSave = 0 + let lastSaveAt = Date.now() const saveProgress = async (): Promise => { if (!isCold || readOnly) return if (!isCacheDirty(diskCache)) return - if (++filesSinceSave < PROGRESS_SAVE_FILE_INTERVAL) return + filesSinceSave++ + if (filesSinceSave < PROGRESS_SAVE_FILE_INTERVAL && Date.now() - lastSaveAt < PROGRESS_SAVE_MAX_INTERVAL_MS) return filesSinceSave = 0 + lastSaveAt = Date.now() try { await saveCache(diskCache) } catch { /* best-effort partial save */ } } diff --git a/src/session-cache.ts b/src/session-cache.ts index 9a0010a1..07d7ef52 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -175,6 +175,12 @@ const ENVELOPE_FILE = 'envelope.json' // versioned file is absent and the legacy file's version matches ours. const LEGACY_CACHE_FILE = 'session-cache.json' const TEMP_FILE_MAX_AGE_MS = 5 * 60 * 1000 +// A shard the published envelope does not name is either superseded garbage or +// a CONCURRENT writer's shard that its envelope has not published yet. The +// second case is why this guard is an order of magnitude above the temp-file +// one: sweeping a live save's shard out from under it would publish an envelope +// naming a file that no longer exists. No save takes an hour. +const UNREFERENCED_SHARD_MAX_AGE_MS = 60 * 60 * 1000 // Env vars that change what a provider discovers or how its sessions parse. // computeEnvFingerprint hashes exactly these to decide when a provider's cache @@ -695,8 +701,11 @@ function shardFileName(provider: string): string { return `${provider.replace(/[^A-Za-z0-9_-]/g, '_')}.${randomBytes(8).toString('hex')}.json` } +// The temp name carries a nonce: two processes writing the SAME final path +// (the envelope, every save) would otherwise share one temp file and interleave +// their writes into a torn or foreign payload. async function writeFileAtomic(finalPath: string, payload: string): Promise { - const tempPath = `${finalPath}.tmp` + const tempPath = `${finalPath}.${randomBytes(8).toString('hex')}.tmp` const handle = await open(tempPath, 'w', 0o600) try { await handle.writeFile(payload, { encoding: 'utf-8' }) @@ -723,21 +732,32 @@ async function writeFileAtomic(finalPath: string, payload: string): Promise Promise): Promise { const dir = sessionCacheDir() - if (!existsSync(dir)) await mkdir(dir, { recursive: true }) + if (!existsSync(dir)) await mkdir(dir, { recursive: true, mode: 0o700 }) const state = stateOf(cache) const priorShards = state.shards const shards: Record = {} const written: string[] = [] + const writeShard = async (provider: string): Promise => { + const name = shardFileName(provider) + await writeFileAtomic(join(dir, name), JSON.stringify(cache.providers[provider])) + written.push(name) + shards[provider] = name + } + try { - for (const [provider, section] of Object.entries(cache.providers)) { + for (const provider of Object.keys(cache.providers)) { const prior = priorShards[provider] - if (prior && !state.dirtyProviders.has(provider)) { shards[provider] = prior; continue } - const name = shardFileName(provider) - await writeFileAtomic(join(dir, name), JSON.stringify(section)) - written.push(name) - shards[provider] = name + // `priorShards` is this process's snapshot from its last load or save. + // ANOTHER process may have republished that provider since, unlinking the + // file we are about to name — so reuse is conditional on the file still + // being there, and a vanished one is rewritten from memory. + if (prior && !state.dirtyProviders.has(provider) && existsSync(join(dir, prior))) { + shards[provider] = prior + continue + } + await writeShard(provider) } // The warm refresh transaction passes an ownership fence. It must be the @@ -750,6 +770,15 @@ export async function saveCache(cache: SessionCache, verifyStillOwner?: () => Pr return false } + // Last look before publishing: a concurrent save may have unlinked a reused + // shard while this one was writing its own. An envelope must never name a + // file that is already gone — that reads back as a corrupt provider and + // drops its history, including orphans no re-parse can recover. + for (const [provider, name] of Object.entries(shards)) { + if (written.includes(name) || existsSync(join(dir, name))) continue + await writeShard(provider) + } + const envelope: CacheEnvelope = { version: CACHE_VERSION, complete: cache.complete === true, @@ -928,11 +957,29 @@ export function mergeCallByDedupKey( // ── Temp Cleanup ─────────────────────────────────────────────────────── +async function unlinkIfOlderThan(path: string, maxAgeMs: number, now: number): Promise { + try { + const s = await stat(path) + if (now - s.mtimeMs > maxAgeMs) await unlink(path) + } catch {} +} + // Sweeps our own shard directory: interrupted temp writes, plus shards the -// published envelope no longer references (a save supersedes a provider's shard -// rather than overwriting it). Temps from OTHER cache versions belong to old -// binaries mid-write and are never touched. +// published envelope no longer references. Also retires the single-file layout's +// leftover temps in the parent directory, which nothing writes anymore. export async function cleanupOrphanedTempFiles(): Promise { + const now = Date.now() + const parent = getCodeburnCacheDir() + + // `session-cache.v.json..tmp` from a pre-v8 binary interrupted + // mid-write. Age-guarded, so an old binary's in-flight write is left alone. + try { + for (const entry of await readdir(parent)) { + if (!/^session-cache\.v\d+\.json\..*\.tmp$/.test(entry)) continue + await unlinkIfOlderThan(join(parent, entry), TEMP_FILE_MAX_AGE_MS, now) + } + } catch {} + const dir = sessionCacheDir() if (!existsSync(dir)) return @@ -947,17 +994,13 @@ export async function cleanupOrphanedTempFiles(): Promise { } catch {} try { - const entries = await readdir(dir) - const now = Date.now() - for (const entry of entries) { - if (!entry.endsWith('.tmp') && (!envelopeRead || referenced.has(entry))) continue - try { - const fullPath = join(dir, entry) - const s = await stat(fullPath) - if (now - s.mtimeMs > TEMP_FILE_MAX_AGE_MS) { - await unlink(fullPath) - } - } catch {} + for (const entry of await readdir(dir)) { + if (entry.endsWith('.tmp')) { + await unlinkIfOlderThan(join(dir, entry), TEMP_FILE_MAX_AGE_MS, now) + continue + } + if (!envelopeRead || referenced.has(entry)) continue + await unlinkIfOlderThan(join(dir, entry), UNREFERENCED_SHARD_MAX_AGE_MS, now) } } catch {} } diff --git a/tests/parser-hydration-lock.test.ts b/tests/parser-hydration-lock.test.ts index bc637b0b..5a41adb8 100644 --- a/tests/parser-hydration-lock.test.ts +++ b/tests/parser-hydration-lock.test.ts @@ -119,8 +119,9 @@ describe('parseAllSessions hydration lock', () => { expect(totalOutput(result)).toBe(50) // Lock released in the finally. expect(existsSync(lockPath())).toBe(false) - // The parse warmed the versioned cache. - expect(existsSync(sessionCacheDir())).toBe(true) + // The parse warmed the versioned cache: the envelope is what publishes it, + // so the directory merely existing proves nothing. + expect(existsSync(join(sessionCacheDir(), 'envelope.json'))).toBe(true) }) it('ignores a fresh lock whose pid is dead', async () => { diff --git a/tests/providers/codex-resume.test.ts b/tests/providers/codex-resume.test.ts index 35f0d45a..67c69c0f 100644 --- a/tests/providers/codex-resume.test.ts +++ b/tests/providers/codex-resume.test.ts @@ -21,7 +21,7 @@ vi.mock('../../src/fs-utils.js', async (importOriginal) => { } }) -import { flushCodexCache, withCodexCacheDirectory } from '../../src/codex-cache.js' +import { clearCodexMemCaches, flushCodexCache, withCodexCacheDirectory } from '../../src/codex-cache.js' import { createCodexProvider } from '../../src/providers/codex.js' import type { ParsedProviderCall } from '../../src/providers/types.js' @@ -111,9 +111,10 @@ async function writeRollout(lines: string[]): Promise { return path } -async function parse(cacheDir: string): Promise { +async function parse(cacheDir: string, codexDir = tmpDir): Promise { + clearCodexMemCaches() return withCodexCacheDirectory(cacheDir, async () => { - const provider = createCodexProvider(tmpDir) + const provider = createCodexProvider(codexDir) const sources = await provider.discoverSessions() const seenKeys = new Set() const calls: ParsedProviderCall[] = [] @@ -121,6 +122,7 @@ async function parse(cacheDir: string): Promise { for await (const call of provider.createSessionParser!(source, seenKeys).parse()) calls.push(call) } await flushCodexCache() + clearCodexMemCaches() return calls }) } @@ -198,3 +200,88 @@ describe('codex incremental resume', () => { expect(JSON.stringify(resumed)).toBe(JSON.stringify(full)) }) }) + +// The resume snapshot has to carry EVERY field the decode reads from an earlier +// line. Missing one shows up only when the split lands between the line that +// sets it and the line that reads it — so split at every line boundary and +// require the resumed decode to equal a full one each time. +const J = (o: unknown) => JSON.stringify(o) +const ts = (n: number, s: number) => `2026-04-14T${String(10 + Math.floor(n / 60)).padStart(2, '0')}:${String(n % 60).padStart(2, '0')}:${String(s).padStart(2, '0')}Z` + +function richTask(n: number, opts: { tokens?: false | 'empty'; reasoning?: number; model?: string; noComplete?: boolean } = {}): string[] { + const lines: string[] = [J({ type: 'event_msg', timestamp: ts(n, 0), payload: { type: 'task_started' } })] + if (opts.model) lines.push(J({ type: 'turn_context', timestamp: ts(n, 0), payload: { model: opts.model } })) + lines.push( + J({ type: 'response_item', timestamp: ts(n, 1), payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `please do task ${n} with some length` }] } }), + J({ type: 'response_item', timestamp: ts(n, 2), payload: { type: 'function_call', name: 'shell', call_id: `c${n}`, arguments: J({ command: `ls ${n}` }) } }), + J({ type: 'response_item', timestamp: ts(n, 3), payload: { type: 'function_call_output', call_id: `c${n}` } }), + J({ type: 'event_msg', timestamp: ts(n, 4), payload: { type: 'patch_apply_end', success: n % 3 !== 0, changes: { [`/Users/test/proj/f${n}.ts`]: { unified_diff: '@@ -1 +1,2 @@\n-old\n+new\n+extra\n' } } } }), + J({ type: 'event_msg', timestamp: ts(n, 5), payload: { type: 'mcp_tool_call_end', call_id: `m${n}`, invocation: { server: 'github', tool: 'list' }, duration_ms: 120, result: { Ok: {} } } }), + J({ type: 'response_item', timestamp: ts(n, 6), payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'y'.repeat(120) }] } }), + ) + if (opts.tokens === 'empty') { + // No `info`: the estimated-usage path, which advances estCounter. + lines.push(J({ type: 'event_msg', timestamp: ts(n, 7), payload: { type: 'token_count' } })) + } else if (opts.tokens !== false) { + const c = { input: 100 * n, cached: 20 * n, output: 50 * n, reasoning: (opts.reasoning ?? 10) * n } + lines.push(J({ type: 'event_msg', timestamp: ts(n, 7), payload: { type: 'token_count', info: { + last_token_usage: { input_tokens: 100, cached_input_tokens: 20, output_tokens: 50, reasoning_output_tokens: opts.reasoning ?? 10, total_tokens: 180 }, + total_token_usage: { input_tokens: c.input, cached_input_tokens: c.cached, output_tokens: c.output, reasoning_output_tokens: c.reasoning, total_tokens: c.input + c.output + c.reasoning }, + } } })) + } + if (!opts.noComplete) lines.push(J({ type: 'event_msg', timestamp: ts(n, 8), payload: { type: 'task_complete', duration_ms: 5000 } })) + return lines +} + +const RICH_LINES = [ + J({ type: 'session_meta', timestamp: ts(0, 0), payload: { cwd: '/Users/test/proj', originator: 'codex-cli', session_id: 'sess-1', model: 'gpt-5.3-codex' } }), + ...richTask(1), + ...richTask(2, { tokens: 'empty' }), + ...richTask(7, { tokens: 'empty' }), + ...richTask(3, { model: 'gpt-5.3-codex-mini' }), + ...richTask(4, { reasoning: 33 }), + ...richTask(5, { noComplete: true }), + ...richTask(6), +] + +const FORK_LINES = [ + J({ type: 'session_meta', timestamp: ts(0, 0), payload: { cwd: '/Users/test/proj', originator: 'codex-cli', session_id: 'sess-2', forked_from_id: 'sess-1', model: 'gpt-5.3-codex' } }), + // Parent history replayed inside the 5s fork cutoff: must stay skipped across a split. + ...richTask(0).map(l => l.replace(/2026-04-14T10:00:0\d/g, '2026-04-14T10:00:01')), + ...richTask(11), + ...richTask(12, { tokens: 'empty' }), +] + +describe('codex resume differential', () => { + async function rollout(lines: string[]): Promise<{ codexDir: string; path: string }> { + const codexDir = await mkdtemp(join(tmpdir(), 'codex-split-')) + const dir = join(codexDir, '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 { codexDir, path } + } + + async function assertEverySplitMatches(lines: string[]): Promise { + const base = await rollout(lines) + const full = await parse(await mkdtemp(join(tmpdir(), 'codex-c-')), base.codexDir) + expect(full.length).toBeGreaterThan(0) + + for (let k = 1; k < lines.length; k++) { + const split = await rollout(lines.slice(0, k)) + const cacheDir = await mkdtemp(join(tmpdir(), 'codex-c-')) + await parse(cacheDir, split.codexDir) + await appendFile(split.path, lines.slice(k).join('\n') + '\n') + const resumed = await parse(cacheDir, split.codexDir) + expect(JSON.stringify(resumed), `split after line ${k}: ${lines[k - 1]!.slice(0, 80)}`).toBe(JSON.stringify(full)) + } + } + + it('matches a full re-parse at every line boundary of a rich session', async () => { + await assertEverySplitMatches(RICH_LINES) + }) + + it('matches a full re-parse at every line boundary of a forked session', async () => { + await assertEverySplitMatches(FORK_LINES) + }) +}) diff --git a/tests/session-cache-shards.test.ts b/tests/session-cache-shards.test.ts index f7708fa1..a448a597 100644 --- a/tests/session-cache-shards.test.ts +++ b/tests/session-cache-shards.test.ts @@ -183,26 +183,134 @@ describe('cleanupOrphanedTempFiles', () => { const dir = sessionCacheDir() const live = (await shardNames()).find(n => n.startsWith('claude.'))! - const stale = new Date(Date.now() - 10 * 60 * 1000) + const backdate = async (path: string, minutes: number) => { + const at = new Date(Date.now() - minutes * 60 * 1000) + await utimes(path, at, at) + } + const oldTemp = join(dir, 'claude.deadbeef.json.tmp') await writeFile(oldTemp, 'partial') - await utimes(oldTemp, stale, stale) + await backdate(oldTemp, 10) const orphanShard = join(dir, 'codex.deadbeef.json') await writeFile(orphanShard, '{}') - await utimes(orphanShard, stale, stale) + await backdate(orphanShard, 90) + // Unreferenced but fresh: this is what a CONCURRENT save's shard looks like + // before its envelope lands, so the sweep must leave it alone. + const inFlightShard = join(dir, 'codex.c0ffee00.json') + await writeFile(inFlightShard, '{}') const recentTemp = join(dir, 'claude.feedface.json.tmp') await writeFile(recentTemp, 'in flight') + // The live shard is far older than the temp cutoff; being referenced is what + // protects it, not its age. + await backdate(join(dir, live), 120) await cleanupOrphanedTempFiles() expect(existsSync(oldTemp)).toBe(false) expect(existsSync(orphanShard)).toBe(false) + expect(existsSync(inFlightShard)).toBe(true) expect(existsSync(recentTemp)).toBe(true) expect(existsSync(join(dir, live))).toBe(true) expect(existsSync(join(dir, 'envelope.json'))).toBe(true) - // The live shard is untouched by the sweep even though it is older than the - // temp-file age cutoff. - const info = await stat(join(dir, live)) - expect(info.size).toBeGreaterThan(0) + }) + + it('retires the pre-v8 single-file layout temps left in the parent directory', async () => { + await saveCache({ version: CACHE_VERSION, complete: true, providers: {} }) + const legacyTemp = join(TMP_DIR, 'session-cache.v7.json.abc123.tmp') + await writeFile(legacyTemp, 'orphan from an older build') + const at = new Date(Date.now() - 10 * 60 * 1000) + await utimes(legacyTemp, at, at) + const freshLegacyTemp = join(TMP_DIR, 'session-cache.v7.json.def456.tmp') + await writeFile(freshLegacyTemp, 'an old binary mid-write') + + await cleanupOrphanedTempFiles() + + expect(existsSync(legacyTemp)).toBe(false) + expect(existsSync(freshLegacyTemp)).toBe(true) + }) +}) + +// Two live processes share one cache directory routinely: a one-shot CLI beside +// the resident serve child, or two menubar polls. Neither may publish an +// envelope naming a file that is not there — that reads back as a corrupt +// provider and silently drops its history. +describe('concurrent writers', () => { + function seed(provider: string, tag: string, files: number): SessionCache { + const cache: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + [provider]: { + envFingerprint: tag, + files: Object.fromEntries( + Array.from({ length: files }, (_, i) => [`/f/${provider}/${i}.jsonl`, cachedFile()]), + ), + }, + }, + } + markCacheDirty(cache, provider) + return cache + } + + async function assertReferentialIntegrity(expected: string[]): Promise { + const dir = sessionCacheDir() + const envelope = JSON.parse(await readFile(join(dir, 'envelope.json'), 'utf-8')) + for (const name of Object.values(envelope.shards) as string[]) { + expect(existsSync(join(dir, name)), `envelope names a missing shard: ${name}`).toBe(true) + } + clearLoadCacheMemo() + const loaded = await loadCache() + expect(Object.keys(loaded.providers).length).toBeGreaterThan(0) + for (const provider of expected) expect(loaded.providers[provider]).toBeDefined() + } + + it('never publishes a dangling envelope when two saves race', async () => { + for (let round = 0; round < 15; round++) { + await Promise.allSettled([ + saveCache(seed('claude', `a${round}`, 40)), + saveCache(seed('codex', `b${round}`, 40)), + ]) + // Whichever won, the published set has to be internally consistent and + // hold at least the provider that got there last. + await assertReferentialIntegrity([]) + } + }) + + it('a stale writer rewrites a shard another process retired instead of orphaning it', async () => { + // Seed: claude holds an expired-source PR orphan no re-parse can recover. + const initial: SessionCache = { + version: CACHE_VERSION, + complete: true, + providers: { + claude: { envFingerprint: 'fp', files: { '/gone/pruned.jsonl': cachedFile({ prLinks: ['https://github.com/o/r/pull/1'] }) } }, + codex: { envFingerprint: 'fp', durable: true, files: { '/live/r.jsonl': cachedFile() } }, + }, + } + markCacheDirty(initial, 'claude') + markCacheDirty(initial, 'codex') + await saveCache(initial) + + // Process B loads now, recording claude's current shard name. + clearLoadCacheMemo() + const b = await loadCache() + + // Process A independently touches ONLY claude and republishes, retiring the + // shard file B is still holding a name for. + clearLoadCacheMemo() + const a = await loadCache() + a.providers['claude']!.files['/live/new.jsonl'] = cachedFile() + markCacheDirty(a, 'claude') + await saveCache(a) + + // B now saves an unrelated codex change. + b.providers['codex']!.files['/live/r2.jsonl'] = cachedFile() + markCacheDirty(b, 'codex') + await saveCache(b) + + await assertReferentialIntegrity(['claude', 'codex']) + clearLoadCacheMemo() + const final = await loadCache() + expect(final.providers['claude']!.files['/gone/pruned.jsonl']).toBeDefined() + expect(final.providers['codex']!.files['/live/r2.jsonl']).toBeDefined() }) })