From 2f41d62d37276adf6b19371f79aaa2e65a99b924 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 15 Jul 2026 17:34:15 -0700 Subject: [PATCH] feat(codex): background sqlite index heal for backfilled sessions Codex's own state-DB metadata backfill is one-shot, so rollouts hardlinked in by Orca's session backfill never become visible to Codex's DB-driven surfaces. Extract the app-server stdio JSONL transport into codex-app-server-session (shared with the trust-grant client) and add a bounded, resumable background pass that drives Codex's lazy indexing via thread/read per backfilled session: recent-first, batched onto one short-lived server per batch with small concurrency, ledger + marker so steady-state startups are a no-op, stop-aware on quit, and capability-aware on CLIs without the app-server surface. --- config/tsconfig.cli.json | 1 + src/main/codex/codex-app-server-client.ts | 331 +++---------- src/main/codex/codex-app-server-session.ts | 264 ++++++++++ .../codex/codex-session-index-heal-state.ts | 197 ++++++++ .../codex/codex-session-index-heal.test.ts | 451 ++++++++++++++++++ src/main/codex/codex-session-index-heal.ts | 235 +++++++++ src/main/index.ts | 13 +- 7 files changed, 1215 insertions(+), 277 deletions(-) create mode 100644 src/main/codex/codex-app-server-session.ts create mode 100644 src/main/codex/codex-session-index-heal-state.ts create mode 100644 src/main/codex/codex-session-index-heal.test.ts create mode 100644 src/main/codex/codex-session-index-heal.ts diff --git a/config/tsconfig.cli.json b/config/tsconfig.cli.json index 0d278a1e854..dffdba0d872 100644 --- a/config/tsconfig.cli.json +++ b/config/tsconfig.cli.json @@ -15,6 +15,7 @@ "../src/main/codex/codex-app-server-client.ts", "../src/main/codex/codex-app-server-grant-bridge.ts", "../src/main/codex/codex-app-server-grant-envelope.ts", + "../src/main/codex/codex-app-server-session.ts", "../src/main/codex/codex-config-mirror.ts", "../src/main/codex/codex-config-path-reference-rewrite.ts", "../src/main/codex/codex-home-paths.ts", diff --git a/src/main/codex/codex-app-server-client.ts b/src/main/codex/codex-app-server-client.ts index 0273893fcc8..82b62b0e154 100644 --- a/src/main/codex/codex-app-server-client.ts +++ b/src/main/codex/codex-app-server-client.ts @@ -1,5 +1,6 @@ -import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process' +import { spawn } from 'node:child_process' import { normalizeHookTrustKeyForLookup } from './config-toml-trust' +import { runCodexAppServerSession, type CodexAppServerInvocation } from './codex-app-server-session' // Why: Codex gates hooks on a `trusted_hash` it computes from a private // canonical-JSON identity. Orca used to replicate that algorithm @@ -11,14 +12,12 @@ import { normalizeHookTrustKeyForLookup } from './config-toml-trust' // Codex as the only hash authority. See upstream codex-rs/tui/src/hooks_rpc.rs // and codex-rs/tui/src/startup_hooks_review.rs. -export type CodexAppServerInvocation = { - command: string - args: string[] - /** Overlay applied on top of the inherited environment (e.g. CODEX_HOME). */ - env?: Record - /** Whole-session deadline. The codex child is SIGKILLed when it lapses. */ - timeoutMs: number -} +export { + CodexAppServerTimeoutError, + CodexAppServerUnsupportedError, + isCodexAppServerUnsupportedError, + type CodexAppServerInvocation +} from './codex-app-server-session' export type CodexHookTrustGrantRequest = { invocation: CodexAppServerInvocation @@ -51,33 +50,6 @@ export type CodexHookTrustGrantSessionResult = } | { outcome: 'verify-failed'; reason: string } -/** Codex-side absence of the trust-grant RPC surface (old CLI without the - * app-server subcommand, or a server without hooks/list / config/batchWrite). - * This is the ONLY error class the capability cache marks unsupported. */ -export class CodexAppServerUnsupportedError extends Error { - constructor(message: string) { - super(message) - this.name = 'CodexAppServerUnsupportedError' - } -} - -export class CodexAppServerTimeoutError extends Error { - constructor(message: string) { - super(message) - this.name = 'CodexAppServerTimeoutError' - } -} - -export function isCodexAppServerUnsupportedError(error: unknown): boolean { - return error instanceof Error && error.name === 'CodexAppServerUnsupportedError' -} - -type JsonRpcResponse = { - id?: number - result?: unknown - error?: { code?: number; message?: string } -} - type CodexHookListing = { key: string command: string | null @@ -85,20 +57,6 @@ type CodexHookListing = { trustStatus: string } -const JSON_RPC_METHOD_NOT_FOUND = -32601 -const STDERR_TAIL_MAX_BYTES = 8192 -const STDOUT_LINE_MAX_BYTES = 1024 * 1024 - -function isMethodNotFoundError(error: { code?: number; message?: string }): boolean { - return error.code === JSON_RPC_METHOD_NOT_FOUND || /method not found/i.test(error.message ?? '') -} - -// Why: a CLI predating the app-server subcommand fails argv parsing before -// speaking any JSON-RPC; that shape is a capability signal, not a transient. -function stderrIndicatesMissingAppServer(stderrTail: string): boolean { - return /unrecognized subcommand|unexpected argument|invalid subcommand/i.test(stderrTail) -} - function collectHookListings(result: unknown): CodexHookListing[] { const data = result && typeof result === 'object' && Array.isArray((result as { data?: unknown }).data) @@ -137,239 +95,64 @@ function collectHookListings(result: unknown): CodexHookListing[] { * Runs one short-lived `codex app-server` session over stdio JSON-RPC (JSONL) * and grants trust for exactly the expected managed entries: * initialize → initialized → hooks/list → config/batchWrite → hooks/list. - * The child is reaped on every path; the session deadline SIGKILLs it. */ export async function runCodexHookTrustGrantSession( request: CodexHookTrustGrantRequest, spawnImpl: typeof spawn = spawn ): Promise { - const { invocation } = request - const child = spawnImpl(invocation.command, invocation.args, { - env: { ...process.env, ...invocation.env }, - stdio: ['pipe', 'pipe', 'pipe'], - windowsHide: true - }) as ChildProcessWithoutNullStreams + return runCodexAppServerSession( + request.invocation, + async (rpc) => { + const expectedKeys = new Set(request.expectedTrustKeys) + const matchManaged = (listing: CodexHookListing): boolean => + listing.command === request.managedCommand && + expectedKeys.has(normalizeHookTrustKeyForLookup(listing.key)) - let stderrTail = '' - let exited = false - let nextRequestId = 1 - let timedOut = false - const pending = new Map< - number, - { resolve: (r: JsonRpcResponse) => void; reject: (e: Error) => void } - >() - - const exitPromise = new Promise((resolve) => { - child.on('exit', () => { - exited = true - resolve() - }) - }) - // Why: 'error' fires instead of 'exit' when the spawn itself fails - // (ENOENT); surface it to every in-flight request or they wait forever. - let spawnError: Error | null = null - child.on('error', (error) => { - spawnError = error - exited = true - failPending(error) - }) - // Why: 'close' (not 'exit') guarantees the stderr tail is complete, so an - // early death classifies correctly as missing-subcommand vs transient. - child.on('close', () => { - failPending(buildEarlyExitError()) - }) - // Why: JSONL can contain non-ASCII hook paths. Stream decoding must retain a - // multibyte character split across pipe chunks or the response becomes invalid JSON. - child.stderr.setEncoding('utf8').on('data', (chunk: string) => { - stderrTail = (stderrTail + chunk).slice(-STDERR_TAIL_MAX_BYTES) - }) - // Why: a child can exit between the liveness check and stdin.write(); an - // EPIPE must reject the RPC instead of becoming an unhandled stream error. - child.stdin.on('error', (error) => { - failPending(error) - }) - - let stdoutBuffer = '' - child.stdout.setEncoding('utf8').on('data', (chunk: string) => { - stdoutBuffer += chunk - if (Buffer.byteLength(stdoutBuffer) > STDOUT_LINE_MAX_BYTES) { - child.kill('SIGKILL') - failPending(new Error('codex app-server emitted an oversized JSONL response')) - return - } - let newlineIndex - while ((newlineIndex = stdoutBuffer.indexOf('\n')) !== -1) { - const line = stdoutBuffer.slice(0, newlineIndex).trim() - stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1) - if (!line) { - continue + const listResult = await rpc.request('hooks/list', { cwds: [request.hooksListCwd] }) + const managedListings = collectHookListings(listResult).filter(matchManaged) + if (managedListings.length !== expectedKeys.size) { + return { + outcome: 'verify-failed', + reason: `hooks/list reported ${managedListings.length} of ${expectedKeys.size} expected managed entries` + } } - let message: JsonRpcResponse - try { - message = JSON.parse(line) as JsonRpcResponse - } catch { - continue + + const needingTrust = managedListings.filter((listing) => listing.trustStatus !== 'trusted') + if (needingTrust.length > 0) { + // Why: same wire shape as the Codex TUI "Trust all" flow — one upsert + // edit under hooks.state with each key's Codex-computed current hash. + const value: Record = {} + for (const listing of needingTrust) { + value[listing.key] = { trusted_hash: listing.currentHash } + } + await rpc.request('config/batchWrite', { + edits: [{ keyPath: 'hooks.state', value, mergeStrategy: 'upsert' }], + reloadUserConfig: true + }) } - if (typeof message.id === 'number' && pending.has(message.id)) { - const waiter = pending.get(message.id)! - pending.delete(message.id) - waiter.resolve(message) + + const verifyResult = await rpc.request('hooks/list', { cwds: [request.hooksListCwd] }) + const verifiedListings = collectHookListings(verifyResult).filter(matchManaged) + const untrusted = verifiedListings.filter((listing) => listing.trustStatus !== 'trusted') + if (verifiedListings.length !== expectedKeys.size || untrusted.length > 0) { + return { + outcome: 'verify-failed', + reason: + untrusted.length > 0 + ? `post-grant verify left ${untrusted.length} entries ${untrusted[0].trustStatus}` + : `post-grant verify reported ${verifiedListings.length} of ${expectedKeys.size} entries` + } } - } - }) - - function failPending(error: Error): void { - for (const waiter of pending.values()) { - waiter.reject(error) - } - pending.clear() - } - - const deadline = setTimeout(() => { - timedOut = true - child.kill('SIGKILL') - failPending( - new CodexAppServerTimeoutError( - `codex app-server session exceeded ${invocation.timeoutMs}ms (${invocation.command})` - ) - ) - }, invocation.timeoutMs) - - function sendLine(payload: Record): void { - child.stdin.write(`${JSON.stringify(payload)}\n`) - } - - async function requestRpc(method: string, params?: Record): Promise { - if (spawnError) { - throw spawnError - } - if (timedOut) { - throw new CodexAppServerTimeoutError('codex app-server session already timed out') - } - if (exited) { - throw buildEarlyExitError() - } - const id = nextRequestId++ - const response = await new Promise((resolve, reject) => { - pending.set(id, { resolve, reject }) - const payload: Record = { method, id } - if (params !== undefined) { - payload.params = params - } - try { - sendLine(payload) - } catch (error) { - pending.delete(id) - reject(error instanceof Error ? error : new Error(String(error))) - } - }) - if (response.error) { - if (isMethodNotFoundError(response.error)) { - throw new CodexAppServerUnsupportedError( - `codex app-server does not support ${method}: ${response.error.message ?? 'method not found'}` - ) - } - throw new Error( - `codex app-server ${method} failed: ${response.error.message ?? 'unknown error'}` - ) - } - return response.result - } - - function buildEarlyExitError(): Error { - if (stderrIndicatesMissingAppServer(stderrTail)) { - return new CodexAppServerUnsupportedError( - `codex CLI does not support the app-server subcommand: ${stderrTail.trim().slice(0, 400)}` - ) - } - return new Error( - `codex app-server exited before completing the session${stderrTail ? `: ${stderrTail.trim().slice(0, 400)}` : ''}` - ) - } - - try { - await requestRpc('initialize', { - clientInfo: { name: 'orca_desktop', title: 'Orca', version: '0.0.0' } - }) - sendLine({ method: 'initialized' }) - - const expectedKeys = new Set(request.expectedTrustKeys) - const matchManaged = (listing: CodexHookListing): boolean => - listing.command === request.managedCommand && - expectedKeys.has(normalizeHookTrustKeyForLookup(listing.key)) - - const listResult = await requestRpc('hooks/list', { cwds: [request.hooksListCwd] }) - const managedListings = collectHookListings(listResult).filter(matchManaged) - if (managedListings.length !== expectedKeys.size) { return { - outcome: 'verify-failed', - reason: `hooks/list reported ${managedListings.length} of ${expectedKeys.size} expected managed entries` + outcome: 'granted', + wroteTrust: needingTrust.length > 0, + entries: verifiedListings.map((listing) => ({ + key: listing.key, + normalizedKey: normalizeHookTrustKeyForLookup(listing.key), + trustedHash: listing.currentHash + })) } - } - - const needingTrust = managedListings.filter((listing) => listing.trustStatus !== 'trusted') - if (needingTrust.length > 0) { - // Why: same wire shape as the Codex TUI "Trust all" flow — one upsert - // edit under hooks.state with each key's Codex-computed current hash. - const value: Record = {} - for (const listing of needingTrust) { - value[listing.key] = { trusted_hash: listing.currentHash } - } - await requestRpc('config/batchWrite', { - edits: [{ keyPath: 'hooks.state', value, mergeStrategy: 'upsert' }], - reloadUserConfig: true - }) - } - - const verifyResult = await requestRpc('hooks/list', { cwds: [request.hooksListCwd] }) - const verifiedListings = collectHookListings(verifyResult).filter(matchManaged) - const untrusted = verifiedListings.filter((listing) => listing.trustStatus !== 'trusted') - if (verifiedListings.length !== expectedKeys.size || untrusted.length > 0) { - return { - outcome: 'verify-failed', - reason: - untrusted.length > 0 - ? `post-grant verify left ${untrusted.length} entries ${untrusted[0].trustStatus}` - : `post-grant verify reported ${verifiedListings.length} of ${expectedKeys.size} entries` - } - } - return { - outcome: 'granted', - wroteTrust: needingTrust.length > 0, - entries: verifiedListings.map((listing) => ({ - key: listing.key, - normalizedKey: normalizeHookTrustKeyForLookup(listing.key), - trustedHash: listing.currentHash - })) - } - } catch (error) { - if ( - error instanceof Error && - !(error instanceof CodexAppServerUnsupportedError) && - !(error instanceof CodexAppServerTimeoutError) && - stderrIndicatesMissingAppServer(stderrTail) - ) { - throw new CodexAppServerUnsupportedError( - `codex CLI does not support the app-server subcommand: ${stderrTail.trim().slice(0, 400)}` - ) - } - throw error - } finally { - try { - child.stdin.end() - } catch { - // stdin may already be destroyed after a kill; reaping below still runs. - } - if (!exited) { - // Why: the server exits promptly on stdin EOF; the grace period only - // bounds a wedged child before the guaranteed SIGKILL reap. - const grace = new Promise((resolve) => setTimeout(resolve, 1500)) - await Promise.race([exitPromise, grace]) - if (!exited) { - child.kill('SIGKILL') - await Promise.race([exitPromise, new Promise((resolve) => setTimeout(resolve, 1000))]) - } - } - clearTimeout(deadline) - } + }, + spawnImpl + ) } diff --git a/src/main/codex/codex-app-server-session.ts b/src/main/codex/codex-app-server-session.ts new file mode 100644 index 00000000000..ac074006343 --- /dev/null +++ b/src/main/codex/codex-app-server-session.ts @@ -0,0 +1,264 @@ +import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process' + +// Why: `codex app-server` is Orca's sanctioned RPC surface into Codex-owned +// state (hook trust hashes, the sqlite thread index). This module owns the +// stdio JSONL transport — spawn, handshake, framing, deadline, reap — so every +// RPC consumer (trust grant, session index heal) shares one hardened lifecycle. + +export type CodexAppServerInvocation = { + command: string + args: string[] + /** Overlay applied on top of the inherited environment (e.g. CODEX_HOME). */ + env?: Record + /** Whole-session deadline. The codex child is SIGKILLed when it lapses. */ + timeoutMs: number +} + +/** Codex-side absence of the requested app-server RPC surface (old CLI without + * the app-server subcommand, or a server without the called methods). + * This is the ONLY error class capability caches mark unsupported. */ +export class CodexAppServerUnsupportedError extends Error { + constructor(message: string) { + super(message) + this.name = 'CodexAppServerUnsupportedError' + } +} + +export class CodexAppServerTimeoutError extends Error { + constructor(message: string) { + super(message) + this.name = 'CodexAppServerTimeoutError' + } +} + +export function isCodexAppServerUnsupportedError(error: unknown): boolean { + return error instanceof Error && error.name === 'CodexAppServerUnsupportedError' +} + +type JsonRpcResponse = { + id?: number + result?: unknown + error?: { code?: number; message?: string } +} + +export type CodexAppServerRpc = { + request: (method: string, params?: Record) => Promise + notify: (method: string, params?: Record) => void +} + +const JSON_RPC_METHOD_NOT_FOUND = -32601 +const STDERR_TAIL_MAX_BYTES = 8192 +const STDOUT_LINE_MAX_BYTES = 1024 * 1024 + +function isMethodNotFoundError(error: { code?: number; message?: string }): boolean { + return error.code === JSON_RPC_METHOD_NOT_FOUND || /method not found/i.test(error.message ?? '') +} + +// Why: a CLI predating the app-server subcommand fails argv parsing before +// speaking any JSON-RPC; that shape is a capability signal, not a transient. +function stderrIndicatesMissingAppServer(stderrTail: string): boolean { + return /unrecognized subcommand|unexpected argument|invalid subcommand/i.test(stderrTail) +} + +/** + * Runs one short-lived `codex app-server` session over stdio JSON-RPC (JSONL): + * spawn → initialize → initialized → body(rpc) → EOF/reap. The child is reaped + * on every path; the session deadline SIGKILLs it. + */ +export async function runCodexAppServerSession( + invocation: CodexAppServerInvocation, + body: (rpc: CodexAppServerRpc) => Promise, + spawnImpl: typeof spawn = spawn +): Promise { + const child = spawnImpl(invocation.command, invocation.args, { + env: { ...process.env, ...invocation.env }, + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true + }) as ChildProcessWithoutNullStreams + + let stderrTail = '' + let exited = false + let nextRequestId = 1 + let timedOut = false + const pending = new Map< + number, + { resolve: (r: JsonRpcResponse) => void; reject: (e: Error) => void } + >() + + const exitPromise = new Promise((resolve) => { + child.on('exit', () => { + exited = true + resolve() + }) + }) + // Why: 'error' fires instead of 'exit' when the spawn itself fails + // (ENOENT); surface it to every in-flight request or they wait forever. + let spawnError: Error | null = null + child.on('error', (error) => { + spawnError = error + exited = true + failPending(error) + }) + // Why: 'close' (not 'exit') guarantees the stderr tail is complete, so an + // early death classifies correctly as missing-subcommand vs transient. + child.on('close', () => { + failPending(buildEarlyExitError()) + }) + // Why: JSONL can contain non-ASCII hook paths. Stream decoding must retain a + // multibyte character split across pipe chunks or the response becomes invalid JSON. + child.stderr.setEncoding('utf8').on('data', (chunk: string) => { + stderrTail = (stderrTail + chunk).slice(-STDERR_TAIL_MAX_BYTES) + }) + // Why: a child can exit between the liveness check and stdin.write(); an + // EPIPE must reject the RPC instead of becoming an unhandled stream error. + child.stdin.on('error', (error) => { + failPending(error) + }) + + let stdoutBuffer = '' + child.stdout.setEncoding('utf8').on('data', (chunk: string) => { + stdoutBuffer += chunk + if (Buffer.byteLength(stdoutBuffer) > STDOUT_LINE_MAX_BYTES) { + child.kill('SIGKILL') + failPending(new Error('codex app-server emitted an oversized JSONL response')) + return + } + let newlineIndex + while ((newlineIndex = stdoutBuffer.indexOf('\n')) !== -1) { + const line = stdoutBuffer.slice(0, newlineIndex).trim() + stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1) + if (!line) { + continue + } + let message: JsonRpcResponse + try { + message = JSON.parse(line) as JsonRpcResponse + } catch { + continue + } + if (typeof message.id === 'number' && pending.has(message.id)) { + const waiter = pending.get(message.id)! + pending.delete(message.id) + waiter.resolve(message) + } + } + }) + + function failPending(error: Error): void { + for (const waiter of pending.values()) { + waiter.reject(error) + } + pending.clear() + } + + const deadline = setTimeout(() => { + timedOut = true + child.kill('SIGKILL') + failPending( + new CodexAppServerTimeoutError( + `codex app-server session exceeded ${invocation.timeoutMs}ms (${invocation.command})` + ) + ) + }, invocation.timeoutMs) + + function sendLine(payload: Record): void { + child.stdin.write(`${JSON.stringify(payload)}\n`) + } + + function notify(method: string, params?: Record): void { + const payload: Record = { method } + if (params !== undefined) { + payload.params = params + } + try { + sendLine(payload) + } catch { + // Notifications are fire-and-forget; a dead child fails the next request. + } + } + + async function requestRpc(method: string, params?: Record): Promise { + if (spawnError) { + throw spawnError + } + if (timedOut) { + throw new CodexAppServerTimeoutError('codex app-server session already timed out') + } + if (exited) { + throw buildEarlyExitError() + } + const id = nextRequestId++ + const response = await new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }) + const payload: Record = { method, id } + if (params !== undefined) { + payload.params = params + } + try { + sendLine(payload) + } catch (error) { + pending.delete(id) + reject(error instanceof Error ? error : new Error(String(error))) + } + }) + if (response.error) { + if (isMethodNotFoundError(response.error)) { + throw new CodexAppServerUnsupportedError( + `codex app-server does not support ${method}: ${response.error.message ?? 'method not found'}` + ) + } + throw new Error( + `codex app-server ${method} failed: ${response.error.message ?? 'unknown error'}` + ) + } + return response.result + } + + function buildEarlyExitError(): Error { + if (stderrIndicatesMissingAppServer(stderrTail)) { + return new CodexAppServerUnsupportedError( + `codex CLI does not support the app-server subcommand: ${stderrTail.trim().slice(0, 400)}` + ) + } + return new Error( + `codex app-server exited before completing the session${stderrTail ? `: ${stderrTail.trim().slice(0, 400)}` : ''}` + ) + } + + try { + await requestRpc('initialize', { + clientInfo: { name: 'orca_desktop', title: 'Orca', version: '0.0.0' } + }) + notify('initialized') + return await body({ request: requestRpc, notify }) + } catch (error) { + if ( + error instanceof Error && + !(error instanceof CodexAppServerUnsupportedError) && + !(error instanceof CodexAppServerTimeoutError) && + stderrIndicatesMissingAppServer(stderrTail) + ) { + throw new CodexAppServerUnsupportedError( + `codex CLI does not support the app-server subcommand: ${stderrTail.trim().slice(0, 400)}` + ) + } + throw error + } finally { + try { + child.stdin.end() + } catch { + // stdin may already be destroyed after a kill; reaping below still runs. + } + if (!exited) { + // Why: the server exits promptly on stdin EOF; the grace period only + // bounds a wedged child before the guaranteed SIGKILL reap. + const grace = new Promise((resolve) => setTimeout(resolve, 1500)) + await Promise.race([exitPromise, grace]) + if (!exited) { + child.kill('SIGKILL') + await Promise.race([exitPromise, new Promise((resolve) => setTimeout(resolve, 1000))]) + } + } + clearTimeout(deadline) + } +} diff --git a/src/main/codex/codex-session-index-heal-state.ts b/src/main/codex/codex-session-index-heal-state.ts new file mode 100644 index 00000000000..2db9a679705 --- /dev/null +++ b/src/main/codex/codex-session-index-heal-state.ts @@ -0,0 +1,197 @@ +import { appendFileSync, mkdirSync, readFileSync, statSync } from 'node:fs' +import { dirname } from 'node:path' +import { writeFileAtomically } from '../codex-accounts/fs-utils' + +// State files for the session index heal: which backfilled rollouts exist +// (the backfill audit ledger), which thread ids this pass already processed +// (the heal ledger), and the completion marker that makes steady-state +// startups a two-stat no-op. + +// 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 + +// Why: an unsupported CLI stays unsupported until upgraded; re-probing once a +// day is enough to notice an upgrade without a per-startup spawn. +const HEAL_UNSUPPORTED_RETRY_INTERVAL_MS = 24 * 60 * 60 * 1000 + +const CODEX_ROLLOUT_THREAD_ID_PATTERN = + /^rollout-(.+)-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/i + +export type CodexSessionIndexHealPaths = { + auditLogPath: string + systemSessionsRoot: string + healLedgerPath: string + healMarkerPath: string +} + +export type HealLedgerOutcome = 'healed' | 'missing' | 'failed' + +export type PendingHealThread = { + threadId: string + /** Timestamp segment of the rollout file name; lexicographic recency order. */ + rolloutStamp: string +} + +export type HealMarkerSummary = { + healedThreads: number + missingThreads: number + failedThreads: number +} + +/** + * Diffs the backfill audit ledger against the heal ledger: every hardlinked or + * 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 pendingByThreadId = new Map() + for (const line of readJsonlLines(paths.auditLogPath)) { + if ((line.action !== 'hardlink' && line.action !== 'copy') || typeof line.target !== 'string') { + continue + } + const match = CODEX_ROLLOUT_THREAD_ID_PATTERN.exec(lastPathSegment(line.target)) + if (!match) { + continue + } + const threadId = match[2].toLowerCase() + if (processedThreadIds.has(threadId)) { + continue + } + pendingByThreadId.set(threadId, { threadId, rolloutStamp: match[1] }) + } + return [...pendingByThreadId.values()].sort((left, right) => + left.rolloutStamp < right.rolloutStamp ? 1 : left.rolloutStamp > right.rolloutStamp ? -1 : 0 + ) +} + +function lastPathSegment(filePath: string): string { + return filePath.split(/[\\/]/).at(-1) ?? '' +} + +function readProcessedHealThreadIds(healLedgerPath: string): Set { + const processed = new Set() + for (const line of readJsonlLines(healLedgerPath)) { + if (line.v === CODEX_SESSION_INDEX_HEAL_VERSION && typeof line.threadId === 'string') { + processed.add(line.threadId.toLowerCase()) + } + } + return processed +} + +export function appendHealLedgerRecord( + healLedgerPath: string, + threadId: string, + outcome: HealLedgerOutcome +): void { + try { + mkdirSync(dirname(healLedgerPath), { recursive: true }) + appendFileSync( + healLedgerPath, + `${JSON.stringify({ + v: CODEX_SESSION_INDEX_HEAL_VERSION, + threadId, + outcome, + at: new Date().toISOString() + })}\n` + ) + } catch (error) { + // Why: losing a ledger line only costs one redundant thread/read on the + // next pass; it must not fail the heal. + console.warn('[codex-session-index-heal] Failed to append heal ledger record:', error) + } +} + +function readJsonlLines(filePath: string): Record[] { + let contents: string + try { + contents = readFileSync(filePath, 'utf-8') + } catch { + return [] + } + const lines: Record[] = [] + for (const raw of contents.split('\n')) { + if (!raw.trim()) { + continue + } + try { + const parsed: unknown = JSON.parse(raw) + if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) { + lines.push(parsed as Record) + } + } catch { + // Skip torn/corrupt lines; both ledgers are append-only diagnostics. + } + } + return lines +} + +export function readAuditLogSize(auditLogPath: string): number { + try { + return statSync(auditLogPath).size + } catch { + return 0 + } +} + +export function isHealMarkerCurrent( + paths: CodexSessionIndexHealPaths, + auditBytes: number +): boolean { + try { + const parsed: unknown = JSON.parse(readFileSync(paths.healMarkerPath, 'utf-8')) + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return false + } + const marker = parsed as { + version?: unknown + systemSessionsRoot?: unknown + auditBytes?: unknown + unsupportedAt?: unknown + } + if ( + marker.version !== CODEX_SESSION_INDEX_HEAL_VERSION || + marker.systemSessionsRoot !== paths.systemSessionsRoot + ) { + return false + } + if (typeof marker.unsupportedAt === 'number') { + return Date.now() - marker.unsupportedAt < HEAL_UNSUPPORTED_RETRY_INTERVAL_MS + } + // Why: the audit ledger is append-only, so an unchanged byte size means no + // new backfilled sessions since this marker was written. + return marker.auditBytes === auditBytes + } catch { + return false + } +} + +export function writeHealMarker( + paths: CodexSessionIndexHealPaths, + auditBytes: number, + summary: HealMarkerSummary, + unsupportedAt?: number +): void { + try { + mkdirSync(dirname(paths.healMarkerPath), { recursive: true }) + writeFileAtomically( + paths.healMarkerPath, + `${JSON.stringify( + { + version: CODEX_SESSION_INDEX_HEAL_VERSION, + systemSessionsRoot: paths.systemSessionsRoot, + auditBytes, + healedThreads: summary.healedThreads, + missingThreads: summary.missingThreads, + failedThreads: summary.failedThreads, + ...(unsupportedAt === undefined ? {} : { unsupportedAt }), + completedAt: Date.now() + }, + null, + 2 + )}\n` + ) + } catch (error) { + console.warn('[codex-session-index-heal] Failed to write heal marker:', error) + } +} diff --git a/src/main/codex/codex-session-index-heal.test.ts b/src/main/codex/codex-session-index-heal.test.ts new file mode 100644 index 00000000000..a8f478a1be8 --- /dev/null +++ b/src/main/codex/codex-session-index-heal.test.ts @@ -0,0 +1,451 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { + appendFileSync, + mkdtempSync, + mkdirSync, + readFileSync, + rmSync, + writeFileSync +} from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import type { CodexAppServerInvocation } from './codex-app-server-session' +import { + runCodexSessionIndexHeal, + type CodexSessionIndexHealPaths +} from './codex-session-index-heal' + +// Stub codex app-server speaking the JSONL protocol for the heal pass: +// initialize → initialized → thread/read×N. Scenario-driven via STUB_CONFIG; +// every thread/read is appended to readLogFile so tests can assert order, +// batching (one spawn appends a server-start marker), and skip behavior. +const STUB_SERVER_SOURCE = ` +const fs = require('node:fs') +const config = JSON.parse(process.env.STUB_CONFIG) +fs.appendFileSync(config.readLogFile, JSON.stringify({ serverStart: true }) + '\\n') +let buffer = '' +let inFlight = 0 +let maxInFlight = 0 +function send(message) { + process.stdout.write(JSON.stringify(message) + '\\n') +} +if (config.scenario === 'no-subcommand') { + process.stderr.write("error: unrecognized subcommand 'app-server'\\n") + process.exit(2) +} +process.stdin.setEncoding('utf8') +process.stdin.on('data', (chunk) => { + buffer += chunk + let index + while ((index = buffer.indexOf('\\n')) !== -1) { + const line = buffer.slice(0, index).trim() + buffer = buffer.slice(index + 1) + if (!line) continue + const message = JSON.parse(line) + if (message.method === 'initialize') { + send({ id: message.id, result: { userAgent: 'stub/0.0.0', codexHome: process.env.CODEX_HOME } }) + continue + } + if (message.method === 'initialized') continue + if (message.method === 'thread/read') { + const threadId = message.params.threadId + if (config.scenario === 'unknown-method') { + send({ id: message.id, error: { code: -32601, message: 'Method not found' } }) + continue + } + inFlight += 1 + maxInFlight = Math.max(maxInFlight, inFlight) + setTimeout(() => { + inFlight -= 1 + fs.appendFileSync(config.readLogFile, JSON.stringify({ threadId, maxInFlight }) + '\\n') + if ((config.missingThreadIds || []).includes(threadId)) { + send({ id: message.id, error: { code: -32600, message: 'no rollout found for thread id ' + threadId } }) + return + } + if ((config.failingThreadIds || []).includes(threadId)) { + send({ id: message.id, error: { code: -32600, message: 'failed to parse rollout' } }) + return + } + if (config.scenario === 'die-mid-batch' && threadId === config.dieOnThreadId) { + process.exit(7) + } + send({ id: message.id, result: { thread: { id: threadId } } }) + }, 5) + continue + } + } +}) +process.stdin.on('end', () => process.exit(0)) +` + +let tempRoots: string[] = [] + +afterEach(() => { + for (const root of tempRoots) { + rmSync(root, { recursive: true, force: true }) + } + tempRoots = [] +}) + +function threadId(suffix: string): string { + return `019f0000-1111-7222-8333-${suffix.padStart(12, '0')}` +} + +function rolloutTarget(sessionsRoot: string, stamp: string, id: string): string { + return join(sessionsRoot, '2026', '07', '01', `rollout-${stamp}-${id}.jsonl`) +} + +function createHealRig(options: { + scenario?: string + auditedThreads?: { stamp: string; id: string; action?: string }[] + missingThreadIds?: string[] + failingThreadIds?: string[] + dieOnThreadId?: string +}): { + paths: CodexSessionIndexHealPaths + readLogFile: string + buildInvocation: (systemCodexHomePath: string, timeoutMs: number) => CodexAppServerInvocation + readLog: () => { serverStarts: number; threadIds: string[]; maxInFlight: number } +} { + const root = mkdtempSync(join(tmpdir(), 'orca-codex-heal-')) + tempRoots.push(root) + const systemSessionsRoot = join(root, 'real-home', 'sessions') + const stateDir = join(root, 'state') + mkdirSync(stateDir, { recursive: true }) + const paths: CodexSessionIndexHealPaths = { + auditLogPath: join(stateDir, 'audit.jsonl'), + systemSessionsRoot, + healLedgerPath: join(stateDir, 'index-heal-ledger.jsonl'), + healMarkerPath: join(stateDir, 'index-heal-complete.json') + } + for (const audited of options.auditedThreads ?? []) { + appendFileSync( + paths.auditLogPath, + `${JSON.stringify({ + at: '2026-07-01T00:00:00.000Z', + action: audited.action ?? 'hardlink', + source: '/managed/sessions/x.jsonl', + target: rolloutTarget(systemSessionsRoot, audited.stamp, audited.id) + })}\n` + ) + } + const stubPath = join(root, 'stub-app-server.cjs') + writeFileSync(stubPath, STUB_SERVER_SOURCE) + const readLogFile = join(root, 'reads.jsonl') + writeFileSync(readLogFile, '') + return { + paths, + readLogFile, + buildInvocation: (_systemCodexHomePath, timeoutMs) => ({ + command: process.execPath, + args: [stubPath], + env: { + STUB_CONFIG: JSON.stringify({ + scenario: options.scenario ?? 'ok', + readLogFile, + missingThreadIds: options.missingThreadIds ?? [], + failingThreadIds: options.failingThreadIds ?? [], + dieOnThreadId: options.dieOnThreadId + }) + }, + timeoutMs + }), + readLog: () => { + const lines = readFileSync(readLogFile, 'utf-8') + .split('\n') + .filter(Boolean) + .map( + (line) => + JSON.parse(line) as { serverStart?: boolean; threadId?: string; maxInFlight?: number } + ) + return { + serverStarts: lines.filter((line) => line.serverStart).length, + threadIds: lines.map((line) => line.threadId).filter((id): id is string => Boolean(id)), + maxInFlight: Math.max(0, ...lines.map((line) => line.maxInFlight ?? 0)) + } + } + } +} + +function readLedgerOutcomes(paths: CodexSessionIndexHealPaths): Record { + let contents = '' + try { + contents = readFileSync(paths.healLedgerPath, 'utf-8') + } catch { + return {} + } + const outcomes: Record = {} + for (const line of contents.split('\n').filter(Boolean)) { + const record = JSON.parse(line) as { threadId: string; outcome: string } + outcomes[record.threadId] = record.outcome + } + return outcomes +} + +describe('runCodexSessionIndexHeal', () => { + it('reads every backfilled session recent-first and completes with a marker', async () => { + const rig = createHealRig({ + auditedThreads: [ + { stamp: '2026-07-01T10-00-00', id: threadId('1') }, + { stamp: '2026-07-03T10-00-00', id: threadId('3'), action: 'copy' }, + { stamp: '2026-07-02T10-00-00', id: threadId('2') } + ] + }) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + + expect(summary).toMatchObject({ + outcome: 'completed', + pendingThreads: 3, + healedThreads: 3, + missingThreads: 0, + failedThreads: 0 + }) + expect(rig.readLog().threadIds).toEqual([threadId('3'), threadId('2'), threadId('1')]) + expect(readLedgerOutcomes(rig.paths)).toEqual({ + [threadId('1')]: 'healed', + [threadId('2')]: 'healed', + [threadId('3')]: 'healed' + }) + const marker = JSON.parse(readFileSync(rig.paths.healMarkerPath, 'utf-8')) as { + systemSessionsRoot: string + healedThreads: number + } + expect(marker.systemSessionsRoot).toBe(rig.paths.systemSessionsRoot) + expect(marker.healedThreads).toBe(3) + }) + + it('is a no-op when the marker matches the audit ledger size', async () => { + const rig = createHealRig({ + auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }] + }) + const first = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(first.outcome).toBe('completed') + const second = 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(rig.readLog().serverStarts).toBe(1) + }) + + it('resumes only unprocessed sessions when the audit ledger grows', async () => { + const rig = createHealRig({ + auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }] + }) + await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + appendFileSync( + rig.paths.auditLogPath, + `${JSON.stringify({ + action: 'hardlink', + target: rolloutTarget(rig.paths.systemSessionsRoot, '2026-07-04T10-00-00', threadId('4')) + })}\n` + ) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + + expect(summary).toMatchObject({ outcome: 'completed', pendingThreads: 1, healedThreads: 1 }) + expect(rig.readLog().threadIds).toEqual([threadId('1'), threadId('4')]) + }) + + it('records missing and failed sessions without retrying them next run', async () => { + const rig = createHealRig({ + auditedThreads: [ + { stamp: '2026-07-01T10-00-00', id: threadId('1') }, + { stamp: '2026-07-02T10-00-00', id: threadId('2') }, + { stamp: '2026-07-03T10-00-00', id: threadId('3') } + ], + missingThreadIds: [threadId('2')], + failingThreadIds: [threadId('1')] + }) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + + expect(summary).toMatchObject({ + outcome: 'completed', + healedThreads: 1, + missingThreads: 1, + failedThreads: 1 + }) + expect(readLedgerOutcomes(rig.paths)).toEqual({ + [threadId('1')]: 'failed', + [threadId('2')]: 'missing', + [threadId('3')]: 'healed' + }) + + const again = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(again.outcome).toBe('up-to-date') + }) + + it('splits work into batches with one server session each and bounded concurrency', async () => { + const rig = createHealRig({ + auditedThreads: Array.from({ length: 5 }, (_, index) => ({ + stamp: `2026-07-0${index + 1}T10-00-00`, + id: threadId(String(index + 1)) + })) + }) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + readsPerServerSession: 2, + readConcurrency: 2, + interBatchDelayMs: 0 + }) + + expect(summary).toMatchObject({ outcome: 'completed', healedThreads: 5 }) + const log = rig.readLog() + expect(log.serverStarts).toBe(3) + expect(log.maxInFlight).toBeLessThanOrEqual(2) + }) + + it('stops promptly when shouldStop flips and resumes on the next pass', async () => { + const rig = createHealRig({ + auditedThreads: Array.from({ length: 4 }, (_, index) => ({ + stamp: `2026-07-0${index + 1}T10-00-00`, + id: threadId(String(index + 1)) + })) + }) + let reads = 0 + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + readsPerServerSession: 1, + interBatchDelayMs: 0, + shouldStop: () => reads++ >= 2 + }) + expect(summary.outcome).toBe('stopped') + expect(summary.healedThreads).toBeLessThan(4) + + const resumed = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(resumed.outcome).toBe('completed') + expect(resumed.healedThreads + summary.healedThreads).toBe(4) + }) + + it('marks the pass unsupported without ledger writes when thread/read is unavailable', async () => { + const rig = createHealRig({ + scenario: 'unknown-method', + auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }] + }) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(summary.outcome).toBe('unsupported') + expect(summary.healedThreads).toBe(0) + expect(readLedgerOutcomes(rig.paths)).toEqual({}) + + // Within the retry interval the unsupported marker suppresses re-probing. + const again = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(again.outcome).toBe('up-to-date') + expect(rig.readLog().serverStarts).toBe(1) + }) + + it('marks the pass unsupported when the CLI lacks the app-server subcommand', async () => { + const rig = createHealRig({ + scenario: 'no-subcommand', + auditedThreads: [{ stamp: '2026-07-01T10-00-00', id: threadId('1') }] + }) + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(summary.outcome).toBe('unsupported') + }) + + it('aborts without recording when the server dies mid-batch, then retries next pass', async () => { + const rig = createHealRig({ + scenario: 'die-mid-batch', + auditedThreads: [ + { stamp: '2026-07-02T10-00-00', id: threadId('2') }, + { stamp: '2026-07-01T10-00-00', id: threadId('1') } + ], + dieOnThreadId: threadId('2') + }) + + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + readConcurrency: 1, + interBatchDelayMs: 0 + }) + expect(summary.outcome).toBe('aborted') + expect(readLedgerOutcomes(rig.paths)[threadId('1')]).toBeUndefined() + + 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: [] + }) + } + } + }, + interBatchDelayMs: 0 + }) + expect(retried.outcome).toBe('completed') + expect(retried.healedThreads).toBe(2) + }) + + it('completes immediately with no server spawn when there is nothing to heal', async () => { + const rig = createHealRig({ auditedThreads: [] }) + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(summary).toMatchObject({ outcome: 'completed', pendingThreads: 0 }) + expect(rig.readLog().serverStarts).toBe(0) + }) + + it('ignores audit records outside the backfill link/copy actions', async () => { + const rig = createHealRig({ auditedThreads: [] }) + appendFileSync( + rig.paths.auditLogPath, + `${[ + JSON.stringify({ action: 'run-summary', scannedFiles: 3 }), + JSON.stringify({ action: 'scan-failed', source: '/managed/sessions/2026' }), + JSON.stringify({ + action: 'failed', + target: rolloutTarget(rig.paths.systemSessionsRoot, '2026-07-01T10-00-00', threadId('9')) + }), + 'not-json', + '' + ].join('\n')}\n` + ) + const summary = await runCodexSessionIndexHeal(rig.paths, { + buildInvocation: rig.buildInvocation, + interBatchDelayMs: 0 + }) + expect(summary).toMatchObject({ outcome: 'completed', pendingThreads: 0 }) + expect(rig.readLog().serverStarts).toBe(0) + }) +}) diff --git a/src/main/codex/codex-session-index-heal.ts b/src/main/codex/codex-session-index-heal.ts new file mode 100644 index 00000000000..d31c9bba17a --- /dev/null +++ b/src/main/codex/codex-session-index-heal.ts @@ -0,0 +1,235 @@ +import { dirname, join } from 'node:path' +import { resolveCodexCommand } from '../codex-cli/command' +import { getSpawnArgsForWindows } from '../win32-utils' +import { getCodexSessionBackfillStateDirPath } from './codex-home-paths' +import { resolveCodexSessionBackfillPaths } from './codex-session-backfill' +import { + appendHealLedgerRecord, + collectPendingHealThreads, + isHealMarkerCurrent, + readAuditLogSize, + writeHealMarker, + type CodexSessionIndexHealPaths, + type PendingHealThread +} from './codex-session-index-heal-state' +import { + isCodexAppServerUnsupportedError, + runCodexAppServerSession, + type CodexAppServerInvocation +} from './codex-app-server-session' + +export type { CodexSessionIndexHealPaths } from './codex-session-index-heal-state' + +// Why: Codex's own sqlite metadata backfill is one-shot (backfill_state is +// stamped `complete` on first app-server startup), so rollouts that Orca's +// session backfill hardlinks in later never reach the state DB on their own. +// `thread/read` is Codex's sanctioned lazy-indexing path: it parses the +// rollout and upserts the thread row, making backfilled sessions visible to +// Codex's DB-driven surfaces. Orca never writes Codex's sqlite schema itself. + +// Why: one server session per batch bounds child memory and keeps a wedged +// server from stalling the whole pass; small in-session concurrency keeps the +// disk/CPU cost background-grade instead of a thundering read storm. +const HEAL_READS_PER_SERVER_SESSION = 50 +const HEAL_READ_CONCURRENCY = 2 +const HEAL_INTER_BATCH_DELAY_MS = 500 +const HEAL_BATCH_TIMEOUT_BASE_MS = 15_000 +const HEAL_BATCH_TIMEOUT_PER_READ_MS = 2_000 + +export type CodexSessionIndexHealSummary = { + outcome: 'completed' | 'stopped' | 'unsupported' | 'aborted' | 'up-to-date' + pendingThreads: number + healedThreads: number + missingThreads: number + failedThreads: number +} + +export type CodexSessionIndexHealOptions = { + /** Polled between reads and batches; true stops promptly, progress is kept. */ + shouldStop?: () => boolean + buildInvocation?: (systemCodexHomePath: string, timeoutMs: number) => CodexAppServerInvocation + readsPerServerSession?: number + readConcurrency?: number + interBatchDelayMs?: number +} + +let backgroundHealTask: Promise | null = null + +export function resolveCodexSessionIndexHealPaths( + systemCodexHomePathOverride?: string +): CodexSessionIndexHealPaths { + const backfillPaths = resolveCodexSessionBackfillPaths(systemCodexHomePathOverride) + const stateDir = getCodexSessionBackfillStateDirPath() + return { + auditLogPath: backfillPaths.auditLogPath, + systemSessionsRoot: backfillPaths.systemSessionsRoot, + healLedgerPath: join(stateDir, 'index-heal-ledger.jsonl'), + healMarkerPath: join(stateDir, 'index-heal-complete.json') + } +} + +/** + * Starts a single background index-heal pass for backfilled Codex sessions. + * + * Concurrent callers share the in-flight task; an up-to-date marker resolves + * without reading the audit ledger or spawning any app-server. + */ +export function startCodexSessionIndexHealInBackground( + options: CodexSessionIndexHealOptions = {}, + systemCodexHomePathOverride?: string +): Promise { + if (backgroundHealTask) { + return backgroundHealTask + } + const task = runCodexSessionIndexHeal( + resolveCodexSessionIndexHealPaths(systemCodexHomePathOverride), + options + ).catch((error: unknown) => { + console.warn('[codex-session-index-heal] Background index heal failed:', error) + return null + }) + backgroundHealTask = task + void task.finally(() => { + if (backgroundHealTask === task) { + backgroundHealTask = null + } + }) + return task +} + +/** + * Drives Codex's lazy thread indexing (`thread/read`) for every backfilled + * session recorded in the backfill audit ledger that this pass has not + * processed yet, most recent sessions first. + */ +export async function runCodexSessionIndexHeal( + paths: CodexSessionIndexHealPaths, + options: CodexSessionIndexHealOptions = {} +): Promise { + const auditBytes = readAuditLogSize(paths.auditLogPath) + if (isHealMarkerCurrent(paths, auditBytes)) { + return { + outcome: 'up-to-date', + pendingThreads: 0, + healedThreads: 0, + missingThreads: 0, + failedThreads: 0 + } + } + + const pending = collectPendingHealThreads(paths) + const summary: CodexSessionIndexHealSummary = { + outcome: 'completed', + pendingThreads: pending.length, + healedThreads: 0, + missingThreads: 0, + failedThreads: 0 + } + if (pending.length === 0) { + writeHealMarker(paths, auditBytes, summary) + return summary + } + + 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 interBatchDelayMs = options.interBatchDelayMs ?? HEAL_INTER_BATCH_DELAY_MS + const shouldStop = options.shouldStop ?? ((): boolean => false) + + for (let offset = 0; offset < pending.length; offset += readsPerServerSession) { + if (shouldStop()) { + summary.outcome = 'stopped' + return summary + } + if (offset > 0 && interBatchDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, interBatchDelayMs)) + } + const batch = pending.slice(offset, offset + readsPerServerSession) + const timeoutMs = HEAL_BATCH_TIMEOUT_BASE_MS + HEAL_BATCH_TIMEOUT_PER_READ_MS * batch.length + try { + await runCodexAppServerSession( + buildInvocation(systemCodexHomePath, timeoutMs), + async (rpc) => { + let nextIndex = 0 + const worker = async (): Promise => { + while (nextIndex < batch.length && !shouldStop()) { + const thread = batch[nextIndex] + nextIndex += 1 + await healOneThread(rpc, thread, paths, summary) + } + } + await Promise.all(Array.from({ length: Math.max(1, readConcurrency) }, () => worker())) + } + ) + } catch (error) { + if (isCodexAppServerUnsupportedError(error)) { + // Why: no retry churn on old CLIs — remember unsupported and re-probe + // after the retry interval or a version bump; nothing is marked healed. + writeHealMarker(paths, auditBytes, summary, Date.now()) + summary.outcome = 'unsupported' + return summary + } + // Transport failure (timeout, early exit, spawn error): unprocessed ids + // were never appended to the ledger, so the next pass resumes them. + console.warn('[codex-session-index-heal] Heal batch aborted:', error) + summary.outcome = 'aborted' + return summary + } + } + + if (shouldStop()) { + summary.outcome = 'stopped' + return summary + } + writeHealMarker(paths, auditBytes, summary) + return summary +} + +async function healOneThread( + rpc: { request: (method: string, params?: Record) => Promise }, + thread: PendingHealThread, + paths: CodexSessionIndexHealPaths, + summary: CodexSessionIndexHealSummary +): Promise { + try { + await rpc.request('thread/read', { threadId: thread.threadId }) + summary.healedThreads += 1 + appendHealLedgerRecord(paths.healLedgerPath, thread.threadId, 'healed') + } catch (error) { + if (isCodexAppServerUnsupportedError(error)) { + throw error + } + const message = error instanceof Error ? error.message : String(error) + if (!message.startsWith('codex app-server thread/read failed')) { + // Not an RPC-level response: the server died or timed out. Abort the + // batch without recording, so the id is retried on the next pass. + throw error + } + 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') + return + } + summary.failedThreads += 1 + appendHealLedgerRecord(paths.healLedgerPath, thread.threadId, 'failed') + } +} + +function buildNativeHealInvocation( + systemCodexHomePath: string, + timeoutMs: number +): CodexAppServerInvocation { + const command = resolveCodexCommand() + const { spawnCmd, spawnArgs } = getSpawnArgsForWindows(command, ['app-server']) + return { + command: spawnCmd, + args: spawnArgs, + // Why: pin the real home explicitly — nested Orca launches can inherit a + // managed CODEX_HOME from the daemon environment, which would index the + // wrong sqlite DB. + env: { CODEX_HOME: systemCodexHomePath }, + timeoutMs + } +} diff --git a/src/main/index.ts b/src/main/index.ts index 26105bb8b82..ddfd90e9ffd 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -135,6 +135,7 @@ import { } from './codex/codex-real-home-hook-install' import { setCodexTrustGrantTelemetry } from './codex/codex-hook-trust-grant' import { startCodexSessionBackfillInBackground } from './codex/codex-session-backfill' +import { startCodexSessionIndexHealInBackground } from './codex/codex-session-index-heal' import { resolveHostCodexSessionSourceHome } from './codex/codex-session-source-home' import { getDefaultWslDistro } from './wsl' import { ClaudeAccountService } from './claude-accounts/service' @@ -1807,9 +1808,15 @@ app.whenReady().then(async () => { // #8612). Deferred so startup and first PTY spawns never compete with the // sessions tree walk. setTimeout(() => { - void startCodexSessionBackfillInBackground( - {}, - resolveHostCodexSessionSourceHome(store!.getSettings()) + const systemCodexHomePathOverride = resolveHostCodexSessionSourceHome(store!.getSettings()) + // Why: the heal pass chains after the backfill settles so thread/read only + // runs once the audit ledger covers this startup's newly linked rollouts; + // it also drains sessions left pending by an interrupted earlier pass. + void startCodexSessionBackfillInBackground({}, systemCodexHomePathOverride).then(() => + startCodexSessionIndexHealInBackground( + { shouldStop: () => isQuitting }, + systemCodexHomePathOverride + ) ) }, 15_000) claudeRuntimeAuth = new ClaudeRuntimeAuthService(store)