fix(codex): bound repeated session migration audits

This commit is contained in:
Brennan Benson
2026-08-04 18:51:45 -07:00
parent 5dd907fbea
commit d0981ddb03
5 changed files with 294 additions and 62 deletions
@@ -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<void>
recordPublished(
summary: CodexSessionBackfillSummary,
action: 'hardlink' | 'copy',
source: string,
target: string
): Promise<void>
finish(summary: CodexSessionBackfillSummary): Promise<void>
}
export async function createCodexSessionBackfillAuditPass(
auditLogPath: string
): Promise<CodexSessionBackfillAuditPass> {
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<void> {
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<void> {
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<void> {
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<Stats | null> {
try {
return await lstat(entryPath)
} catch {
return null
}
}
+74 -7
View File
@@ -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<string, unknown>) => Promise<boolean>
export type CodexSessionBackfillAuditCoverage = {
fileEventIds: Set<string>
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<CodexSessionBackfillAuditCoverage> {
const coverage: CodexSessionBackfillAuditCoverage = {
fileEventIds: new Set<string>(),
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<string, unknown>
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<string, unknown>
): Promise<void> {
if (!(await writer(record))) {
): Promise<boolean> {
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<void> {
target: string,
fileEventId?: string
): Promise<boolean> {
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'
}
+81 -2
View File
@@ -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 })
+39 -51
View File
@@ -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<string>()
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<boolean> {
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<string>
ensuredTargetDirectories: Set<string>,
auditPass: CodexSessionBackfillAuditPass
): Promise<void> {
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<boolean> {
}
}
/** Existence via lstat so a broken symlink at the target still counts as taken. */
async function pathEntryExists(entryPath: string): Promise<boolean> {
try {
await lstat(entryPath)
return true
} catch {
return false
}
}
function isExistsError(error: unknown): boolean {
return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST'
}
@@ -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)
})