fix(codex): preserve session migration recovery

This commit is contained in:
Brennan Benson
2026-08-07 14:29:53 -07:00
parent cd25d1fee7
commit fed70a5cf6
10 changed files with 336 additions and 27 deletions
@@ -2,6 +2,8 @@ import type { Stats } from 'node:fs'
import { lstat } from 'node:fs/promises'
import {
appendCodexSessionHealAuditRecord,
addOwnedCopyEventId,
createCodexSessionBackfillCopyOwnershipKey,
createCodexSessionBackfillAuditWriter,
createCodexSessionBackfillDiagnosticEventId,
createCodexSessionBackfillFileEventId,
@@ -13,6 +15,7 @@ import type { CodexSessionBackfillSummary } from './codex-session-backfill-types
export type CodexSessionBackfillAuditPass = {
appendRecord: CodexSessionBackfillAuditWriter
isOwnedCopy(source: string, target: string, targetStat: Stats): boolean
recordExisting(
summary: CodexSessionBackfillSummary,
source: string,
@@ -54,6 +57,12 @@ export async function createCodexSessionBackfillAuditPass(
return {
appendRecord,
isOwnedCopy(source, target, targetStat): boolean {
const eventIds = coverage.ownedCopyEventIds.get(
createCodexSessionBackfillCopyOwnershipKey(source, target)
)
return eventIds?.has(createCodexSessionBackfillFileEventId(target, targetStat)) === true
},
async recordExisting(summary, source, target, targetStat): Promise<void> {
const fileEventId = targetStat
? createCodexSessionBackfillFileEventId(target, targetStat)
@@ -85,6 +94,9 @@ export async function createCodexSessionBackfillAuditPass(
) {
if (fileEventId) {
coverage.fileEventIds.add(fileEventId)
if (action === 'copy') {
addOwnedCopyEventId(coverage, source, target, fileEventId)
}
}
}
},
+34 -5
View File
@@ -11,6 +11,7 @@ export type CodexSessionBackfillAuditWriter = (record: Record<string, unknown>)
export type CodexSessionBackfillAuditCoverage = {
fileEventIds: Set<string>
diagnosticEventIds: Set<string>
ownedCopyEventIds: Map<string, Set<string>>
hasRunSummary: boolean
}
@@ -66,6 +67,7 @@ export async function readCodexSessionBackfillAuditCoverage(
const coverage: CodexSessionBackfillAuditCoverage = {
fileEventIds: new Set<string>(),
diagnosticEventIds: new Set<string>(),
ownedCopyEventIds: new Map<string, Set<string>>(),
hasRunSummary: false
}
const input = createReadStream(auditLogPath, { encoding: 'utf-8' })
@@ -85,6 +87,13 @@ export async function readCodexSessionBackfillAuditCoverage(
typeof record.fileEventId === 'string'
) {
coverage.fileEventIds.add(record.fileEventId)
if (
record.action === 'copy' &&
typeof record.source === 'string' &&
typeof record.target === 'string'
) {
addOwnedCopyEventId(coverage, record.source, record.target, record.fileEventId)
}
}
if (
typeof record.action === 'string' &&
@@ -106,17 +115,37 @@ export async function readCodexSessionBackfillAuditCoverage(
}
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}`
const fileIdentity = [
stat.dev,
stat.ino,
stat.birthtimeMs,
stat.size,
stat.mtimeMs,
stat.ctimeMs
].join('\0')
return createHash('sha256')
.update(normalizeRuntimePathForComparison(targetPath))
.update('\0')
.update(stableFileIdentity)
.update(fileIdentity)
.digest('hex')
}
export function createCodexSessionBackfillCopyOwnershipKey(source: string, target: string): string {
return `${normalizeRuntimePathForComparison(source)}\0${normalizeRuntimePathForComparison(target)}`
}
export function addOwnedCopyEventId(
coverage: CodexSessionBackfillAuditCoverage,
source: string,
target: string,
fileEventId: string
): void {
const key = createCodexSessionBackfillCopyOwnershipKey(source, target)
const eventIds = coverage.ownedCopyEventIds.get(key) ?? new Set<string>()
eventIds.add(fileEventId)
coverage.ownedCopyEventIds.set(key, eventIds)
}
export function createCodexSessionBackfillDiagnosticEventId(args: {
action: 'copy-unsupported' | 'failed'
source: string
+46 -2
View File
@@ -1,5 +1,6 @@
import { randomUUID } from 'node:crypto'
import { copyFile, link, rm, writeFile } from 'node:fs/promises'
import type { Stats } from 'node:fs'
import { copyFile, link, lstat, rename, rm, writeFile } from 'node:fs/promises'
import { dirname, join } from 'node:path'
const ATOMIC_NO_REPLACE_UNSUPPORTED_CODE = 'ORCA_ATOMIC_NO_REPLACE_UNSUPPORTED'
@@ -13,7 +14,7 @@ export async function copySessionFileWithoutOverwrite(
// copy cannot strand a truncated session that a later retry would skip.
await writeFile(temporaryPath, '', { encoding: 'utf-8', flag: 'wx', mode: 0o600 })
try {
await copyFile(sourcePath, temporaryPath)
await copyStableSessionFile(sourcePath, temporaryPath)
try {
// Why: this same-volume hardlink atomically installs the staged copy
// without risking a collision overwrite after an EXDEV fallback.
@@ -40,6 +41,49 @@ export async function copySessionFileWithoutOverwrite(
}
}
export async function replaceOwnedSessionCopy(
sourcePath: string,
targetPath: string,
expectedTargetStat: Stats
): Promise<void> {
const temporaryPath = join(dirname(targetPath), `.orca-backfill-${randomUUID()}.tmp`)
await writeFile(temporaryPath, '', { encoding: 'utf-8', flag: 'wx', mode: 0o600 })
try {
await copyStableSessionFile(sourcePath, temporaryPath)
const currentTargetStat = await lstat(targetPath)
if (!fileStatsMatch(currentTargetStat, expectedTargetStat)) {
const error = new Error(`Owned backfill target changed before refresh: ${targetPath}`)
;(error as NodeJS.ErrnoException).code = 'EEXIST'
throw error
}
await rename(temporaryPath, targetPath)
} finally {
await rm(temporaryPath, { force: true }).catch(() => undefined)
}
}
async function copyStableSessionFile(sourcePath: string, targetPath: string): Promise<void> {
const sourceStatBefore = await lstat(sourcePath)
await copyFile(sourcePath, targetPath)
const sourceStatAfter = await lstat(sourcePath)
if (!fileStatsMatch(sourceStatAfter, sourceStatBefore)) {
const error = new Error(`Session changed while it was copied: ${sourcePath}`)
;(error as NodeJS.ErrnoException).code = 'EBUSY'
throw error
}
}
function fileStatsMatch(left: Stats, right: Stats): boolean {
return (
left.dev === right.dev &&
left.ino === right.ino &&
left.birthtimeMs === right.birthtimeMs &&
left.size === right.size &&
left.mtimeMs === right.mtimeMs &&
left.ctimeMs === right.ctimeMs
)
}
function isExistsError(error: unknown): boolean {
return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST'
}
+15 -1
View File
@@ -1,4 +1,4 @@
import { join } from 'node:path'
import { join, relative, sep } from 'node:path'
import { listCodexSessionJsonlFilesIncrementally } from './codex-session-file-listing'
import type {
CodexSessionBackfillDate,
@@ -13,6 +13,20 @@ export function getCodexSessionBackfillDate(date = new Date()): CodexSessionBack
]
}
export function isCodexSessionRolloutPath(sessionsRoot: string, filePath: string): boolean {
const pathParts = relative(sessionsRoot, filePath).split(sep)
if (pathParts.length !== 4) {
return false
}
const [year, month, day, fileName] = pathParts
return (
/^\d{4}$/.test(year) &&
/^\d{2}$/.test(month) &&
/^\d{2}$/.test(day) &&
/^rollout-.+\.jsonl$/.test(fileName)
)
}
export async function* listCodexSessionBackfillFilesForDates(
sessionsRoot: string,
options: CodexSessionBackfillOptions,
@@ -0,0 +1,46 @@
import type { Stats } from 'node:fs'
import { describeCodexSessionBackfillErrorCode } from './codex-session-backfill-audit'
import {
readCodexSessionTargetStat,
type CodexSessionBackfillAuditPass
} from './codex-session-backfill-audit-pass'
import { replaceOwnedSessionCopy } from './codex-session-backfill-copy'
import type { CodexSessionBackfillSummary } from './codex-session-backfill-types'
export async function refreshOwnedCodexSessionCopy(args: {
source: string
target: string
targetStat: Stats
summary: CodexSessionBackfillSummary
auditPass: CodexSessionBackfillAuditPass
}): Promise<boolean> {
if (!args.auditPass.isOwnedCopy(args.source, args.target, args.targetStat)) {
return false
}
const sourceStat = await readCodexSessionTargetStat(args.source)
if (!sourceStat || sourceStat.size <= args.targetStat.size) {
return false
}
try {
await replaceOwnedSessionCopy(args.source, args.target, args.targetStat)
args.summary.copiedFiles += 1
await args.auditPass.recordPublished(args.summary, 'copy', args.source, args.target)
} catch (error) {
args.summary.failedFiles += 1
await args.auditPass.recordDiagnostic(
{
action: 'failed',
source: args.source,
target: args.target,
error: describeError(error),
errorCode: describeCodexSessionBackfillErrorCode(error)
},
sourceStat
)
}
return true
}
function describeError(error: unknown): string {
return error instanceof Error ? error.message : String(error)
}
+94 -1
View File
@@ -1,5 +1,6 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import {
appendFileSync,
existsSync,
lstatSync,
mkdirSync,
@@ -25,6 +26,7 @@ const { fsMockState } = vi.hoisted(() => ({
failLink: false,
failInstallLink: false,
failInstallLinkTransiently: false,
growSourceDuringCopy: false,
raceTargetIntoExistence: false,
failCopy: false,
failAuditMkdirOnce: false,
@@ -116,7 +118,11 @@ vi.mock('node:fs/promises', async () => {
error.code = 'EACCES'
throw error
}
return actual.copyFile(...args)
await actual.copyFile(...args)
if (fsMockState.growSourceDuringCopy) {
fsMockState.growSourceDuringCopy = false
await actual.appendFile(args[0], '{"event":"during-copy"}\n', 'utf-8')
}
},
opendir: (...args: Parameters<typeof actual.opendir>) => {
if (args[0] === fsMockState.failDirectoryPath) {
@@ -199,6 +205,7 @@ beforeEach(() => {
fsMockState.failLink = false
fsMockState.failInstallLink = false
fsMockState.failInstallLinkTransiently = false
fsMockState.growSourceDuringCopy = false
fsMockState.raceTargetIntoExistence = false
fsMockState.failCopy = false
fsMockState.failAuditMkdirOnce = false
@@ -399,6 +406,52 @@ describe('backfillManagedCodexSessionsIntoSystemHome', () => {
expect(readAuditActions()).toEqual(['copy', 'run-summary'])
})
it('refreshes an unchanged owned copy after its source grows', async () => {
fsMockState.failLink = true
const relativePath = join('2026', '05', '26', 'rollout-growing.jsonl')
const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n')
const paths = resolveCodexSessionBackfillPaths()
await backfillManagedCodexSessionsIntoSystemHome(paths)
appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8')
const refreshed = await backfillManagedCodexSessionsIntoSystemHome(paths)
expect(refreshed).toMatchObject({ copiedFiles: 1, failedFiles: 0 })
expect(readFileSync(join(getSystemSessionsRoot(), relativePath), 'utf-8')).toBe(
'{"id":"a"}\n{"event":"later"}\n'
)
})
it('does not publish a cross-volume snapshot while its source grows', async () => {
fsMockState.failLink = true
fsMockState.growSourceDuringCopy = true
const relativePath = join('2026', '05', '26', 'rollout-active.jsonl')
writeManagedSession(relativePath, '{"id":"a"}\n')
const summary = await backfillManagedCodexSessionsIntoSystemHome(
resolveCodexSessionBackfillPaths()
)
expect(summary).toMatchObject({ copiedFiles: 0, failedFiles: 1 })
expect(existsSync(join(getSystemSessionsRoot(), relativePath))).toBe(false)
})
it('does not refresh an owned copy after the target changes', async () => {
fsMockState.failLink = true
const relativePath = join('2026', '05', '26', 'rollout-diverged.jsonl')
const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n')
const targetPath = join(getSystemSessionsRoot(), relativePath)
const paths = resolveCodexSessionBackfillPaths()
await backfillManagedCodexSessionsIntoSystemHome(paths)
writeFileSync(targetPath, 'user-owned continuation\n', 'utf-8')
appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8')
const repeated = await backfillManagedCodexSessionsIntoSystemHome(paths)
expect(repeated).toMatchObject({ copiedFiles: 0, skippedExistingFiles: 1 })
expect(readFileSync(targetPath, 'utf-8')).toBe('user-owned continuation\n')
})
it('fails closed when the target filesystem cannot install without overwrite', async () => {
fsMockState.failLink = true
fsMockState.failInstallLink = true
@@ -568,6 +621,46 @@ describe('startCodexSessionBackfillInBackground', () => {
expect(scheduled).toMatchObject({ scannedFiles: 1, linkedFiles: 1 })
})
it('does not let a bounded pass certify older unscanned history', async () => {
writeManagedSession(join('2026', '05', '26', 'rollout-baseline.jsonl'), 'baseline\n')
await startCodexSessionBackfillInBackground()
invalidateCodexSessionBackfillMarker(getMarkerPath())
const missedRelativePath = join('2026', '06', '01', 'rollout-missed.jsonl')
writeManagedSession(missedRelativePath, 'missed\n')
writeManagedSession(join('2026', '08', '05', 'rollout-launch.jsonl'), 'launch\n')
const bounded = await startCodexSessionBackfillInBackground({
ignoreCompletionMarker: true,
scanDates: [['2026', '08', '05']]
})
expect(bounded).toMatchObject({ scannedFiles: 1, linkedFiles: 1 })
expect(existsSync(getMarkerPath())).toBe(false)
const recovered = await startCodexSessionBackfillInBackground()
expect(recovered).toMatchObject({ scannedFiles: 3, linkedFiles: 1 })
expect(existsSync(join(getSystemSessionsRoot(), missedRelativePath))).toBe(true)
expect(existsSync(getMarkerPath())).toBe(true)
})
it('records a new heal event when a linked rollout grows in place', async () => {
const relativePath = join('2026', '05', '26', 'rollout-growing.jsonl')
const managedPath = writeManagedSession(relativePath, '{"id":"a"}\n')
await startCodexSessionBackfillInBackground()
const firstRecord = readBackfillAuditRecords().find((record) => record.action === 'hardlink')
invalidateCodexSessionBackfillMarker(getMarkerPath())
appendFileSync(managedPath, '{"event":"later"}\n', 'utf-8')
await startCodexSessionBackfillInBackground()
const fileRecords = readBackfillAuditRecords().filter((record) =>
['hardlink', 'existing'].includes(record.action)
)
expect(fileRecords).toHaveLength(2)
expect(fileRecords[1]).toMatchObject({ action: 'existing' })
expect(fileRecords[1]?.fileEventId).not.toBe(firstRecord?.fileEventId)
})
it('keeps repeated launch invalidations audit-stable', async () => {
writeManagedSession(join('2026', '05', '26', 'rollout-a.jsonl'), '{"id":"a"}\n')
+19 -17
View File
@@ -1,5 +1,5 @@
import { link, lstat, mkdir } from 'node:fs/promises'
import { dirname, join, relative, sep } from 'node:path'
import { dirname, join, relative } from 'node:path'
import {
getCodexSessionBackfillStateDirPath,
getOrcaManagedCodexHomePath,
@@ -15,12 +15,16 @@ import {
copySessionFileWithoutOverwrite,
isAtomicNoReplaceUnsupportedError
} from './codex-session-backfill-copy'
import { listCodexSessionBackfillFilesForDates } from './codex-session-backfill-date'
import {
isCodexSessionRolloutPath,
listCodexSessionBackfillFilesForDates
} from './codex-session-backfill-date'
import {
captureCodexSessionBackfillMarkerGeneration,
hasCompletedCodexSessionBackfillMarker,
writeCodexSessionBackfillMarker as writeBackfillMarker
} from './codex-session-backfill-marker'
import { refreshOwnedCodexSessionCopy } from './codex-session-backfill-owned-copy'
import type {
CodexSessionBackfillOptions,
CodexSessionBackfillPaths,
@@ -101,6 +105,7 @@ async function runCodexSessionBackfillOncePerHost(
if (
!summary.stopped &&
options.shouldStop?.() !== true &&
options.scanDates === undefined &&
summary.failedFiles === 0 &&
summary.failedDirectories === 0 &&
summary.failedHealAuditRecords === 0
@@ -159,7 +164,7 @@ export async function backfillManagedCodexSessionsIntoSystemHome(
break
}
summary.scannedFiles += 1
if (!isCodexRolloutPath(paths.managedSessionsRoot, managedSessionFilePath)) {
if (!isCodexSessionRolloutPath(paths.managedSessionsRoot, managedSessionFilePath)) {
summary.skippedUnexpectedFiles += 1
continue
}
@@ -206,20 +211,6 @@ async function checkManagedSessionsRoot(
}
}
function isCodexRolloutPath(sessionsRoot: string, filePath: string): boolean {
const pathParts = relative(sessionsRoot, filePath).split(sep)
if (pathParts.length !== 4) {
return false
}
const [year, month, day, fileName] = pathParts
return (
/^\d{4}$/.test(year) &&
/^\d{2}$/.test(month) &&
/^\d{2}$/.test(day) &&
/^rollout-.+\.jsonl$/.test(fileName)
)
}
async function backfillOneManagedSessionFile(
paths: CodexSessionBackfillPaths,
managedSessionFilePath: string,
@@ -237,6 +228,17 @@ async function backfillOneManagedSessionFile(
const systemSessionFilePath = join(paths.systemSessionsRoot, relativePath)
const existingTargetStat = await readCodexSessionTargetStat(systemSessionFilePath)
if (existingTargetStat) {
if (
await refreshOwnedCodexSessionCopy({
source: managedSessionFilePath,
target: systemSessionFilePath,
targetStat: existingTargetStat,
summary,
auditPass
})
) {
return
}
await auditPass.recordExisting(
summary,
managedSessionFilePath,
@@ -308,4 +308,50 @@ describe('createCodexSessionMigrationScheduler', () => {
await vi.waitFor(() => expect(startBackfill).toHaveBeenCalledTimes(2))
await vi.waitFor(() => expect(startIndexHeal).toHaveBeenCalledOnce())
})
it('preserves scheduled identity and target recovery after a stopped pass', async () => {
let eligible = true
let target = '/old-history'
let releaseFirstBackfill: ((result: { stopped: boolean }) => void) | undefined
const prepareScheduledRun = vi.fn().mockReturnValueOnce(false).mockReturnValueOnce(true)
const finishScheduledRun = vi.fn()
const startBackfill = vi
.fn()
.mockImplementationOnce(
() =>
new Promise<{ stopped: boolean }>((resolve) => {
releaseFirstBackfill = resolve
})
)
.mockResolvedValueOnce({ stopped: false })
const scheduler = createCodexSessionMigrationScheduler({
isEligible: () => eligible,
isQuitting: () => false,
resolveSystemCodexHomePathOverride: () => target,
prepareScheduledRun,
finishScheduledRun,
startBackfill,
startIndexHeal: vi.fn().mockResolvedValue(null),
initialDelayMs: 1_000
})
scheduler.scheduleRun()
await vi.advanceTimersByTimeAsync(1_000)
const firstOptions = startBackfill.mock.calls[0]?.[0]
eligible = false
expect(firstOptions?.shouldStop()).toBe(true)
target = '/new-history'
releaseFirstBackfill?.({ stopped: true })
await vi.waitFor(() => expect(prepareScheduledRun).toHaveBeenCalledOnce())
expect(finishScheduledRun).not.toHaveBeenCalled()
eligible = true
scheduler.requestRun()
await vi.waitFor(() => expect(startBackfill).toHaveBeenCalledTimes(2))
expect(startBackfill).toHaveBeenLastCalledWith(
expect.objectContaining({ ignoreCompletionMarker: true, scanDates: undefined }),
'/new-history'
)
await vi.waitFor(() => expect(finishScheduledRun).toHaveBeenCalledOnce())
})
})
@@ -61,6 +61,7 @@ export function createCodexSessionMigrationScheduler(args: {
return
}
const isScheduledRun = pendingScheduledRunGeneration !== null
const activeScheduledRunGeneration = pendingScheduledRunGeneration
let preparationNeedsFullScan = false
if (isScheduledRun) {
pendingScheduledRunGeneration = null
@@ -101,10 +102,15 @@ export function createCodexSessionMigrationScheduler(args: {
void task.finally(() => {
if (migrationTask === task) {
migrationTask = null
const shouldRerun = rerunRequested || stoppedBackfill
const scheduledRunIncomplete = stoppedBackfill || activeRunStopObserved
const shouldRerun = rerunRequested || scheduledRunIncomplete
rerunRequested = false
activeRunStopObserved = false
if (shouldRerun && isScheduledRun) {
pendingScheduledRunGeneration = Math.max(
activeScheduledRunGeneration!,
pendingScheduledRunGeneration ?? activeScheduledRunGeneration!
)
pendingFullScan ||= fullScanRequired
for (const scanDate of scanDates ?? []) {
pendingScanDates.set(scanDate.join('-'), scanDate)
+17
View File
@@ -0,0 +1,17 @@
declare module 'psl' {
export type ParsedDomain = {
input: string
tld: string | null
sld: string | null
domain: string | null
subdomain: string | null
listed: boolean
}
export type ParseError = {
input: string
error: { code: string; message: string }
}
export function parse(input: string): ParsedDomain | ParseError
}