fix(codex): preserve session identity during dedup heal

This commit is contained in:
Brennan Benson
2026-07-15 18:35:44 -07:00
parent ce25b2358e
commit d74aa56813
11 changed files with 492 additions and 77 deletions
@@ -38,26 +38,48 @@ function codexSession(overrides: Partial<AiVaultSession>): AiVaultSession {
}
describe('dedupeCodexRolloutFileAliases', () => {
type Candidate = { agent: string; path: string; codexHome: string | null }
type Candidate = {
agent: string
path: string
codexHome: string | null
hardlinkIdentity?: string
}
const accessors = {
isCodex: (candidate: Candidate) => candidate.agent === 'codex',
getFilePath: (candidate: Candidate) => candidate.path,
getCodexHome: (candidate: Candidate) => candidate.codexHome
getCodexHome: (candidate: Candidate) => candidate.codexHome,
getHardlinkIdentity: (candidate: Candidate) => candidate.hardlinkIdentity ?? null
}
it('keeps the real-home alias when the same rollout exists in both roots', () => {
const managed = { agent: 'codex', path: MANAGED_HOME_ROLLOUT, codexHome: MANAGED_HOME }
const real = { agent: 'codex', path: REAL_HOME_ROLLOUT, codexHome: null }
const managed = {
agent: 'codex',
path: MANAGED_HOME_ROLLOUT,
codexHome: MANAGED_HOME,
hardlinkIdentity: '1:42'
}
const real = {
agent: 'codex',
path: REAL_HOME_ROLLOUT,
codexHome: null,
hardlinkIdentity: '1:42'
}
expect(dedupeCodexRolloutFileAliases([managed, real], accessors)).toEqual([real])
expect(dedupeCodexRolloutFileAliases([real, managed], accessors)).toEqual([real])
})
it('prefers the managed runtime home over other non-default homes', () => {
const managed = { agent: 'codex', path: MANAGED_HOME_ROLLOUT, codexHome: MANAGED_HOME }
const managed = {
agent: 'codex',
path: MANAGED_HOME_ROLLOUT,
codexHome: MANAGED_HOME,
hardlinkIdentity: '1:42'
}
const wslReal = {
agent: 'codex',
path: `\\\\wsl$\\Ubuntu\\home\\ada\\.codex\\sessions\\2026\\07\\01\\rollout-2026-07-01T10-00-00-019f0000-1111-7222-8333-444444444444.jsonl`,
codexHome: '\\\\wsl$\\Ubuntu\\home\\ada\\.codex'
codexHome: '\\\\wsl$\\Ubuntu\\home\\ada\\.codex',
hardlinkIdentity: '1:42'
}
expect(dedupeCodexRolloutFileAliases([wslReal, managed], accessors)).toEqual([managed])
})
@@ -66,12 +88,14 @@ describe('dedupeCodexRolloutFileAliases', () => {
const managed = {
agent: 'codex',
path: 'C:\\Users\\ada\\AppData\\Roaming\\orca\\codex-runtime-home\\home\\sessions\\2026\\07\\01\\rollout-2026-07-01T10-00-00-019f0000-1111-7222-8333-444444444444.jsonl',
codexHome: 'C:\\Users\\ada\\AppData\\Roaming\\orca\\codex-runtime-home\\home'
codexHome: 'C:\\Users\\ada\\AppData\\Roaming\\orca\\codex-runtime-home\\home',
hardlinkIdentity: '7:9'
}
const custom = {
agent: 'codex',
path: 'D:\\codex\\sessions\\2026\\07\\01\\rollout-2026-07-01T10-00-00-019f0000-1111-7222-8333-444444444444.jsonl',
codexHome: 'D:\\codex'
codexHome: 'D:\\codex',
hardlinkIdentity: '7:9'
}
expect(dedupeCodexRolloutFileAliases([custom, managed], accessors)).toEqual([managed])
})
@@ -96,6 +120,35 @@ describe('dedupeCodexRolloutFileAliases', () => {
claude
])
})
it('keeps same-name files unless a shared hardlink identity proves they alias', () => {
const real = {
agent: 'codex',
path: REAL_HOME_ROLLOUT,
codexHome: null,
hardlinkIdentity: '1:10'
}
const differentFile = {
agent: 'codex',
path: MANAGED_HOME_ROLLOUT,
codexHome: MANAGED_HOME,
hardlinkIdentity: '1:11'
}
const unprovenCopy = {
agent: 'codex',
path: MANAGED_HOME_ROLLOUT,
codexHome: MANAGED_HOME
}
expect(dedupeCodexRolloutFileAliases([real, differentFile], accessors)).toEqual([
real,
differentFile
])
expect(dedupeCodexRolloutFileAliases([real, unprovenCopy], accessors)).toEqual([
real,
unprovenCopy
])
})
})
describe('dedupeCodexSessionsBySessionId', () => {
@@ -129,17 +182,26 @@ describe('dedupeCodexSessionsBySessionId', () => {
})
it('never collapses across execution hosts or agents', () => {
const local = codexSession({ sessionId: 'session-1', executionHostId: 'local' })
const local = codexSession({
sessionId: 'session-1',
executionHostId: 'local',
filePath: '/home/ada/.codex/sessions/rollout-shared.jsonl'
})
const remote = codexSession({
sessionId: 'session-1',
executionHostId: 'ssh:build-box',
filePath: '/home/ada/.codex/sessions/rollout-shared.jsonl',
id: 'ssh:build-box:codex:session-1:/home/ada/.codex/sessions/x.jsonl'
})
const claude = codexSession({ sessionId: 'session-1', agent: 'claude' })
const claude = codexSession({
sessionId: 'session-1',
agent: 'claude',
filePath: '/home/ada/.codex/sessions/rollout-shared.jsonl'
})
expect(dedupeCodexSessionsBySessionId([local, remote, claude])).toEqual([local, remote, claude])
})
it('resolves same-rank id collisions to the newest row, then stable path order', () => {
it('preserves same-host session-id collisions when rollout file names differ', () => {
const older = codexSession({
sessionId: 'collision',
filePath: '/Users/ada/.codex/sessions/2026/07/01/rollout-old.jsonl',
@@ -154,16 +216,18 @@ describe('dedupeCodexSessionsBySessionId', () => {
updatedAt: '2026-07-02T10:00:00.000Z',
modifiedAt: '2026-07-02T10:00:00.000Z'
})
expect(dedupeCodexSessionsBySessionId([older, newer])).toEqual([newer])
expect(dedupeCodexSessionsBySessionId([older, newer])).toEqual([older, newer])
})
it('resolves same-rollout aliases with a stable path tie-break', () => {
const tieA = codexSession({
sessionId: 'tie',
filePath: '/Users/ada/.codex/sessions/2026/07/01/rollout-a.jsonl',
filePath: '/Users/ada/a/.codex/sessions/2026/07/01/rollout-tie.jsonl',
codexHome: null
})
const tieB = codexSession({
sessionId: 'tie',
filePath: '/Users/ada/.codex/sessions/2026/07/01/rollout-b.jsonl',
filePath: '/Users/ada/b/.codex/sessions/2026/07/01/rollout-tie.jsonl',
codexHome: null
})
expect(dedupeCodexSessionsBySessionId([tieB, tieA])).toEqual([tieA])
+60 -16
View File
@@ -7,8 +7,8 @@ import { sessionSortTime } from './session-scanner-accumulator'
// per root (#7521). These helpers collapse those aliases to one canonical row.
// Matches Codex rollout logs: rollout-<timestamp>-<session uuid>.jsonl. The
// bridge and backfill both preserve the sessions/YYYY/MM/DD layout, so an
// identical rollout file name across Codex roots is the same session.
// bridge and backfill preserve the name, but the name alone is not identity:
// pre-parse dedup also requires a shared inode and post-parse requires the id.
const CODEX_ROLLOUT_FILE_NAME_PATTERN = /^rollout-.+\.jsonl$/
// Why: not node:path.basename — a posix host scans remote/WSL win32 paths, so
@@ -17,6 +17,28 @@ function lastPathSegment(filePath: string): string {
return filePath.split(/[\\/]/).at(-1) ?? ''
}
/** Returns a pre-parse alias key only when metadata proves a shared hardlink. */
export function codexRolloutHardlinkIdentity(file: {
dev?: number
ino?: number
nlink?: number
}): string | null {
const { dev, ino, nlink } = file
if (
typeof dev !== 'number' ||
typeof ino !== 'number' ||
typeof nlink !== 'number' ||
!Number.isSafeInteger(dev) ||
!Number.isSafeInteger(ino) ||
!Number.isSafeInteger(nlink) ||
nlink <= 1 ||
(dev === 0 && ino === 0)
) {
return null
}
return `${dev}:${ino}`
}
/**
* Ranks a Codex session root for canonical-alias selection, lowest wins.
*
@@ -36,8 +58,8 @@ function codexSessionRootRank(codexHome: string | null): number {
/**
* Drops pre-parse Codex rollout candidates that alias an already-kept rollout
* file name in a preferred root, so duplicate aliases never consume the parse
* budget or crowd the capped listing.
* hardlink in a preferred root, so proven aliases never consume the parse
* budget. Same-name copies remain until parsed identity proves they alias.
*/
export function dedupeCodexRolloutFileAliases<T>(
candidates: readonly T[],
@@ -45,9 +67,10 @@ export function dedupeCodexRolloutFileAliases<T>(
isCodex: (candidate: T) => boolean
getFilePath: (candidate: T) => string
getCodexHome: (candidate: T) => string | null
getHardlinkIdentity: (candidate: T) => string | null
}
): T[] {
const bestByFileName = new Map<string, { candidate: T; rank: number; filePath: string }>()
const bestByAlias = new Map<string, { candidate: T; rank: number; filePath: string }>()
for (const candidate of candidates) {
if (!accessors.isCodex(candidate)) {
continue
@@ -57,10 +80,15 @@ export function dedupeCodexRolloutFileAliases<T>(
if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) {
continue
}
const hardlinkIdentity = accessors.getHardlinkIdentity(candidate)
if (!hardlinkIdentity) {
continue
}
const aliasKey = `${fileName}\0${hardlinkIdentity}`
const rank = codexSessionRootRank(accessors.getCodexHome(candidate))
const best = bestByFileName.get(fileName)
const best = bestByAlias.get(aliasKey)
if (!best || rank < best.rank || (rank === best.rank && filePath < best.filePath)) {
bestByFileName.set(fileName, { candidate, rank, filePath })
bestByAlias.set(aliasKey, { candidate, rank, filePath })
}
}
return candidates.filter((candidate) => {
@@ -68,39 +96,55 @@ export function dedupeCodexRolloutFileAliases<T>(
return true
}
const fileName = lastPathSegment(accessors.getFilePath(candidate))
const best = bestByFileName.get(fileName)
const hardlinkIdentity = accessors.getHardlinkIdentity(candidate)
if (!hardlinkIdentity) {
return true
}
const best = bestByAlias.get(`${fileName}\0${hardlinkIdentity}`)
return !best || best.candidate === candidate
})
}
/**
* Collapses parsed Codex sessions that share a session id on one execution
* host, keeping the canonical root's row (see codexSessionRootRank). Catches
* aliases the file-name pass cannot see: cross-volume backfill copies and
* session_meta ids that differ from the rollout file name.
* Collapses parsed Codex sessions that share a rollout name and session id on
* one execution host, keeping the canonical root's row. Requiring both the
* parsed id and rollout name preserves id collisions and same-name files whose
* parsed ids differ.
*/
export function dedupeCodexSessionsBySessionId(
sessions: readonly AiVaultSession[]
): AiVaultSession[] {
const bestByKey = new Map<string, AiVaultSession>()
for (const session of sessions) {
if (session.agent !== 'codex') {
const key = codexSessionAliasKey(session)
if (!key) {
continue
}
const key = `${session.executionHostId}:${session.sessionId}`
const best = bestByKey.get(key)
if (!best || codexSessionAliasBeats(session, best)) {
bestByKey.set(key, session)
}
}
return sessions.filter((session) => {
if (session.agent !== 'codex') {
const key = codexSessionAliasKey(session)
if (!key) {
return true
}
return bestByKey.get(`${session.executionHostId}:${session.sessionId}`) === session
return bestByKey.get(key) === session
})
}
function codexSessionAliasKey(session: AiVaultSession): string | null {
if (session.agent !== 'codex') {
return null
}
const fileName = lastPathSegment(session.filePath)
if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) {
return null
}
return `${session.executionHostId}\0${session.sessionId}\0${fileName}`
}
function codexSessionAliasBeats(candidate: AiVaultSession, best: AiVaultSession): boolean {
const candidateRank = codexSessionRootRank(candidate.codexHome)
const bestRank = codexSessionRootRank(best.codexHome)
@@ -0,0 +1,37 @@
import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types'
import type { ExecutionHostId } from '../../shared/execution-host'
import type { FileStat, IFilesystemProvider } from '../providers/types'
import type { FileWithMtime } from './session-scanner-types'
import { errorMessage } from './session-scanner-values'
export async function statRemoteSessionFile(
provider: IFilesystemProvider,
path: string,
agent: AiVaultAgent,
executionHostId: ExecutionHostId,
issues: AiVaultScanIssue[]
): Promise<FileWithMtime | null> {
try {
const stat = await provider.stat(path)
const mtimeMs = remoteSessionMtimeMs(stat)
return {
path,
mtimeMs,
modifiedAt: new Date(mtimeMs).toISOString(),
sizeBytes: stat.size,
...(typeof stat.dev === 'number' ? { dev: stat.dev } : {}),
...(typeof stat.ino === 'number' ? { ino: stat.ino } : {}),
...(typeof stat.nlink === 'number' ? { nlink: stat.nlink } : {})
}
} catch (error) {
issues.push({ executionHostId, agent, path, message: errorMessage(error) })
return null
}
}
function remoteSessionMtimeMs(stat: FileStat): number {
if (typeof stat.mtimeMs === 'number' && Number.isFinite(stat.mtimeMs)) {
return stat.mtimeMs
}
return stat.mtime > 10_000_000_000 ? stat.mtime : stat.mtime * 1000
}
+13 -29
View File
@@ -1,21 +1,22 @@
import { extname } from 'node:path'
import type {
AiVaultAgent,
AiVaultListResult,
AiVaultScanIssue,
AiVaultSession
} from '../../shared/ai-vault-types'
import { isPathInsideOrEqual } from '../../shared/cross-platform-path'
import type { ExecutionHostId } from '../../shared/execution-host'
import type { FileStat, IFilesystemProvider } from '../providers/types'
import type { IFilesystemProvider } from '../providers/types'
import type { RemoteHostPlatform } from '../ssh/ssh-remote-platform'
import { joinRemotePath } from '../ssh/ssh-remote-platform'
import { sessionSortTime } from './session-scanner-accumulator'
import {
codexRolloutHardlinkIdentity,
dedupeCodexRolloutFileAliases,
dedupeCodexSessionsBySessionId
} from './codex-session-root-dedup'
import { partitionSubagentTranscriptPaths } from './session-scanner-subagent-transcripts'
import { statRemoteSessionFile } from './remote-session-file-stat'
import type { FileWithMtime } from './session-scanner-types'
import { errorMessage } from './session-scanner-values'
import { remoteSessionSources } from './remote-session-scanner-sources'
@@ -57,7 +58,8 @@ export async function scanRemoteAiVaultSessions(args: {
{
isCodex: (candidate) => candidate.source.agent === 'codex',
getFilePath: (candidate) => candidate.file.path,
getCodexHome: (candidate) => candidate.source.codexHome ?? null
getCodexHome: (candidate) => candidate.source.codexHome ?? null,
getHardlinkIdentity: (candidate) => codexRolloutHardlinkIdentity(candidate.file)
}
)
@@ -77,9 +79,13 @@ export async function scanRemoteAiVaultSessions(args: {
scopePaths,
alreadyParsedFilePaths: parsed.parsedFilePaths
})
const scopeSessions = dedupeCodexSessionsBySessionId([
...parsedScopeSessions,
...extraScopeSessions
])
return {
sessions: mergeRemoteSessions(cappedSessions, [...parsedScopeSessions, ...extraScopeSessions]),
sessions: mergeRemoteSessions(cappedSessions, scopeSessions),
issues,
scannedAt: new Date().toISOString()
}
@@ -100,7 +106,7 @@ async function discoverRemoteSourceCandidates(args: {
: null
const paths = partition ? partition.sessionFilePaths : walked
const files = await mapRemoteScanConcurrently(paths, (path) =>
statRemoteFile(
statRemoteSessionFile(
args.context.provider,
path,
args.source.agent,
@@ -172,6 +178,8 @@ async function parseRemoteSessionCandidates(args: {
batch.map((candidate) => parseRemoteSessionCandidate(candidate, args.context, args.issues))
)
sessions.push(...results.filter(isAiVaultSession))
const uniqueSessions = dedupeCodexSessionsBySessionId(sessions)
sessions.splice(0, sessions.length, ...uniqueSessions)
index += batch.length
}
@@ -266,30 +274,6 @@ function normalizeRemoteScopePaths(scopePaths: readonly string[]): string[] {
return scopePaths.map((scopePath) => scopePath.trim()).filter(Boolean)
}
async function statRemoteFile(
provider: IFilesystemProvider,
path: string,
agent: AiVaultAgent,
executionHostId: ExecutionHostId,
issues: AiVaultScanIssue[]
): Promise<FileWithMtime | null> {
try {
const stat = await provider.stat(path)
const mtimeMs = remoteStatMtimeMs(stat)
return { path, mtimeMs, modifiedAt: new Date(mtimeMs).toISOString() }
} catch (err) {
issues.push({ executionHostId, agent, path, message: errorMessage(err) })
return null
}
}
function remoteStatMtimeMs(stat: FileStat): number {
if (typeof stat.mtimeMs === 'number' && Number.isFinite(stat.mtimeMs)) {
return stat.mtimeMs
}
return stat.mtime > 10_000_000_000 ? stat.mtime : stat.mtime * 1000
}
function canStopParsingRemoteSessions(
sessions: AiVaultSession[],
limit: number,
@@ -1,4 +1,4 @@
import { link, mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'
import { copyFile, link, mkdtemp, mkdir, rm, utimes, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
@@ -113,4 +113,152 @@ describe('scanAiVaultSessions codex dual-root dedup', () => {
resumeCommand: `cd '/repo/app' && CODEX_HOME='${managedHome}' codex resume '029f0000-1111-7222-8333-555555555555'`
})
})
it('keeps different same-name rollouts from separate roots', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-collision-'))
tempRoots.push(root)
const roots = isolatedScanRoots(root)
const realHome = join(root, 'real-codex-home')
const realSessionsDir = join(realHome, 'sessions')
const managedHome = join(root, 'codex-runtime-home', 'home')
const managedSessionsDir = join(managedHome, 'sessions')
const rolloutName = 'rollout-2026-07-01T10-00-00-collision.jsonl'
const realPath = join(realSessionsDir, rolloutName)
const managedPath = join(managedSessionsDir, rolloutName)
await mkdir(realSessionsDir, { recursive: true })
await mkdir(managedSessionsDir, { recursive: true })
await writeFile(
realPath,
jsonLines([
{
timestamp: '2026-07-01T10:00:00.000Z',
type: 'session_meta',
payload: { id: 'real-session', cwd: '/repo/real' }
},
{
timestamp: '2026-07-01T10:00:01.000Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'text', text: 'Real content' }]
}
}
])
)
await writeFile(
managedPath,
jsonLines([
{
timestamp: '2026-07-01T10:00:00.000Z',
type: 'session_meta',
payload: { id: 'managed-session', cwd: '/repo/managed' }
},
{
timestamp: '2026-07-01T10:00:01.000Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'text', text: 'Managed content' }]
}
}
])
)
const result = await scanAiVaultSessions({
...roots,
codexSessionsDir: realSessionsDir,
defaultCodexHomeDir: realHome,
additionalCodexSessionsDirs: [managedSessionsDir],
platform: 'darwin'
})
expect(result.issues).toEqual([])
expect(result.sessions.map((session) => session.sessionId).sort()).toEqual([
'managed-session',
'real-session'
])
})
it('fills the listing past cross-volume-style copy aliases', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-copy-cap-'))
tempRoots.push(root)
const roots = isolatedScanRoots(root)
const realHome = join(root, 'real-codex-home')
const realSessionsDir = join(realHome, 'sessions')
const managedHome = join(root, 'codex-runtime-home', 'home')
const managedSessionsDir = join(managedHome, 'sessions')
const aliasName = 'rollout-2026-07-02T10-00-00-copy-alias.jsonl'
const realAliasPath = join(realSessionsDir, aliasName)
const managedAliasPath = join(managedSessionsDir, aliasName)
const uniquePath = join(managedSessionsDir, 'rollout-2026-07-01T10-00-00-unique.jsonl')
await mkdir(realSessionsDir, { recursive: true })
await mkdir(managedSessionsDir, { recursive: true })
await writeFile(
realAliasPath,
jsonLines([
{
timestamp: '2026-07-02T10:00:00.000Z',
type: 'session_meta',
payload: { id: 'copied-session', cwd: '/repo/copied' }
},
{
timestamp: '2026-07-02T10:00:01.000Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'text', text: 'Copied content' }]
}
}
])
)
await copyFile(realAliasPath, managedAliasPath)
await writeFile(
uniquePath,
jsonLines([
{
timestamp: '2026-07-01T10:00:00.000Z',
type: 'session_meta',
payload: { id: 'unique-session', cwd: '/repo/unique' }
},
{
timestamp: '2026-07-01T10:00:01.000Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'text', text: 'Unique content' }]
}
}
])
)
const newest = new Date('2026-07-02T10:00:00.000Z')
const older = new Date('2026-07-01T10:00:00.000Z')
await utimes(realAliasPath, newest, newest)
await utimes(managedAliasPath, newest, newest)
await utimes(uniquePath, older, older)
const result = await scanAiVaultSessions({
...roots,
codexSessionsDir: realSessionsDir,
defaultCodexHomeDir: realHome,
additionalCodexSessionsDirs: [managedSessionsDir],
platform: 'darwin',
limit: 2
})
expect(result.issues).toEqual([])
expect(result.sessions.map((session) => session.sessionId).sort()).toEqual([
'copied-session',
'unique-session'
])
expect(result.sessions.find((session) => session.sessionId === 'copied-session')).toMatchObject(
{
codexHome: null,
filePath: realAliasPath
}
)
})
})
@@ -26,7 +26,10 @@ export async function discoverFiles(args: {
path,
mtimeMs: fileStat.mtimeMs,
modifiedAt: fileStat.mtime.toISOString(),
sizeBytes: fileStat.size
sizeBytes: fileStat.size,
dev: fileStat.dev,
ino: fileStat.ino,
nlink: fileStat.nlink
})
} catch (err) {
args.issues.push({ agent: args.agent, path, message: errorMessage(err) })
+6 -1
View File
@@ -47,8 +47,13 @@ export type FileWithMtime = {
modifiedAt: string
// Present when discovery statted the file; lets the parse cache detect
// unchanged/truncated files without a second stat. Synthetic candidates
// (OpenCode SQLite rows, remote files) omit it.
// such as OpenCode SQLite rows omit it.
sizeBytes?: number
// Present when discovery can prove filesystem identity. Codex dual-root
// scans use a multi-link inode to collapse only actual hardlink aliases.
dev?: number
ino?: number
nlink?: number
}
export type SessionFileCandidate = {
+8 -1
View File
@@ -7,6 +7,7 @@ import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/exec
import { withSpan } from '../observability/tracer'
import { sessionSortTime } from './session-scanner-accumulator'
import {
codexRolloutHardlinkIdentity,
dedupeCodexRolloutFileAliases,
dedupeCodexSessionsBySessionId
} from './codex-session-root-dedup'
@@ -81,7 +82,8 @@ export async function scanAiVaultSessions(
{
isCodex: (candidate) => candidate.agent === 'codex',
getFilePath: (candidate) => candidate.file.path,
getCodexHome: (candidate) => candidate.codexHome
getCodexHome: (candidate) => candidate.codexHome,
getHardlinkIdentity: (candidate) => codexRolloutHardlinkIdentity(candidate.file)
}
)
@@ -217,6 +219,11 @@ async function parseSessionCandidates(args: {
}
}
// Why: cross-volume backfill copies have no shared inode, so collapse
// parsed aliases before they can crowd the unique-session parse budget.
const uniqueSessions = dedupeCodexSessionsBySessionId(sessions)
sessions.splice(0, sessions.length, ...uniqueSessions)
index += batchSize
}
@@ -1,5 +1,9 @@
import { appendFileSync, mkdirSync, readFileSync, statSync } from 'node:fs'
import { dirname } from 'node:path'
import {
isPathInsideOrEqual,
normalizeRuntimePathForComparison
} from '../../shared/cross-platform-path'
import { writeFileAtomically } from '../codex-accounts/fs-utils'
// State files for the session index heal: which backfilled rollouts exist
@@ -9,7 +13,7 @@ import { writeFileAtomically } from '../codex-accounts/fs-utils'
// Bump to re-drive the heal for every host after a semantics change; already
// processed thread ids are re-read because ledger lines are version-scoped.
export const CODEX_SESSION_INDEX_HEAL_VERSION = 1
export const CODEX_SESSION_INDEX_HEAL_VERSION = 2
// Why: an unsupported CLI stays unsupported until upgraded; re-probing once a
// day is enough to notice an upgrade without a per-startup spawn.
@@ -44,12 +48,17 @@ export type HealMarkerSummary = {
* copied rollout whose thread id has not been processed yet, most recent first.
*/
export function collectPendingHealThreads(paths: CodexSessionIndexHealPaths): PendingHealThread[] {
const processedThreadIds = readProcessedHealThreadIds(paths.healLedgerPath)
const processedThreadIds = readProcessedHealThreadIds(paths)
const pendingByThreadId = new Map<string, PendingHealThread>()
for (const line of readJsonlLines(paths.auditLogPath)) {
if ((line.action !== 'hardlink' && line.action !== 'copy') || typeof line.target !== 'string') {
continue
}
// Why: the append-only audit can contain runs for several custom Codex
// homes; only thread/read ids whose rollout lives in this invocation's DB.
if (!isPathInsideOrEqual(paths.systemSessionsRoot, line.target)) {
continue
}
const match = CODEX_ROLLOUT_THREAD_ID_PATTERN.exec(lastPathSegment(line.target))
if (!match) {
continue
@@ -69,10 +78,16 @@ function lastPathSegment(filePath: string): string {
return filePath.split(/[\\/]/).at(-1) ?? ''
}
function readProcessedHealThreadIds(healLedgerPath: string): Set<string> {
function readProcessedHealThreadIds(paths: CodexSessionIndexHealPaths): Set<string> {
const processed = new Set<string>()
for (const line of readJsonlLines(healLedgerPath)) {
if (line.v === CODEX_SESSION_INDEX_HEAL_VERSION && typeof line.threadId === 'string') {
const expectedRoot = normalizeRuntimePathForComparison(paths.systemSessionsRoot)
for (const line of readJsonlLines(paths.healLedgerPath)) {
if (
line.v === CODEX_SESSION_INDEX_HEAL_VERSION &&
typeof line.threadId === 'string' &&
typeof line.systemSessionsRoot === 'string' &&
normalizeRuntimePathForComparison(line.systemSessionsRoot) === expectedRoot
) {
processed.add(line.threadId.toLowerCase())
}
}
@@ -80,16 +95,17 @@ function readProcessedHealThreadIds(healLedgerPath: string): Set<string> {
}
export function appendHealLedgerRecord(
healLedgerPath: string,
paths: CodexSessionIndexHealPaths,
threadId: string,
outcome: HealLedgerOutcome
): void {
try {
mkdirSync(dirname(healLedgerPath), { recursive: true })
mkdirSync(dirname(paths.healLedgerPath), { recursive: true })
appendFileSync(
healLedgerPath,
paths.healLedgerPath,
`${JSON.stringify({
v: CODEX_SESSION_INDEX_HEAL_VERSION,
systemSessionsRoot: paths.systemSessionsRoot,
threadId,
outcome,
at: new Date().toISOString()
@@ -10,6 +10,7 @@ import {
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import type { CodexAppServerInvocation } from './codex-app-server-session'
import { CODEX_SESSION_INDEX_HEAL_VERSION } from './codex-session-index-heal-state'
import {
runCodexSessionIndexHeal,
type CodexSessionIndexHealPaths
@@ -66,6 +67,10 @@ process.stdin.on('data', (chunk) => {
send({ id: message.id, error: { code: -32600, message: 'failed to parse rollout' } })
return
}
if ((config.busyThreadIds || []).includes(threadId)) {
send({ id: message.id, error: { code: -32600, message: 'database is locked' } })
return
}
if (config.scenario === 'die-mid-batch' && threadId === config.dieOnThreadId) {
process.exit(7)
}
@@ -100,6 +105,7 @@ function createHealRig(options: {
auditedThreads?: { stamp: string; id: string; action?: string }[]
missingThreadIds?: string[]
failingThreadIds?: string[]
busyThreadIds?: string[]
dieOnThreadId?: string
}): {
paths: CodexSessionIndexHealPaths
@@ -145,6 +151,7 @@ function createHealRig(options: {
readLogFile,
missingThreadIds: options.missingThreadIds ?? [],
failingThreadIds: options.failingThreadIds ?? [],
busyThreadIds: options.busyThreadIds ?? [],
dieOnThreadId: options.dieOnThreadId
})
},
@@ -317,6 +324,25 @@ describe('runCodexSessionIndexHeal', () => {
expect(log.maxInFlight).toBeLessThanOrEqual(2)
})
it('caps overrides at the production batch and concurrency limits', async () => {
const rig = createHealRig({
auditedThreads: Array.from({ length: 51 }, (_, index) => ({
stamp: `2026-07-${String(index + 1).padStart(2, '0')}T10-00-00`,
id: threadId(String(index + 1))
}))
})
const summary = await runCodexSessionIndexHeal(rig.paths, {
buildInvocation: rig.buildInvocation,
readsPerServerSession: 1_000,
readConcurrency: 1_000,
interBatchDelayMs: 0
})
expect(summary).toMatchObject({ outcome: 'completed', healedThreads: 51 })
expect(rig.readLog()).toMatchObject({ serverStarts: 2, maxInFlight: 2 })
})
it('stops promptly when shouldStop flips and resumes on the next pass', async () => {
const rig = createHealRig({
auditedThreads: Array.from({ length: 4 }, (_, index) => ({
@@ -416,6 +442,39 @@ describe('runCodexSessionIndexHeal', () => {
expect(retried.healedThreads).toBe(2)
})
it('retries transient sqlite contention instead of marking the thread failed', async () => {
const rig = createHealRig({
auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }],
busyThreadIds: [threadId('1')]
})
const summary = await runCodexSessionIndexHeal(rig.paths, {
buildInvocation: rig.buildInvocation,
interBatchDelayMs: 0
})
expect(summary.outcome).toBe('aborted')
expect(readLedgerOutcomes(rig.paths)).toEqual({})
const retried = await runCodexSessionIndexHeal(rig.paths, {
buildInvocation: (home, timeoutMs) => {
const invocation = rig.buildInvocation(home, timeoutMs)
return {
...invocation,
env: {
STUB_CONFIG: JSON.stringify({
scenario: 'ok',
readLogFile: rig.readLogFile,
missingThreadIds: [],
failingThreadIds: [],
busyThreadIds: []
})
}
}
},
interBatchDelayMs: 0
})
expect(retried).toMatchObject({ outcome: 'completed', healedThreads: 1 })
})
it('completes immediately with no server spawn when there is nothing to heal', async () => {
const rig = createHealRig({ auditedThreads: [] })
const summary = await runCodexSessionIndexHeal(rig.paths, {
@@ -448,4 +507,37 @@ describe('runCodexSessionIndexHeal', () => {
expect(summary).toMatchObject({ outcome: 'completed', pendingThreads: 0 })
expect(rig.readLog().serverStarts).toBe(0)
})
it('scopes audit and processed ledger records to the current Codex home', async () => {
const currentId = threadId('1')
const foreignId = threadId('2')
const rig = createHealRig({
auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: currentId }]
})
const foreignRoot = `${rig.paths.systemSessionsRoot}-other`
appendFileSync(
rig.paths.auditLogPath,
`${JSON.stringify({
action: 'hardlink',
target: rolloutTarget(foreignRoot, '2026-07-02T10-00-00', foreignId)
})}\n`
)
appendFileSync(
rig.paths.healLedgerPath,
`${JSON.stringify({
v: CODEX_SESSION_INDEX_HEAL_VERSION,
systemSessionsRoot: foreignRoot,
threadId: currentId,
outcome: 'healed'
})}\n`
)
const summary = await runCodexSessionIndexHeal(rig.paths, {
buildInvocation: rig.buildInvocation,
interBatchDelayMs: 0
})
expect(summary).toMatchObject({ outcome: 'completed', healedThreads: 1 })
expect(rig.readLog().threadIds).toEqual([currentId])
})
})
+21 -6
View File
@@ -132,8 +132,11 @@ export async function runCodexSessionIndexHeal(
const systemCodexHomePath = dirname(paths.systemSessionsRoot)
const buildInvocation = options.buildInvocation ?? buildNativeHealInvocation
const readsPerServerSession = options.readsPerServerSession ?? HEAL_READS_PER_SERVER_SESSION
const readConcurrency = options.readConcurrency ?? HEAL_READ_CONCURRENCY
const readsPerServerSession = resolveHealWorkLimit(
options.readsPerServerSession,
HEAL_READS_PER_SERVER_SESSION
)
const readConcurrency = resolveHealWorkLimit(options.readConcurrency, HEAL_READ_CONCURRENCY)
const interBatchDelayMs = options.interBatchDelayMs ?? HEAL_INTER_BATCH_DELAY_MS
const shouldStop = options.shouldStop ?? ((): boolean => false)
@@ -159,7 +162,7 @@ export async function runCodexSessionIndexHeal(
await healOneThread(rpc, thread, paths, summary)
}
}
await Promise.all(Array.from({ length: Math.max(1, readConcurrency) }, () => worker()))
await Promise.all(Array.from({ length: readConcurrency }, () => worker()))
}
)
} catch (error) {
@@ -195,7 +198,7 @@ async function healOneThread(
try {
await rpc.request('thread/read', { threadId: thread.threadId })
summary.healedThreads += 1
appendHealLedgerRecord(paths.healLedgerPath, thread.threadId, 'healed')
appendHealLedgerRecord(paths, thread.threadId, 'healed')
} catch (error) {
if (isCodexAppServerUnsupportedError(error)) {
throw error
@@ -209,14 +212,26 @@ async function healOneThread(
if (/no rollout found/i.test(message)) {
// The backfilled rollout was deleted after the audit was written.
summary.missingThreads += 1
appendHealLedgerRecord(paths.healLedgerPath, thread.threadId, 'missing')
appendHealLedgerRecord(paths, thread.threadId, 'missing')
return
}
if (/SQLITE_(?:BUSY|LOCKED)|database (?:is )?(?:busy|locked)/i.test(message)) {
// Why: an active Codex process can briefly own sqlite; leave the id off
// the ledger and abort this pass so a later startup resumes it.
throw error
}
summary.failedThreads += 1
appendHealLedgerRecord(paths.healLedgerPath, thread.threadId, 'failed')
appendHealLedgerRecord(paths, thread.threadId, 'failed')
}
}
function resolveHealWorkLimit(value: number | undefined, maximum: number): number {
if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) {
return maximum
}
return Math.min(Math.floor(value), maximum)
}
function buildNativeHealInvocation(
systemCodexHomePath: string,
timeoutMs: number