mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
Merge remote-tracking branch 'origin/brennanb2025/structured-linux-teardown' into brennanb2025/structured-session-eviction
This commit is contained in:
@@ -321,11 +321,10 @@ describe('openCodexAppServerConnection', () => {
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
await closing
|
||||
|
||||
expect(child.kill).toHaveBeenCalledWith('SIGKILL')
|
||||
await vi.waitFor(() => expect(child.kill).toHaveBeenCalledWith('SIGKILL'))
|
||||
})
|
||||
|
||||
it('reports unproven close when forced termination did not produce an exit event', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { child, spawnImpl } = stubChild({ exitOnStdinEnd: false })
|
||||
answerInitialize(child)
|
||||
const connection = await openCodexAppServerConnection(
|
||||
@@ -334,11 +333,8 @@ describe('openCodexAppServerConnection', () => {
|
||||
spawnImpl
|
||||
)
|
||||
|
||||
const closing = connection.close()
|
||||
await vi.advanceTimersByTimeAsync(2_600)
|
||||
|
||||
await expect(closing).resolves.toBe(false)
|
||||
})
|
||||
await expect(connection.close()).resolves.toBe(false)
|
||||
}, 10_000)
|
||||
|
||||
it('ends the connection rather than buffering an oversized line', async () => {
|
||||
const { child, spawnImpl } = stubChild({ exitOnStdinEnd: false })
|
||||
@@ -398,7 +394,7 @@ describe('openCodexAppServerConnection', () => {
|
||||
expect((await inFlight).message).toContain('structured sink failed')
|
||||
expect(exits).toEqual([expect.stringContaining('structured sink failed')])
|
||||
expect(connection.closed).toBe(true)
|
||||
expect(child.kill).toHaveBeenCalledWith('SIGKILL')
|
||||
await vi.waitFor(() => expect(child.kill).toHaveBeenCalledWith('SIGKILL'))
|
||||
await connection.close()
|
||||
})
|
||||
|
||||
@@ -447,7 +443,7 @@ describe('openCodexAppServerConnection', () => {
|
||||
// A child nobody can write to is not a live session: the owner must see the
|
||||
// connection as gone rather than keep issuing calls that can only time out.
|
||||
expect(connection.closed).toBe(true)
|
||||
expect(child.kill).toHaveBeenCalledWith('SIGKILL')
|
||||
await vi.waitFor(() => expect(child.kill).toHaveBeenCalledWith('SIGKILL'))
|
||||
expect((await rejection(connection.request('turn/start'))).message).toContain('EPIPE')
|
||||
await connection.close()
|
||||
})
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
import { spawnProcess } from '../../shared/child-process/run-process'
|
||||
import { buildCodexAppServerExitError } from './codex-app-server-exit-error'
|
||||
import { isAppServerRecord, parseCodexAppServerJsonLine } from './codex-app-server-jsonl'
|
||||
import { terminateCodexAppServerProcessTree } from './codex-app-server-process-teardown'
|
||||
import { CodexAppServerRequestError } from './codex-app-server-request-error'
|
||||
import { CODEX_SPAWN_TOKEN_ENV } from './codex-structured-owner-identity'
|
||||
import { waitForProcessExitUntil } from './codex-process-exit-deadline'
|
||||
import {
|
||||
CodexAppServerTimeoutError,
|
||||
CodexAppServerUnsupportedError,
|
||||
isCodexMethodNotFoundError,
|
||||
killCodexAppServerProcessTree
|
||||
isCodexMethodNotFoundError
|
||||
} from './codex-app-server-session'
|
||||
import type {
|
||||
CodexAppServerConnection,
|
||||
@@ -18,27 +20,14 @@ export type {
|
||||
CodexAppServerConnectionHandlers,
|
||||
CodexAppServerServerRequest
|
||||
} from './codex-app-server-connection-types'
|
||||
export {
|
||||
CodexAppServerRequestError,
|
||||
isCodexAppServerRequestError
|
||||
} from './codex-app-server-request-error'
|
||||
|
||||
// Structured chat needs a persistent bidirectional child and per-request deadlines;
|
||||
// the request-scoped app-server runner cannot carry approvals or streamed turns.
|
||||
|
||||
/** Codex answered the call and refused it. Distinct from a timeout or a dead
|
||||
* child, which leave the call unsettled rather than declined. */
|
||||
export class CodexAppServerRequestError extends Error {
|
||||
constructor(
|
||||
readonly method: string,
|
||||
readonly code: number | null,
|
||||
message: string
|
||||
) {
|
||||
super(message)
|
||||
this.name = 'CodexAppServerRequestError'
|
||||
}
|
||||
}
|
||||
|
||||
export function isCodexAppServerRequestError(error: unknown): error is CodexAppServerRequestError {
|
||||
return error instanceof Error && error.name === 'CodexAppServerRequestError'
|
||||
}
|
||||
|
||||
export type CodexAppServerLaunch = {
|
||||
command: string
|
||||
args: string[]
|
||||
@@ -82,6 +71,7 @@ export async function openCodexAppServerConnection(
|
||||
env: childEnv,
|
||||
cwd: process.cwd()
|
||||
})
|
||||
const spawnToken = launch.env?.[CODEX_SPAWN_TOKEN_ENV]
|
||||
|
||||
const pending = new Map<number, PendingRequest>()
|
||||
let stderrTail = ''
|
||||
@@ -150,7 +140,7 @@ export async function openCodexAppServerConnection(
|
||||
failPending(error)
|
||||
return
|
||||
}
|
||||
killCodexAppServerProcessTree(child)
|
||||
void terminateCodexAppServerProcessTree(child, spawnToken)
|
||||
handleUnexpectedEnd(error)
|
||||
})
|
||||
|
||||
@@ -206,7 +196,7 @@ export async function openCodexAppServerConnection(
|
||||
stdoutBuffer += chunk
|
||||
if (Buffer.byteLength(stdoutBuffer) > STDOUT_LINE_MAX_BYTES) {
|
||||
child.stdout.destroy()
|
||||
killCodexAppServerProcessTree(child)
|
||||
void terminateCodexAppServerProcessTree(child, spawnToken)
|
||||
handleUnexpectedEnd(new Error('codex app-server emitted an oversized JSONL line'))
|
||||
return
|
||||
}
|
||||
@@ -226,7 +216,7 @@ export async function openCodexAppServerConnection(
|
||||
dispatchMessage(parsed)
|
||||
} catch (error) {
|
||||
child.stdout.destroy()
|
||||
killCodexAppServerProcessTree(child)
|
||||
void terminateCodexAppServerProcessTree(child, spawnToken)
|
||||
handleUnexpectedEnd(error instanceof Error ? error : new Error(String(error)))
|
||||
return
|
||||
}
|
||||
@@ -306,7 +296,11 @@ export async function openCodexAppServerConnection(
|
||||
if (!exited) {
|
||||
await waitForProcessExitUntil(exitPromise, GRACEFUL_EXIT_MS)
|
||||
if (!exited) {
|
||||
killCodexAppServerProcessTree(child)
|
||||
const treeExited = await terminateCodexAppServerProcessTree(child, spawnToken)
|
||||
if (!treeExited) {
|
||||
failPending(new Error('codex app-server process-tree exit was not proven'))
|
||||
return false
|
||||
}
|
||||
await waitForProcessExitUntil(exitPromise, FORCED_EXIT_MS)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import type { ChildProcess } from 'node:child_process'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { terminateCodexAppServerProcessTree } from './codex-app-server-process-teardown'
|
||||
|
||||
function child() {
|
||||
return {
|
||||
pid: 1234,
|
||||
kill: vi.fn(() => true) as ChildProcess['kill']
|
||||
}
|
||||
}
|
||||
|
||||
describe('terminateCodexAppServerProcessTree', () => {
|
||||
it('waits for the Windows tree kill before releasing the wrapper', async () => {
|
||||
const target = child()
|
||||
const release = Promise.withResolvers<void>()
|
||||
const terminateWindowsTree = vi.fn(() => release.promise)
|
||||
|
||||
const teardown = terminateCodexAppServerProcessTree(target, undefined, {
|
||||
platform: 'win32',
|
||||
terminateWindowsTree
|
||||
})
|
||||
expect(target.kill).not.toHaveBeenCalled()
|
||||
release.resolve()
|
||||
await teardown
|
||||
|
||||
expect(terminateWindowsTree).toHaveBeenCalledWith(1234)
|
||||
expect(target.kill).toHaveBeenCalledWith('SIGKILL')
|
||||
})
|
||||
|
||||
it('kills exact Linux spawn-token PIDs before the recorded wrapper', async () => {
|
||||
const target = child()
|
||||
const findSpawnTokenProcesses = vi
|
||||
.fn<() => Promise<number[] | null>>()
|
||||
.mockResolvedValueOnce([1234, 2345, 3456])
|
||||
.mockResolvedValueOnce([1234])
|
||||
.mockResolvedValueOnce([1234])
|
||||
const signalPid = vi.fn()
|
||||
|
||||
await expect(
|
||||
terminateCodexAppServerProcessTree(target, 'spawn-1', {
|
||||
platform: 'linux',
|
||||
findSpawnTokenProcesses,
|
||||
signalPid,
|
||||
isPidPresent: () => false,
|
||||
wait: async () => undefined
|
||||
})
|
||||
).resolves.toBe(true)
|
||||
|
||||
expect(signalPid.mock.calls).toEqual([
|
||||
[2345, 'SIGKILL'],
|
||||
[3456, 'SIGKILL']
|
||||
])
|
||||
expect(target.kill).toHaveBeenCalledTimes(1)
|
||||
expect(target.kill).toHaveBeenCalledWith('SIGKILL')
|
||||
})
|
||||
|
||||
it('keeps the wrapper reachable when Linux cannot prove descendant exit', async () => {
|
||||
const target = child()
|
||||
|
||||
await expect(
|
||||
terminateCodexAppServerProcessTree(target, 'spawn-1', {
|
||||
platform: 'linux',
|
||||
findSpawnTokenProcesses: async () => null
|
||||
})
|
||||
).resolves.toBe(false)
|
||||
|
||||
expect(target.kill).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('waits for an owned POSIX snapshot before killing the wrapper', async () => {
|
||||
const target = child()
|
||||
const snapshot = { rootPgid: 1234, descendants: [], capturedAtMs: 1 }
|
||||
const release = Promise.withResolvers<boolean>()
|
||||
|
||||
const teardown = terminateCodexAppServerProcessTree(target, undefined, {
|
||||
platform: 'darwin',
|
||||
captureDescendants: async () => snapshot,
|
||||
terminateDescendants: () => release.promise
|
||||
})
|
||||
await vi.waitFor(() => expect(target.kill).toHaveBeenCalledWith('SIGSTOP'))
|
||||
expect(target.kill).not.toHaveBeenCalledWith('SIGKILL')
|
||||
release.resolve(true)
|
||||
await teardown
|
||||
|
||||
expect(target.kill).toHaveBeenLastCalledWith('SIGKILL')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,158 @@
|
||||
import type { ChildProcess } from 'node:child_process'
|
||||
import { captureDescendantSnapshot, type DescendantSnapshot } from '../pty-descendant-termination'
|
||||
import { terminateDescendantSnapshotAndWait } from '../pty-descendant-exit-verification'
|
||||
import { findAgentSessionSpawnTokenProcesses } from '../runtime/agent-session-spawn-token-process-scan'
|
||||
import { terminateWindowsProcessTree } from '../windows-process-tree-kill'
|
||||
|
||||
const TOKEN_PROCESS_EXIT_TIMEOUT_MS = 3_500
|
||||
const TOKEN_PROCESS_POLL_MS = 25
|
||||
const activeTeardowns = new WeakMap<object, Promise<boolean>>()
|
||||
|
||||
type TeardownChild = Pick<ChildProcess, 'pid' | 'kill'>
|
||||
|
||||
export type CodexAppServerProcessTeardownDeps = {
|
||||
platform?: NodeJS.Platform
|
||||
findSpawnTokenProcesses?: (spawnToken: string) => Promise<number[] | null>
|
||||
captureDescendants?: (rootPid: number) => Promise<DescendantSnapshot | null>
|
||||
terminateDescendants?: (snapshot: DescendantSnapshot) => Promise<boolean>
|
||||
terminateWindowsTree?: (rootPid: number) => Promise<void>
|
||||
signalPid?: (pid: number, signal: NodeJS.Signals) => void
|
||||
isPidPresent?: (pid: number) => boolean
|
||||
wait?: (ms: number) => Promise<void>
|
||||
now?: () => number
|
||||
}
|
||||
|
||||
function sendSignal(pid: number, signal: NodeJS.Signals): void {
|
||||
try {
|
||||
process.kill(pid, signal)
|
||||
} catch {
|
||||
// An already-gone exact PID is the desired outcome.
|
||||
}
|
||||
}
|
||||
|
||||
function wait(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(resolve, ms)
|
||||
timer.unref?.()
|
||||
})
|
||||
}
|
||||
|
||||
function isPidPresent(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== 'ESRCH'
|
||||
}
|
||||
}
|
||||
|
||||
async function terminateSpawnTokenProcesses(
|
||||
rootPid: number,
|
||||
spawnToken: string,
|
||||
deps: CodexAppServerProcessTeardownDeps
|
||||
): Promise<boolean> {
|
||||
const find = deps.findSpawnTokenProcesses ?? findAgentSessionSpawnTokenProcesses
|
||||
const signal = deps.signalPid ?? sendSignal
|
||||
const pidPresent = deps.isPidPresent ?? isPidPresent
|
||||
const delay = deps.wait ?? wait
|
||||
const now = deps.now ?? Date.now
|
||||
const deadline = now() + TOKEN_PROCESS_EXIT_TIMEOUT_MS
|
||||
const signalled = new Set<number>()
|
||||
let emptyScans = 0
|
||||
|
||||
while (now() < deadline) {
|
||||
const pids = await find(spawnToken).catch(() => null)
|
||||
if (pids === null) {
|
||||
return false
|
||||
}
|
||||
const descendants = pids.filter((pid) => pid !== rootPid)
|
||||
for (const pid of descendants) {
|
||||
signalled.add(pid)
|
||||
signal(pid, 'SIGKILL')
|
||||
}
|
||||
const allSignalledPidsGone = [...signalled].every((pid) => !pidPresent(pid))
|
||||
emptyScans = descendants.length === 0 && allSignalledPidsGone ? emptyScans + 1 : 0
|
||||
if (emptyScans >= 2) {
|
||||
return true
|
||||
}
|
||||
await delay(TOKEN_PROCESS_POLL_MS)
|
||||
}
|
||||
const remaining = await find(spawnToken).catch(() => null)
|
||||
return (
|
||||
remaining !== null &&
|
||||
remaining.every((pid) => pid === rootPid) &&
|
||||
[...signalled].every((pid) => !pidPresent(pid))
|
||||
)
|
||||
}
|
||||
|
||||
async function terminatePosixTree(
|
||||
child: TeardownChild,
|
||||
rootPid: number,
|
||||
spawnToken: string | undefined,
|
||||
deps: CodexAppServerProcessTeardownDeps
|
||||
): Promise<boolean> {
|
||||
const platform = deps.platform ?? process.platform
|
||||
if (platform === 'linux' && spawnToken) {
|
||||
const descendantsExited = await terminateSpawnTokenProcesses(rootPid, spawnToken, deps)
|
||||
if (descendantsExited) {
|
||||
child.kill('SIGKILL')
|
||||
}
|
||||
return descendantsExited
|
||||
}
|
||||
child.kill('SIGSTOP')
|
||||
const capture = deps.captureDescendants ?? captureDescendantSnapshot
|
||||
const snapshot = await capture(rootPid).catch(() => null)
|
||||
if (!snapshot) {
|
||||
child.kill('SIGKILL')
|
||||
return true
|
||||
}
|
||||
const terminate = deps.terminateDescendants ?? terminateDescendantSnapshotAndWait
|
||||
const descendantsExited = await terminate(snapshot)
|
||||
if (!descendantsExited) {
|
||||
child.kill('SIGCONT')
|
||||
return false
|
||||
}
|
||||
child.kill('SIGKILL')
|
||||
return true
|
||||
}
|
||||
|
||||
/** Stops every process owned by one app-server launch before releasing its wrapper. */
|
||||
async function terminateOnce(
|
||||
child: TeardownChild,
|
||||
spawnToken: string | undefined,
|
||||
deps: CodexAppServerProcessTeardownDeps
|
||||
): Promise<boolean> {
|
||||
const rootPid = child.pid
|
||||
if (!rootPid) {
|
||||
child.kill('SIGKILL')
|
||||
return false
|
||||
}
|
||||
if ((deps.platform ?? process.platform) === 'win32') {
|
||||
const terminate = deps.terminateWindowsTree ?? terminateWindowsProcessTree
|
||||
await terminate(rootPid)
|
||||
// taskkill owns the tree; this preserves the prior direct-child fallback when it fails.
|
||||
child.kill('SIGKILL')
|
||||
return true
|
||||
}
|
||||
return terminatePosixTree(child, rootPid, spawnToken, deps)
|
||||
}
|
||||
|
||||
export function terminateCodexAppServerProcessTree(
|
||||
child: TeardownChild,
|
||||
spawnToken?: string,
|
||||
deps: CodexAppServerProcessTeardownDeps = {}
|
||||
): Promise<boolean> {
|
||||
const key = child as object
|
||||
const active = activeTeardowns.get(key)
|
||||
if (active) {
|
||||
return active
|
||||
}
|
||||
const attempt = terminateOnce(child, spawnToken, deps).catch(() => false)
|
||||
activeTeardowns.set(key, attempt)
|
||||
void attempt.then(() => {
|
||||
if (activeTeardowns.get(key) === attempt) {
|
||||
activeTeardowns.delete(key)
|
||||
}
|
||||
})
|
||||
return attempt
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
/** Codex answered the call and refused it, rather than timing out or exiting. */
|
||||
export class CodexAppServerRequestError extends Error {
|
||||
constructor(
|
||||
readonly method: string,
|
||||
readonly code: number | null,
|
||||
message: string
|
||||
) {
|
||||
super(message)
|
||||
this.name = 'CodexAppServerRequestError'
|
||||
}
|
||||
}
|
||||
|
||||
export function isCodexAppServerRequestError(error: unknown): error is CodexAppServerRequestError {
|
||||
return error instanceof Error && error.name === 'CodexAppServerRequestError'
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { CODEX_SPAWN_TOKEN_ENV } from './codex-structured-owner-identity'
|
||||
import {
|
||||
openCodexAppServerConnection,
|
||||
type CodexAppServerConnection
|
||||
} from './codex-app-server-connection'
|
||||
|
||||
const ITERATIONS = 40
|
||||
|
||||
const FORCE_KILL_APP_SERVER = String.raw`
|
||||
const { spawn } = require('node:child_process')
|
||||
const readline = require('node:readline')
|
||||
const descendant = spawn(process.execPath, ['-e', 'setInterval(() => {}, 60000)'], {
|
||||
stdio: 'ignore'
|
||||
})
|
||||
const send = (payload) => process.stdout.write(JSON.stringify(payload) + '\n')
|
||||
readline.createInterface({ input: process.stdin }).on('line', (line) => {
|
||||
const message = JSON.parse(line)
|
||||
if (message.method === 'initialize') return send({ id: message.id, result: {} })
|
||||
if (message.method === 'initialized') {
|
||||
send({ method: 'test/descendant', params: { pid: descendant.pid } })
|
||||
}
|
||||
})
|
||||
setInterval(() => {}, 60000)
|
||||
`
|
||||
|
||||
function processExists(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
type RunningServer = {
|
||||
connection: CodexAppServerConnection
|
||||
descendantPid: number
|
||||
}
|
||||
|
||||
async function openServer(iteration: number): Promise<RunningServer> {
|
||||
const descendant = Promise.withResolvers<number>()
|
||||
const connection = await openCodexAppServerConnection(
|
||||
{
|
||||
command: process.execPath,
|
||||
args: ['-e', FORCE_KILL_APP_SERVER],
|
||||
env: { [CODEX_SPAWN_TOKEN_ENV]: `teardown-test-${process.pid}-${iteration}` }
|
||||
},
|
||||
{
|
||||
onNotification: (method, params) => {
|
||||
if (method === 'test/descendant') {
|
||||
descendant.resolve((params as { pid: number }).pid)
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
return { connection, descendantPid: await descendant.promise }
|
||||
}
|
||||
|
||||
describe.runIf(process.platform === 'linux')('Codex app-server process teardown', () => {
|
||||
it('reaps the forced-close descendant in 40 consecutive launches', async () => {
|
||||
const running: RunningServer[] = []
|
||||
try {
|
||||
for (let iteration = 0; iteration < ITERATIONS; iteration += 1) {
|
||||
running.push(await openServer(iteration))
|
||||
}
|
||||
expect(running.every(({ descendantPid }) => processExists(descendantPid))).toBe(true)
|
||||
|
||||
const closed = await Promise.all(running.map(({ connection }) => connection.close()))
|
||||
|
||||
expect(closed).toEqual(Array.from({ length: ITERATIONS }, () => true))
|
||||
expect(running.filter(({ descendantPid }) => processExists(descendantPid))).toEqual([])
|
||||
} finally {
|
||||
for (const { connection, descendantPid } of running) {
|
||||
await connection.close().catch(() => false)
|
||||
if (processExists(descendantPid)) {
|
||||
process.kill(descendantPid, 'SIGKILL')
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 30_000)
|
||||
})
|
||||
Reference in New Issue
Block a user