diff --git a/src/main/codex/codex-session-backfill-audit-pass.ts b/src/main/codex/codex-session-backfill-audit-pass.ts new file mode 100644 index 00000000000..5f7e8329dc2 --- /dev/null +++ b/src/main/codex/codex-session-backfill-audit-pass.ts @@ -0,0 +1,93 @@ +import type { Stats } from 'node:fs' +import { lstat } from 'node:fs/promises' +import { + appendCodexSessionHealAuditRecord, + createCodexSessionBackfillAuditWriter, + createCodexSessionBackfillFileEventId, + readCodexSessionBackfillAuditCoverage, + recordExistingCodexSessionForHeal, + type CodexSessionBackfillAuditWriter +} from './codex-session-backfill-audit' +import type { CodexSessionBackfillSummary } from './codex-session-backfill-types' + +export type CodexSessionBackfillAuditPass = { + appendRecord: CodexSessionBackfillAuditWriter + recordExisting( + summary: CodexSessionBackfillSummary, + source: string, + target: string, + targetStat: Stats | null + ): Promise + recordPublished( + summary: CodexSessionBackfillSummary, + action: 'hardlink' | 'copy', + source: string, + target: string + ): Promise + finish(summary: CodexSessionBackfillSummary): Promise +} + +export async function createCodexSessionBackfillAuditPass( + auditLogPath: string +): Promise { + const coverage = await readCodexSessionBackfillAuditCoverage(auditLogPath) + const writeAuditRecord = createCodexSessionBackfillAuditWriter(auditLogPath) + let auditChanged = false + const appendRecord: CodexSessionBackfillAuditWriter = async (record) => { + const appended = await writeAuditRecord(record) + auditChanged ||= appended + return appended + } + + return { + appendRecord, + async recordExisting(summary, source, target, targetStat): Promise { + const fileEventId = targetStat + ? createCodexSessionBackfillFileEventId(target, targetStat) + : undefined + if (fileEventId && coverage.fileEventIds.has(fileEventId)) { + summary.skippedExistingFiles += 1 + return + } + if ( + await recordExistingCodexSessionForHeal(appendRecord, summary, source, target, fileEventId) + ) { + if (fileEventId) { + coverage.fileEventIds.add(fileEventId) + } + } + }, + async recordPublished(summary, action, source, target): Promise { + const targetStat = await readCodexSessionTargetStat(target) + const fileEventId = targetStat + ? createCodexSessionBackfillFileEventId(target, targetStat) + : undefined + if ( + await appendCodexSessionHealAuditRecord(appendRecord, summary, { + action, + source, + target, + ...(fileEventId ? { fileEventId } : {}) + }) + ) { + if (fileEventId) { + coverage.fileEventIds.add(fileEventId) + } + } + }, + async finish(summary): Promise { + if (auditChanged || !coverage.hasRunSummary) { + await appendRecord({ action: 'run-summary', ...summary }) + } + } + } +} + +/** A broken symlink at the target still counts as taken. */ +export async function readCodexSessionTargetStat(entryPath: string): Promise { + try { + return await lstat(entryPath) + } catch { + return null + } +} diff --git a/src/main/codex/codex-session-backfill-audit.ts b/src/main/codex/codex-session-backfill-audit.ts index c4740b95572..66fe7717130 100644 --- a/src/main/codex/codex-session-backfill-audit.ts +++ b/src/main/codex/codex-session-backfill-audit.ts @@ -1,10 +1,20 @@ -import { randomUUID } from 'node:crypto' +import { createHash, randomUUID } from 'node:crypto' +import { createReadStream, type Stats } from 'node:fs' import { appendFile, mkdir } from 'node:fs/promises' import { dirname } from 'node:path' +import { createInterface } from 'node:readline' +import { normalizeRuntimePathForComparison } from '../../shared/cross-platform-path' import type { CodexSessionBackfillSummary } from './codex-session-backfill-types' export type CodexSessionBackfillAuditWriter = (record: Record) => Promise +export type CodexSessionBackfillAuditCoverage = { + fileEventIds: Set + hasRunSummary: boolean +} + +const HEAL_AUDIT_ACTIONS = new Set(['hardlink', 'copy', 'existing']) + export function createCodexSessionBackfillAuditWriter( auditLogPath: string ): CodexSessionBackfillAuditWriter { @@ -48,28 +58,85 @@ export function createCodexSessionBackfillAuditWriter( } } +export async function readCodexSessionBackfillAuditCoverage( + auditLogPath: string +): Promise { + const coverage: CodexSessionBackfillAuditCoverage = { + fileEventIds: new Set(), + hasRunSummary: false + } + const input = createReadStream(auditLogPath, { encoding: 'utf-8' }) + const lines = createInterface({ input, crlfDelay: Infinity }) + try { + for await (const raw of lines) { + try { + const parsed: unknown = JSON.parse(raw) + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + continue + } + const record = parsed as Record + coverage.hasRunSummary ||= record.action === 'run-summary' + if ( + typeof record.action === 'string' && + HEAL_AUDIT_ACTIONS.has(record.action) && + typeof record.fileEventId === 'string' + ) { + coverage.fileEventIds.add(record.fileEventId) + } + } catch { + // Torn audit tails are quarantined by the writer's leading newline. + } + } + } catch (error) { + if (!isNotFoundError(error)) { + throw error + } + } + return coverage +} + +export function createCodexSessionBackfillFileEventId(targetPath: string, stat: Stats): string { + const stableFileIdentity = + stat.ino !== 0 || stat.birthtimeMs !== 0 + ? `${stat.dev}\0${stat.ino}\0${stat.birthtimeMs}` + : `${stat.size}\0${stat.mtimeMs}\0${stat.ctimeMs}` + return createHash('sha256') + .update(normalizeRuntimePathForComparison(targetPath)) + .update('\0') + .update(stableFileIdentity) + .digest('hex') +} + export async function appendCodexSessionHealAuditRecord( writer: CodexSessionBackfillAuditWriter, summary: CodexSessionBackfillSummary, record: Record -): Promise { - if (!(await writer(record))) { +): Promise { + const appended = await writer(record) + if (!appended) { summary.failedHealAuditRecords += 1 } + return appended } export async function recordExistingCodexSessionForHeal( writer: CodexSessionBackfillAuditWriter, summary: CodexSessionBackfillSummary, source: string, - target: string -): Promise { + target: string, + fileEventId?: string +): Promise { summary.skippedExistingFiles += 1 // Why: this also recovers a rollout installed before a crash or audit // failure; thread/read is idempotent for a pre-existing real-home file. - await appendCodexSessionHealAuditRecord(writer, summary, { + return appendCodexSessionHealAuditRecord(writer, summary, { action: 'existing', source, - target + target, + ...(fileEventId ? { fileEventId } : {}) }) } + +function isNotFoundError(error: unknown): boolean { + return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT' +} diff --git a/src/main/codex/codex-session-backfill.test.ts b/src/main/codex/codex-session-backfill.test.ts index f8427e8b40f..b1ef73d8edd 100644 --- a/src/main/codex/codex-session-backfill.test.ts +++ b/src/main/codex/codex-session-backfill.test.ts @@ -140,6 +140,7 @@ import { resolveCodexSessionBackfillPaths, startCodexSessionBackfillInBackground } from './codex-session-backfill' +import { invalidateCodexSessionBackfillMarker } from './codex-session-backfill-marker' let fakeHomeDir: string let userDataDir: string @@ -168,19 +169,29 @@ function writeManagedSession(relativePath: string, contents: string): string { return filePath } -function readAuditActions(): string[] { +type BackfillAuditRecord = { + action: string + target?: string + fileEventId?: string +} + +function readBackfillAuditRecords(): BackfillAuditRecord[] { return readFileSync(getAuditLogPath(), 'utf-8') .split('\n') .filter(Boolean) .flatMap((line) => { try { - return [(JSON.parse(line) as { action: string }).action] + return [JSON.parse(line) as BackfillAuditRecord] } catch { return [] } }) } +function readAuditActions(): string[] { + return readBackfillAuditRecords().map((record) => record.action) +} + beforeEach(() => { fsMockState.failLink = false fsMockState.failInstallLink = false @@ -496,6 +507,74 @@ describe('startCodexSessionBackfillInBackground', () => { ).toBe(false) }) + it('keeps repeated launch invalidations audit-stable', async () => { + writeManagedSession(join('2026', '05', '26', 'rollout-a.jsonl'), '{"id":"a"}\n') + + const first = await startCodexSessionBackfillInBackground() + expect(first).toMatchObject({ linkedFiles: 1, failedHealAuditRecords: 0 }) + const firstAudit = readFileSync(getAuditLogPath(), 'utf-8') + + for (let pass = 0; pass < 2; pass += 1) { + invalidateCodexSessionBackfillMarker(getMarkerPath()) + const repeated = await startCodexSessionBackfillInBackground() + expect(repeated).toMatchObject({ + linkedFiles: 0, + copiedFiles: 0, + skippedExistingFiles: 1, + failedHealAuditRecords: 0 + }) + expect(readFileSync(getAuditLogPath(), 'utf-8')).toBe(firstAudit) + } + + const fileRecords = readBackfillAuditRecords().filter((record) => + ['hardlink', 'copy', 'existing'].includes(record.action) + ) + expect(fileRecords).toEqual([ + expect.objectContaining({ action: 'hardlink', fileEventId: expect.any(String) }) + ]) + }) + + it('recovers a post-install audit interruption without duplicating prior events', async () => { + const firstRelativePath = join('2026', '05', '26', 'rollout-a.jsonl') + const secondRelativePath = join('2026', '05', '26', 'rollout-b.jsonl') + writeManagedSession(firstRelativePath, '{"id":"a"}\n') + await startCodexSessionBackfillInBackground() + + invalidateCodexSessionBackfillMarker(getMarkerPath()) + writeManagedSession(secondRelativePath, '{"id":"b"}\n') + fsMockState.failAuditWrites = true + + const interrupted = await startCodexSessionBackfillInBackground() + + expect(interrupted).toMatchObject({ linkedFiles: 1, failedHealAuditRecords: 1 }) + expect(existsSync(getMarkerPath())).toBe(false) + expect( + readBackfillAuditRecords().filter((record) => + ['hardlink', 'copy', 'existing'].includes(record.action) + ) + ).toHaveLength(1) + + fsMockState.failAuditWrites = false + const recovered = await startCodexSessionBackfillInBackground() + + expect(recovered).toMatchObject({ + linkedFiles: 0, + skippedExistingFiles: 2, + failedHealAuditRecords: 0 + }) + expect(existsSync(getMarkerPath())).toBe(true) + const recoveredFileRecords = readBackfillAuditRecords().filter((record) => + ['hardlink', 'copy', 'existing'].includes(record.action) + ) + expect(recoveredFileRecords).toHaveLength(2) + expect(new Set(recoveredFileRecords.map((record) => record.target))).toEqual( + new Set([ + join(getSystemSessionsRoot(), firstRelativePath), + join(getSystemSessionsRoot(), secondRelativePath) + ]) + ) + }) + it('self-heals a zero-file marker when managed rollouts appear later', async () => { const empty = await startCodexSessionBackfillInBackground() expect(empty).toMatchObject({ scannedFiles: 0, linkedFiles: 0 }) diff --git a/src/main/codex/codex-session-backfill.ts b/src/main/codex/codex-session-backfill.ts index 403c4135b52..018e6da38d0 100644 --- a/src/main/codex/codex-session-backfill.ts +++ b/src/main/codex/codex-session-backfill.ts @@ -6,11 +6,10 @@ import { getSystemCodexHomePath } from './codex-home-paths' import { - appendCodexSessionHealAuditRecord, - createCodexSessionBackfillAuditWriter, - recordExistingCodexSessionForHeal, - type CodexSessionBackfillAuditWriter -} from './codex-session-backfill-audit' + createCodexSessionBackfillAuditPass, + readCodexSessionTargetStat, + type CodexSessionBackfillAuditPass +} from './codex-session-backfill-audit-pass' import { copySessionFileWithoutOverwrite, isAtomicNoReplaceUnsupportedError @@ -135,13 +134,9 @@ export async function backfillManagedCodexSessionsIntoSystemHome( failedFiles: 0, failedHealAuditRecords: 0 } - const appendAuditRecord = createCodexSessionBackfillAuditWriter(paths.auditLogPath) + const auditPass = await createCodexSessionBackfillAuditPass(paths.auditLogPath) const ensuredTargetDirectories = new Set() - const managedSessionsRootExists = await checkManagedSessionsRoot( - paths, - summary, - appendAuditRecord - ) + const managedSessionsRootExists = await checkManagedSessionsRoot(paths, summary, auditPass) if (managedSessionsRootExists) { for await (const managedSessionFilePath of listCodexSessionJsonlFilesIncrementally( paths.managedSessionsRoot, @@ -150,7 +145,7 @@ export async function backfillManagedCodexSessionsIntoSystemHome( // Why: a partial walk must remain retryable; otherwise an unreadable // date directory would be silently omitted behind a completion marker. summary.failedDirectories += 1 - await appendAuditRecord({ + await auditPass.appendRecord({ action: 'scan-failed', source: directoryPath, error: describeError(error) @@ -174,13 +169,13 @@ export async function backfillManagedCodexSessionsIntoSystemHome( paths, managedSessionFilePath, summary, - appendAuditRecord, - ensuredTargetDirectories + ensuredTargetDirectories, + auditPass ) } } summary.stopped ||= options.shouldStop?.() === true - await appendAuditRecord({ action: 'run-summary', ...summary }) + await auditPass.finish(summary) // Why: opt-out can land while the async summary append is pending; carry it // back to the marker gate so a managed launch cannot be hidden by stale completion. summary.stopped ||= options.shouldStop?.() === true @@ -190,7 +185,7 @@ export async function backfillManagedCodexSessionsIntoSystemHome( async function checkManagedSessionsRoot( paths: CodexSessionBackfillPaths, summary: CodexSessionBackfillSummary, - appendAuditRecord: CodexSessionBackfillAuditWriter + auditPass: CodexSessionBackfillAuditPass ): Promise { try { await lstat(paths.managedSessionsRoot) @@ -202,7 +197,7 @@ async function checkManagedSessionsRoot( // Why: existsSync collapses access failures into "missing," which could // permanently hide sessions behind an incorrect completion marker. summary.failedDirectories += 1 - await appendAuditRecord({ + await auditPass.appendRecord({ action: 'scan-failed', source: paths.managedSessionsRoot, error: describeError(error) @@ -229,8 +224,8 @@ async function backfillOneManagedSessionFile( paths: CodexSessionBackfillPaths, managedSessionFilePath: string, summary: CodexSessionBackfillSummary, - appendAuditRecord: CodexSessionBackfillAuditWriter, - ensuredTargetDirectories: Set + ensuredTargetDirectories: Set, + auditPass: CodexSessionBackfillAuditPass ): Promise { if (await isSymbolicLink(managedSessionFilePath)) { // Why: bridge-created symlinks already point at a file in the user's own @@ -240,12 +235,13 @@ async function backfillOneManagedSessionFile( } const relativePath = relative(paths.managedSessionsRoot, managedSessionFilePath) const systemSessionFilePath = join(paths.systemSessionsRoot, relativePath) - if (await pathEntryExists(systemSessionFilePath)) { - await recordExistingCodexSessionForHeal( - appendAuditRecord, + const existingTargetStat = await readCodexSessionTargetStat(systemSessionFilePath) + if (existingTargetStat) { + await auditPass.recordExisting( summary, managedSessionFilePath, - systemSessionFilePath + systemSessionFilePath, + existingTargetStat ) return } @@ -260,20 +256,21 @@ async function backfillOneManagedSessionFile( } await link(managedSessionFilePath, systemSessionFilePath) summary.linkedFiles += 1 - await appendCodexSessionHealAuditRecord(appendAuditRecord, summary, { - action: 'hardlink', - source: managedSessionFilePath, - target: systemSessionFilePath - }) + await auditPass.recordPublished( + summary, + 'hardlink', + managedSessionFilePath, + systemSessionFilePath + ) } catch (linkError) { if (isExistsError(linkError)) { // Why: another window can publish the target after our existence probe; // enqueue it here too in case that writer died before its audit append. - await recordExistingCodexSessionForHeal( - appendAuditRecord, + await auditPass.recordExisting( summary, managedSessionFilePath, - systemSessionFilePath + systemSessionFilePath, + await readCodexSessionTargetStat(systemSessionFilePath) ) return } @@ -285,24 +282,25 @@ async function backfillOneManagedSessionFile( // truncated rollout, then installed without overwriting collisions. await copySessionFileWithoutOverwrite(managedSessionFilePath, systemSessionFilePath) summary.copiedFiles += 1 - await appendCodexSessionHealAuditRecord(appendAuditRecord, summary, { - action: 'copy', - source: managedSessionFilePath, - target: systemSessionFilePath - }) + await auditPass.recordPublished( + summary, + 'copy', + managedSessionFilePath, + systemSessionFilePath + ) } catch (copyError) { if (isExistsError(copyError)) { - await recordExistingCodexSessionForHeal( - appendAuditRecord, + await auditPass.recordExisting( summary, managedSessionFilePath, - systemSessionFilePath + systemSessionFilePath, + await readCodexSessionTargetStat(systemSessionFilePath) ) return } if (isAtomicNoReplaceUnsupportedError(copyError)) { summary.skippedUnsupportedFilesystemFiles += 1 - await appendAuditRecord({ + await auditPass.appendRecord({ action: 'copy-unsupported', source: managedSessionFilePath, target: systemSessionFilePath @@ -310,7 +308,7 @@ async function backfillOneManagedSessionFile( return } summary.failedFiles += 1 - await appendAuditRecord({ + await auditPass.appendRecord({ action: 'failed', source: managedSessionFilePath, target: systemSessionFilePath, @@ -329,16 +327,6 @@ async function isSymbolicLink(filePath: string): Promise { } } -/** Existence via lstat so a broken symlink at the target still counts as taken. */ -async function pathEntryExists(entryPath: string): Promise { - try { - await lstat(entryPath) - return true - } catch { - return false - } -} - function isExistsError(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST' } diff --git a/src/main/codex/codex-session-index-heal.test.ts b/src/main/codex/codex-session-index-heal.test.ts index 861c7f72b4e..efd5eac317a 100644 --- a/src/main/codex/codex-session-index-heal.test.ts +++ b/src/main/codex/codex-session-index-heal.test.ts @@ -232,7 +232,7 @@ describe('runCodexSessionIndexHeal', () => { expect(marker.healedThreads).toBe(3) }) - it('is a no-op when the marker matches the audit ledger size', async () => { + it('keeps second and later passes cheap when the audit ledger is unchanged', async () => { const rig = createHealRig({ auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }] }) @@ -245,8 +245,13 @@ describe('runCodexSessionIndexHeal', () => { buildInvocation: rig.buildInvocation, interBatchDelayMs: 0 }) + const third = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) expect(second.outcome).toBe('up-to-date') - // One spawn from the first run only — the no-op run must not hit the CLI. + expect(third.outcome).toBe('up-to-date') + // One spawn from the first run only — no-op runs must not hit the CLI. expect(rig.readLog().serverStarts).toBe(1) })