mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 00:03:20 +00:00
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.
This commit is contained in:
Brennan Benson
@@ -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",
|
||||
|
||||
@@ -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<string, string>
|
||||
/** 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<CodexHookTrustGrantSessionResult> {
|
||||
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<void>((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<string, { trusted_hash: string }> = {}
|
||||
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<string, unknown>): void {
|
||||
child.stdin.write(`${JSON.stringify(payload)}\n`)
|
||||
}
|
||||
|
||||
async function requestRpc(method: string, params?: Record<string, unknown>): Promise<unknown> {
|
||||
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<JsonRpcResponse>((resolve, reject) => {
|
||||
pending.set(id, { resolve, reject })
|
||||
const payload: Record<string, unknown> = { 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<string, { trusted_hash: string }> = {}
|
||||
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<void>((resolve) => setTimeout(resolve, 1500))
|
||||
await Promise.race([exitPromise, grace])
|
||||
if (!exited) {
|
||||
child.kill('SIGKILL')
|
||||
await Promise.race([exitPromise, new Promise<void>((resolve) => setTimeout(resolve, 1000))])
|
||||
}
|
||||
}
|
||||
clearTimeout(deadline)
|
||||
}
|
||||
},
|
||||
spawnImpl
|
||||
)
|
||||
}
|
||||
|
||||
@@ -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<string, string>
|
||||
/** 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<string, unknown>) => Promise<unknown>
|
||||
notify: (method: string, params?: Record<string, unknown>) => 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<T>(
|
||||
invocation: CodexAppServerInvocation,
|
||||
body: (rpc: CodexAppServerRpc) => Promise<T>,
|
||||
spawnImpl: typeof spawn = spawn
|
||||
): Promise<T> {
|
||||
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<void>((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<string, unknown>): void {
|
||||
child.stdin.write(`${JSON.stringify(payload)}\n`)
|
||||
}
|
||||
|
||||
function notify(method: string, params?: Record<string, unknown>): void {
|
||||
const payload: Record<string, unknown> = { 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<string, unknown>): Promise<unknown> {
|
||||
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<JsonRpcResponse>((resolve, reject) => {
|
||||
pending.set(id, { resolve, reject })
|
||||
const payload: Record<string, unknown> = { 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<void>((resolve) => setTimeout(resolve, 1500))
|
||||
await Promise.race([exitPromise, grace])
|
||||
if (!exited) {
|
||||
child.kill('SIGKILL')
|
||||
await Promise.race([exitPromise, new Promise<void>((resolve) => setTimeout(resolve, 1000))])
|
||||
}
|
||||
}
|
||||
clearTimeout(deadline)
|
||||
}
|
||||
}
|
||||
@@ -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<string, PendingHealThread>()
|
||||
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<string> {
|
||||
const processed = new Set<string>()
|
||||
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<string, unknown>[] {
|
||||
let contents: string
|
||||
try {
|
||||
contents = readFileSync(filePath, 'utf-8')
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
const lines: Record<string, unknown>[] = []
|
||||
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<string, unknown>)
|
||||
}
|
||||
} 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)
|
||||
}
|
||||
}
|
||||
@@ -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<string, string> {
|
||||
let contents = ''
|
||||
try {
|
||||
contents = readFileSync(paths.healLedgerPath, 'utf-8')
|
||||
} catch {
|
||||
return {}
|
||||
}
|
||||
const outcomes: Record<string, string> = {}
|
||||
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)
|
||||
})
|
||||
})
|
||||
@@ -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<CodexSessionIndexHealSummary | null> | 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<CodexSessionIndexHealSummary | null> {
|
||||
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<CodexSessionIndexHealSummary> {
|
||||
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<void> => {
|
||||
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<string, unknown>) => Promise<unknown> },
|
||||
thread: PendingHealThread,
|
||||
paths: CodexSessionIndexHealPaths,
|
||||
summary: CodexSessionIndexHealSummary
|
||||
): Promise<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
+10
-3
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user