diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts index 1775180913e..0f41bbd5156 100644 --- a/src/main/codex/codex-structured-session-adapter.test.ts +++ b/src/main/codex/codex-structured-session-adapter.test.ts @@ -11,12 +11,12 @@ import type { openCodexAppServerConnection } from './codex-app-server-connection' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' -import { CodexAppServerUnsupportedError } from './codex-app-server-session' import { CODEX_SPAWN_TOKEN_ENV } from './codex-structured-owner-identity' import { encodeCodexQuestionOptionId } from './codex-structured-prompt-replies' import { CodexStructuredSessionAdapter, type CodexStructuredLaunch, + type CodexStructuredSessionAdapterDeps, type CodexStructuredSessionEvent } from './codex-structured-session-adapter' @@ -100,7 +100,10 @@ function fakeCodex(routes: Record = {}): { function adapterFor( codex: ReturnType, launch: Partial = {}, - events: CodexStructuredSessionEvent[] = [] + events: CodexStructuredSessionEvent[] = [], + processControl: Partial< + Pick + > = {} ): CodexStructuredSessionAdapter { return new CodexStructuredSessionAdapter({ resolveLaunch: async () => ({ @@ -114,7 +117,10 @@ function adapterFor( onEvent: (event) => events.push(event), openConnection: codex.openConnection, readProcessStartTime: async () => 1_700_000_000_000, - now: () => 1_700_000_000_500 + captureTurnProcesses: async () => ({ platform: 'win32', identities: new Map() }), + terminateTurnProcesses: async () => true, + now: () => 1_700_000_000_500, + ...processControl }) } @@ -564,58 +570,6 @@ describe('CodexStructuredSessionAdapter.dispatch', () => { }) }) -describe('CodexStructuredSessionAdapter.cancelTurn', () => { - it('confirms an interrupt Codex acknowledged', async () => { - const codex = fakeCodex() - const adapter = await acquired(codex) - - expect( - await adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).toEqual({ cancelled: true }) - expect(codex.connections[0].calls[1]).toEqual({ - method: 'turn/interrupt', - params: { threadId: THREAD_ID, turnId: 'turn-1' } - }) - }) - - it('reports not-cancelled when Codex declines or lacks the method', async () => { - const declined = fakeCodex({ - 'turn/interrupt': () => { - throw new CodexAppServerRequestError('turn/interrupt', -32602, 'no such turn') - } - }) - const absent = fakeCodex({ - 'turn/interrupt': () => { - throw new CodexAppServerUnsupportedError('no turn/interrupt') - } - }) - - expect( - await ( - await acquired(declined) - ).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).toEqual({ cancelled: false }) - expect( - await ( - await acquired(absent) - ).cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).toEqual({ cancelled: false }) - }) - - it('rethrows an unsettled interrupt so the turn is not shown as cancelled', async () => { - const codex = fakeCodex({ - 'turn/interrupt': () => { - throw new Error('codex app-server turn/interrupt exceeded 30000ms') - } - }) - const adapter = await acquired(codex) - - await expect( - adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).rejects.toThrow('exceeded 30000ms') - }) -}) - describe('CodexStructuredSessionAdapter prompts', () => { function askApproval(codex: ReturnType): void { codex.connections[0].handlers.onServerRequest?.({ diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index e7d27ac699b..35fa0b17666 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -11,11 +11,7 @@ import { type StructuredAgentSessionSetOptionInput } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { createCodexJournalTranslator } from './codex-structured-journal-translation' -import { - isCodexAppServerRequestError, - openCodexAppServerConnection -} from './codex-app-server-connection' -import { isCodexAppServerUnsupportedError } from './codex-app-server-session' +import { openCodexAppServerConnection } from './codex-app-server-connection' import { codexProcessIdentity, codexProviderHandleLink } from './codex-structured-owner-identity' import { buildCodexStructuredChildEnvironment } from './codex-structured-child-environment' import { answerCodexPrompt } from './codex-structured-prompt-replies' @@ -42,6 +38,7 @@ import { deliverCodexServerRequest, deliverCodexUnhandledFrame } from './codex-structured-provider-events' +import { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation' export type { CodexStructuredLaunch, @@ -52,8 +49,16 @@ export type { export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdapter { private readonly sessions = new Map() private readonly acquisitions = new CodexAcquisitionRegistry() + private readonly turnCancellation: CodexStructuredTurnCancellation - constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {} + constructor(private readonly deps: CodexStructuredSessionAdapterDeps) { + this.turnCancellation = new CodexStructuredTurnCancellation({ + captureTurnProcesses: deps.captureTurnProcesses, + terminateTurnProcesses: deps.terminateTurnProcesses, + requestTimeoutMs: deps.requestTimeoutMs, + emit: (session, event) => this.emit(session, event) + }) + } supportsLocation = supportsCodexStructuredLocation @@ -132,7 +137,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap } this.acquisitions.assertCurrent(sessionId, attempt) this.acquisitions.deleteIfCurrent(sessionId, attempt) - this.sessions.set(sessionId, { + const session: CodexSession = { connection, threadId: opened.threadId, historyPath: opened.historyPath, @@ -141,7 +146,9 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap reportedOptions: reportedCodexThreadOptions(opened), turnIdWaiters: [], translator - }) + } + this.turnCancellation.register(session) + this.sessions.set(sessionId, session) for (const event of acquisition.drain()) { event() } @@ -191,12 +198,12 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap } private handleNotification(sessionId: string, method: string, params: unknown): void { - deliverCodexNotification( - sessionId, - this.sessions.get(sessionId), - method, - params, - (session, event) => this.emit(session, event) + const session = this.sessions.get(sessionId) + if (session && this.turnCancellation.handleNotification(sessionId, session, method, params)) { + return + } + deliverCodexNotification(sessionId, session, method, params, (session, event) => + this.emit(session, event) ) } @@ -236,7 +243,9 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap body: AgentJournalMessageItem fence: number }): Promise { - return dispatchCodexTurn(this.session(input.sessionId), input, this.deps.requestTimeoutMs) + const session = this.session(input.sessionId) + await this.turnCancellation.captureBaseline(session) + return dispatchCodexTurn(session, input, this.deps.requestTimeoutMs) } async cancelTurn(input: { @@ -245,21 +254,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap fence: number }): Promise<{ cancelled: boolean }> { const session = this.session(input.sessionId) - try { - await session.connection.request( - 'turn/interrupt', - { threadId: session.threadId, turnId: input.turnId }, - { timeoutMs: this.deps.requestTimeoutMs } - ) - return { cancelled: true } - } catch (error) { - // Codex declining names a turn it no longer owns; anything else leaves the - // cancel unconfirmed and must surface as such. - if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) { - return { cancelled: false } - } - throw error - } + return this.turnCancellation.cancel(session, input.turnId) } async answerPrompt(input: { diff --git a/src/main/codex/codex-structured-session-cancel.test.ts b/src/main/codex/codex-structured-session-cancel.test.ts new file mode 100644 index 00000000000..50fc623d45c --- /dev/null +++ b/src/main/codex/codex-structured-session-cancel.test.ts @@ -0,0 +1,278 @@ +import { describe, expect, it, vi } from 'vitest' +import type { + AgentJournalMessageItem, + AgentSessionJournalIdentity +} from '../../shared/agent-session-journal-types' +import { + CodexAppServerRequestError, + type CodexAppServerConnection, + type CodexAppServerConnectionHandlers, + type CodexAppServerLaunch, + type openCodexAppServerConnection +} from './codex-app-server-connection' +import { CodexAppServerUnsupportedError } from './codex-app-server-session' +import { + CodexStructuredSessionAdapter, + type CodexStructuredSessionAdapterDeps, + type CodexStructuredSessionEvent +} from './codex-structured-session-adapter' + +const THREAD_ID = 'thread-abc' +const USER_MESSAGE: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'ship it' }] +} + +type Route = (params: Record | undefined) => unknown +type FakeConnection = Omit & { + closed: boolean + launch: CodexAppServerLaunch + handlers: CodexAppServerConnectionHandlers + calls: { method: string; params?: Record }[] +} + +function identity(): AgentSessionJournalIdentity { + return { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD_ID } + } +} + +function fakeCodex(): { + connections: FakeConnection[] + openConnection: typeof openCodexAppServerConnection + routes: Record +} { + const connections: FakeConnection[] = [] + const routes: Record = { + 'thread/resume': () => ({ thread: { id: THREAD_ID } }) + } + const openConnection = (async (launch, handlers = {}) => { + const connection: FakeConnection = { + launch, + handlers, + calls: [], + pid: 4321, + closed: false, + request: async (method, params) => { + connection.calls.push({ method, params }) + return routes[method]?.(params) ?? {} + }, + notify: () => {}, + respond: () => {}, + respondWithError: () => {}, + close: async () => { + connection.closed = true + } + } + connections.push(connection) + return connection + }) as typeof openCodexAppServerConnection + return { connections, openConnection, routes } +} + +async function acquired( + codex: ReturnType, + events: CodexStructuredSessionEvent[] = [], + processControl: Partial< + Pick + > = {} +): Promise { + const adapter = new CodexStructuredSessionAdapter({ + resolveLaunch: async () => ({ + command: 'codex', + args: ['app-server'], + cwd: '/work/repo', + codexHome: null, + resumeThreadId: THREAD_ID + }), + onEvent: (event) => events.push(event), + openConnection: codex.openConnection, + readProcessStartTime: async () => 1_700_000_000_000, + captureTurnProcesses: async () => ({ platform: 'win32', identities: new Map() }), + terminateTurnProcesses: async () => true, + ...processControl + }) + await adapter.acquire({ identity: identity(), fence: 7, spawnToken: 'spawn-9' }) + return adapter +} + +function completeTurn(codex: ReturnType, turnId = 'turn-1'): void { + codex.connections[0].handlers.onNotification?.('turn/completed', { + threadId: THREAD_ID, + turn: { id: turnId, status: 'interrupted' } + }) +} + +describe('CodexStructuredSessionAdapter.cancelTurn', () => { + it('confirms an interrupt Codex acknowledged', async () => { + const codex = fakeCodex() + const adapter = await acquired(codex) + + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + ).resolves.toEqual({ cancelled: true }) + expect(codex.connections[0].calls.at(-1)).toEqual({ + method: 'turn/interrupt', + params: { threadId: THREAD_ID, turnId: 'turn-1' } + }) + }) + + it('reports not-cancelled when Codex declines or lacks the method', async () => { + const declined = fakeCodex() + declined.routes['turn/interrupt'] = () => { + throw new CodexAppServerRequestError('turn/interrupt', -32602, 'no such turn') + } + const absent = fakeCodex() + absent.routes['turn/interrupt'] = () => { + throw new CodexAppServerUnsupportedError('no turn/interrupt') + } + + await expect( + (await acquired(declined)).cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7 + }) + ).resolves.toEqual({ cancelled: false }) + await expect( + (await acquired(absent)).cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7 + }) + ).resolves.toEqual({ cancelled: false }) + }) + + it('rethrows an unsettled interrupt so the turn is not shown as cancelled', async () => { + const codex = fakeCodex() + codex.routes['turn/interrupt'] = () => { + throw new Error('codex app-server turn/interrupt exceeded 30000ms') + } + + await expect( + (await acquired(codex)).cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7 + }) + ).rejects.toThrow('exceeded 30000ms') + }) + + it('publishes terminal state only after streaming interruption is physically settled', async () => { + const events: CodexStructuredSessionEvent[] = [] + let finishTermination!: (terminated: boolean) => void + const termination = new Promise((resolve) => { + finishTermination = resolve + }) + const codex = fakeCodex() + codex.routes['turn/interrupt'] = () => { + completeTurn(codex) + return {} + } + const adapter = await acquired(codex, events, { + terminateTurnProcesses: async () => termination + }) + codex.connections[0].handlers.onNotification?.('item/agentMessage/delta', { + threadId: THREAD_ID, + turnId: 'turn-1', + itemId: 'item-1', + delta: 'still streaming' + }) + + const pending = adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + await vi.waitFor(() => expect(codex.connections[0].calls.at(-1)?.method).toBe('turn/interrupt')) + expect(events).toContainEqual(expect.objectContaining({ method: 'item/agentMessage/delta' })) + expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' })) + + finishTermination(true) + await expect(pending).resolves.toEqual({ cancelled: true }) + expect(events.at(-1)).toMatchObject({ method: 'turn/completed' }) + }) + + it('starts physical termination without waiting for the interrupt receipt', async () => { + let finishInterrupt!: () => void + const interruptReceipt = new Promise((resolve) => { + finishInterrupt = resolve + }) + const terminateTurnProcesses = vi.fn(async () => true) + const codex = fakeCodex() + codex.routes['turn/interrupt'] = () => interruptReceipt + const adapter = await acquired(codex, [], { terminateTurnProcesses }) + + const pending = adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + await vi.waitFor(() => expect(terminateTurnProcesses).toHaveBeenCalledOnce()) + finishInterrupt() + + await expect(pending).resolves.toEqual({ cancelled: true }) + }) + + it('keeps the turn live when process termination cannot be verified', async () => { + const events: CodexStructuredSessionEvent[] = [] + const codex = fakeCodex() + codex.routes['turn/interrupt'] = () => { + completeTurn(codex) + return {} + } + const adapter = await acquired(codex, events, { + terminateTurnProcesses: async () => false + }) + + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + ).resolves.toEqual({ cancelled: false }) + expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' })) + }) + + it('accepts an immediate resend after verified interruption', async () => { + let nextTurn = 0 + const codex = fakeCodex() + codex.routes['turn/start'] = () => ({ turn: { id: `turn-${++nextTurn}` } }) + codex.routes['turn/interrupt'] = () => { + completeTurn(codex) + return {} + } + const adapter = await acquired(codex) + + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 + }) + await adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-2', + body: USER_MESSAGE, + fence: 7 + }) + ).resolves.toMatchObject({ + state: 'accepted', + providerIdentity: { turnId: 'turn-2' } + }) + }) + + it('does not publish deferred completion when the interrupt receipt fails', async () => { + const events: CodexStructuredSessionEvent[] = [] + const codex = fakeCodex() + codex.routes['turn/interrupt'] = () => { + completeTurn(codex) + throw new Error('interrupt receipt lost') + } + const adapter = await acquired(codex, events, { + terminateTurnProcesses: async () => true + }) + + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + ).rejects.toThrow('interrupt receipt lost') + expect(events).not.toContainEqual(expect.objectContaining({ method: 'turn/completed' })) + }) +}) diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index e65942e5475..ceb37d92ac2 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -5,6 +5,7 @@ import type { } from './codex-app-server-connection' import { CodexAcquisitionWindow } from './codex-structured-acquisition-window' import type { CodexJournalTranslator } from './codex-structured-journal-translation' +import type { CodexTurnProcessSnapshot } from './codex-structured-turn-processes' export type CodexStructuredLaunch = { command: string @@ -40,6 +41,11 @@ export type CodexStructuredSessionAdapterDeps = { mintLinkId?: () => string now?: () => number requestTimeoutMs?: number + captureTurnProcesses?: (rootPid: number) => Promise + terminateTurnProcesses?: ( + rootPid: number, + baseline: CodexTurnProcessSnapshot | null + ) => Promise } export type CodexSession = { diff --git a/src/main/codex/codex-structured-turn-cancellation.ts b/src/main/codex/codex-structured-turn-cancellation.ts new file mode 100644 index 00000000000..ced740d67f9 --- /dev/null +++ b/src/main/codex/codex-structured-turn-cancellation.ts @@ -0,0 +1,150 @@ +import { + isCodexAppServerRequestError, + type CodexAppServerConnection +} from './codex-app-server-connection' +import { isCodexAppServerUnsupportedError } from './codex-app-server-session' +import type { + CodexSession, + CodexStructuredSessionAdapterDeps, + CodexStructuredSessionEvent +} from './codex-structured-session-state' +import { readCodexThreadId, readCodexTurnId } from './codex-structured-thread-facts' +import { + captureCodexTurnProcesses, + terminateCodexTurnProcesses, + type CodexTurnProcessSnapshot +} from './codex-structured-turn-processes' + +type TurnProcessState = { + baseline: Promise + blockedCompletions: Set + deferredCompletions: Map +} + +type TurnCancellationDeps = Pick< + CodexStructuredSessionAdapterDeps, + 'captureTurnProcesses' | 'requestTimeoutMs' | 'terminateTurnProcesses' +> & { + emit: (session: CodexSession, event: CodexStructuredSessionEvent) => void +} + +export class CodexStructuredTurnCancellation { + private readonly states = new WeakMap() + + constructor(private readonly deps: TurnCancellationDeps) {} + + register(session: CodexSession): void { + this.states.set(session, { + baseline: Promise.resolve(null), + blockedCompletions: new Set(), + deferredCompletions: new Map() + }) + } + + captureBaseline(session: CodexSession): Promise { + this.refreshBaseline(session) + return this.state(session).baseline + } + + handleNotification( + sessionId: string, + session: CodexSession, + method: string, + params: unknown + ): boolean { + const threadId = readCodexThreadId(params) ?? session.threadId + if (method !== 'turn/completed' || threadId !== session.threadId) { + return false + } + const turnId = readCodexTurnId(params) + const state = this.state(session) + if (!turnId || !state.blockedCompletions.has(turnId)) { + return false + } + const event = { + type: 'notification' as const, + sessionId, + threadId, + method, + params + } + state.deferredCompletions.set(turnId, event) + return true + } + + async cancel(session: CodexSession, turnId: string): Promise<{ cancelled: boolean }> { + const state = this.state(session) + state.blockedCompletions.add(turnId) + const baseline = await state.baseline + let requestError: unknown + const interruptReceipt = session.connection + .request( + 'turn/interrupt', + { threadId: session.threadId, turnId }, + { timeoutMs: this.deps.requestTimeoutMs } + ) + .then( + () => true, + (error: unknown) => { + requestError = error + return false + } + ) + const [acknowledged, terminated] = await Promise.all([ + interruptReceipt, + this.terminate(session.connection, baseline) + ]) + if (terminated && acknowledged) { + this.releaseCompletion(session, turnId) + return { cancelled: true } + } + if ( + requestError && + !isCodexAppServerRequestError(requestError) && + !isCodexAppServerUnsupportedError(requestError) + ) { + throw requestError + } + return { cancelled: false } + } + + private capture(pid: number | undefined): Promise { + return pid + ? (this.deps.captureTurnProcesses ?? captureCodexTurnProcesses)(pid) + : Promise.resolve(null) + } + + private terminate( + connection: Pick, + baseline: CodexTurnProcessSnapshot | null + ): Promise { + return connection.pid + ? (this.deps.terminateTurnProcesses ?? terminateCodexTurnProcesses)(connection.pid, baseline) + : Promise.resolve(false) + } + + private refreshBaseline(session: CodexSession): void { + this.state(session).baseline = this.capture(session.connection.pid) + } + + private releaseCompletion( + session: CodexSession, + turnId: string, + completion = this.state(session).deferredCompletions.get(turnId) + ): void { + const state = this.state(session) + state.blockedCompletions.delete(turnId) + state.deferredCompletions.delete(turnId) + if (completion) { + this.deps.emit(session, completion) + } + } + + private state(session: CodexSession): TurnProcessState { + const state = this.states.get(session) + if (!state) { + throw new Error('codex turn process state is unavailable') + } + return state + } +} diff --git a/src/main/codex/codex-structured-turn-processes.integration.test.ts b/src/main/codex/codex-structured-turn-processes.integration.test.ts new file mode 100644 index 00000000000..2a63c8099ea --- /dev/null +++ b/src/main/codex/codex-structured-turn-processes.integration.test.ts @@ -0,0 +1,103 @@ +import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process' +import { describe, expect, it } from 'vitest' +import { + captureCodexTurnProcesses, + terminateCodexTurnProcesses +} from './codex-structured-turn-processes' + +function nextLine(child: ChildProcessWithoutNullStreams): Promise { + return new Promise((resolve, reject) => { + let buffer = '' + const onData = (chunk: Buffer): void => { + buffer += chunk.toString('utf8') + const newline = buffer.indexOf('\n') + if (newline === -1) { + return + } + child.stdout.off('data', onData) + resolve(buffer.slice(0, newline)) + } + child.once('error', reject) + child.stdout.on('data', onData) + }) +} + +function processExists(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch { + return false + } +} + +describe.runIf(process.platform !== 'win32')('Codex structured turn process termination', () => { + it('removes the exact PID of a stopped 60-second command', async () => { + const root = spawn( + process.execPath, + [ + '-e', + `const { spawn } = require('node:child_process'); + process.stdin.once('data', () => { + const child = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' }); + process.stdout.write(String(child.pid) + '\\n'); + }); + setTimeout(() => {}, 60000);` + ], + { stdio: ['pipe', 'pipe', 'pipe'] } + ) + let commandPid = 0 + try { + const baseline = await captureCodexTurnProcesses(root.pid!) + root.stdin.write('start\n') + commandPid = Number(await nextLine(root)) + expect(processExists(commandPid)).toBe(true) + + await expect(terminateCodexTurnProcesses(root.pid!, baseline)).resolves.toBe(true) + expect(processExists(commandPid)).toBe(false) + } finally { + if (commandPid > 0 && processExists(commandPid)) { + process.kill(commandPid, 'SIGKILL') + } + root.kill('SIGKILL') + } + }, 15_000) + + it('preserves descendants that predate the turn', async () => { + const root = spawn( + process.execPath, + [ + '-e', + `const { spawn } = require('node:child_process'); + const persistent = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' }); + process.stdout.write(String(persistent.pid) + '\\n'); + process.stdin.once('data', () => { + const child = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 60000)'], { stdio: 'ignore' }); + process.stdout.write(String(child.pid) + '\\n'); + }); + setTimeout(() => {}, 60000);` + ], + { stdio: ['pipe', 'pipe', 'pipe'] } + ) + let persistentPid = 0 + let commandPid = 0 + try { + persistentPid = Number(await nextLine(root)) + const baseline = await captureCodexTurnProcesses(root.pid!) + root.stdin.write('start\n') + commandPid = Number(await nextLine(root)) + + await expect(terminateCodexTurnProcesses(root.pid!, baseline)).resolves.toBe(true) + expect(processExists(persistentPid)).toBe(true) + expect(processExists(commandPid)).toBe(false) + } finally { + if (persistentPid > 0 && processExists(persistentPid)) { + process.kill(persistentPid, 'SIGKILL') + } + if (commandPid > 0 && processExists(commandPid)) { + process.kill(commandPid, 'SIGKILL') + } + root.kill('SIGKILL') + } + }, 15_000) +}) diff --git a/src/main/codex/codex-structured-turn-processes.ts b/src/main/codex/codex-structured-turn-processes.ts new file mode 100644 index 00000000000..f266d795292 --- /dev/null +++ b/src/main/codex/codex-structured-turn-processes.ts @@ -0,0 +1,85 @@ +import { captureDescendantSnapshot, type DescendantSnapshot } from '../pty-descendant-termination' +import { terminateDescendantSnapshotAndWait } from '../pty-descendant-exit-verification' +import { queryWindowsProcessDescendants } from '../providers/windows-foreground-process-rows' +import { terminateWindowsProcessTree } from '../windows-process-tree-kill' + +export type CodexTurnProcessSnapshot = + | { platform: 'posix'; snapshot: DescendantSnapshot } + | { platform: 'win32'; identities: ReadonlyMap } + +function windowsIdentity(row: { + ppid: number + name: string + command: string + executablePath: string +}): string { + return [row.ppid, row.name, row.command, row.executablePath].join('\0') +} + +export async function captureCodexTurnProcesses( + rootPid: number +): Promise { + if (process.platform === 'win32') { + const descendants = await queryWindowsProcessDescendants(rootPid, { fresh: true }) + return descendants + ? { + platform: 'win32', + identities: new Map(descendants.map((row) => [row.pid, windowsIdentity(row)])) + } + : null + } + const snapshot = await captureDescendantSnapshot(rootPid) + return snapshot ? { platform: 'posix', snapshot } : null +} + +function addedPosixDescendants( + baseline: DescendantSnapshot, + current: DescendantSnapshot +): DescendantSnapshot { + const baselineRows = new Map(baseline.descendants.map((row) => [row.pid, row])) + return { + ...current, + descendants: current.descendants.filter((row) => { + const prior = baselineRows.get(row.pid) + return prior?.startedAt !== row.startedAt || prior.pgid !== row.pgid + }) + } +} + +async function terminateWindowsAddedProcesses( + rootPid: number, + baseline: ReadonlyMap +): Promise { + const current = await queryWindowsProcessDescendants(rootPid, { fresh: true }) + if (!current) { + return false + } + const added = current.filter((row) => baseline.get(row.pid) !== windowsIdentity(row)) + const addedPids = new Set(added.map((row) => row.pid)) + const roots = added.filter((row) => !addedPids.has(row.ppid)) + await Promise.all(roots.map((row) => terminateWindowsProcessTree(row.pid))) + const targetIdentities = new Map(added.map((row) => [row.pid, windowsIdentity(row)])) + const remaining = await queryWindowsProcessDescendants(rootPid, { fresh: true }) + return ( + remaining !== null && + remaining.every((row) => targetIdentities.get(row.pid) !== windowsIdentity(row)) + ) +} + +export async function terminateCodexTurnProcesses( + rootPid: number, + baseline: CodexTurnProcessSnapshot | null +): Promise { + if (!baseline) { + return false + } + if (baseline.platform === 'win32') { + return terminateWindowsAddedProcesses(rootPid, baseline.identities) + } + const current = await captureDescendantSnapshot(rootPid) + if (!current) { + return false + } + const added = addedPosixDescendants(baseline.snapshot, current) + return terminateDescendantSnapshotAndWait(added) +} diff --git a/src/main/pty-descendant-exit-verification.ts b/src/main/pty-descendant-exit-verification.ts new file mode 100644 index 00000000000..c0ed223704a --- /dev/null +++ b/src/main/pty-descendant-exit-verification.ts @@ -0,0 +1,78 @@ +import { + DESCENDANT_KILL_GRACE_MS, + DESCENDANT_SNAPSHOT_TIMEOUT_MS, + hasUnambiguousStartIdentity, + readProcessTable, + readProcessTableBeforeDeadline, + sendDescendantSignal, + type DescendantSnapshot, + type ProcessTableRow, + type TerminateDeps +} from './pty-descendant-termination' + +export const DESCENDANT_KILL_VERIFY_MS = 3_500 + +function waitForDelay(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms) + timer.unref?.() + }) +} + +function matchingSnapshotRows( + snapshot: DescendantSnapshot, + table: readonly ProcessTableRow[] +): ProcessTableRow[] { + const expected = new Map(snapshot.descendants.map((row) => [row.pid, row])) + return table.filter((live) => { + const row = expected.get(live.pid) + return row?.startedAt === live.startedAt && row.pgid === live.pgid + }) +} + +type VerificationDeps = TerminateDeps & { + verifyMs?: number +} + +/** An unreadable process table is never proof that a stopped descendant exited. */ +export async function terminateDescendantSnapshotAndWait( + snapshot: DescendantSnapshot, + deps: VerificationDeps = {} +): Promise { + const sendSignal = deps.sendSignal ?? sendDescendantSignal + const readTable = deps.readTable ?? readProcessTable + const graceMs = deps.graceMs ?? DESCENDANT_KILL_GRACE_MS + const verifyMs = deps.verifyMs ?? DESCENDANT_KILL_VERIFY_MS + const deadline = Date.now() + verifyMs + for (const row of snapshot.descendants) { + sendSignal(row.pid, 'SIGTERM') + } + let forced = false + while (Date.now() < deadline) { + const capture = await readProcessTableBeforeDeadline( + readTable, + deps.timeoutMs ?? DESCENDANT_SNAPSHOT_TIMEOUT_MS + ) + if (!capture) { + return false + } + const live = matchingSnapshotRows(snapshot, capture.rows) + if (live.length === 0) { + return true + } + if (!forced && Date.now() >= deadline - verifyMs + graceMs) { + forced = true + for (const row of live) { + if (hasUnambiguousStartIdentity(row, snapshot.capturedAtMs)) { + sendSignal(row.pid, 'SIGKILL') + } + } + } + await waitForDelay(50) + } + const finalCapture = await readProcessTableBeforeDeadline( + readTable, + deps.timeoutMs ?? DESCENDANT_SNAPSHOT_TIMEOUT_MS + ) + return finalCapture !== null && matchingSnapshotRows(snapshot, finalCapture.rows).length === 0 +} diff --git a/src/main/pty-descendant-termination.test.ts b/src/main/pty-descendant-termination.test.ts index c6941e1cb28..2f0ae698033 100644 --- a/src/main/pty-descendant-termination.test.ts +++ b/src/main/pty-descendant-termination.test.ts @@ -15,6 +15,7 @@ import { type ProcessTableCapture, type ProcessTableRow } from './pty-descendant-termination' +import { terminateDescendantSnapshotAndWait } from './pty-descendant-exit-verification' const CAPTURED_AT_MS = Date.parse('Tue Jul 14 12:00:00 2026') @@ -299,6 +300,49 @@ describe('terminateDescendantSnapshot', () => { }) }) +describe('terminateDescendantSnapshotAndWait', () => { + beforeEach(() => { + vi.useFakeTimers() + }) + afterEach(() => { + vi.useRealTimers() + }) + + it('escalates an identity-matched survivor and verifies its exit', async () => { + const survivor = row(20, 10, 20) + const sendSignal = vi.fn() + const readTable = vi + .fn() + .mockResolvedValueOnce(tableCapture([survivor])) + .mockResolvedValueOnce(tableCapture([])) + + const pending = terminateDescendantSnapshotAndWait(snapshot([survivor]), { + sendSignal, + readTable, + graceMs: 0, + verifyMs: 200 + }) + await vi.advanceTimersByTimeAsync(50) + + await expect(pending).resolves.toBe(true) + expect(sendSignal.mock.calls).toEqual([ + [20, 'SIGTERM'], + [20, 'SIGKILL'] + ]) + }) + + it('does not claim exit when the verification table is unavailable', async () => { + const sendSignal = vi.fn() + const result = await terminateDescendantSnapshotAndWait(snapshot([row(20, 10, 20)]), { + sendSignal, + readTable: vi.fn().mockRejectedValue(new Error('ps exploded')) + }) + + expect(result).toBe(false) + expect(sendSignal).toHaveBeenCalledWith(20, 'SIGTERM') + }) +}) + describe('createProcessTableSnapshotReader', () => { it('coalesces same-turn teardown requests onto one fresh scan', async () => { const capture = tableCapture([row(10, 1, 10)]) diff --git a/src/main/pty-descendant-termination.ts b/src/main/pty-descendant-termination.ts index 34ad425b86c..0df23277d7e 100644 --- a/src/main/pty-descendant-termination.ts +++ b/src/main/pty-descendant-termination.ts @@ -119,9 +119,9 @@ export function createProcessTableSnapshotReader( } } -const readProcessTable = createProcessTableSnapshotReader(readFreshProcessTable) +export const readProcessTable = createProcessTableSnapshotReader(readFreshProcessTable) -function readProcessTableBeforeDeadline( +export function readProcessTableBeforeDeadline( readTable: ProcessTableReader, timeoutMs: number ): Promise { @@ -284,7 +284,7 @@ export async function killWithDescendantSweep( } } -function defaultSendSignal(pid: number, signal: NodeJS.Signals): void { +export function sendDescendantSignal(pid: number, signal: NodeJS.Signals): void { try { process.kill(pid, signal) } catch { @@ -292,14 +292,14 @@ function defaultSendSignal(pid: number, signal: NodeJS.Signals): void { } } -type TerminateDeps = { +export type TerminateDeps = { readTable?: ProcessTableReader sendSignal?: SignalSender graceMs?: number timeoutMs?: number } -function hasUnambiguousStartIdentity(row: ProcessTableRow, capturedAtMs: number): boolean { +export function hasUnambiguousStartIdentity(row: ProcessTableRow, capturedAtMs: number): boolean { const startedAtMs = Date.parse(row.startedAt) if (!Number.isFinite(startedAtMs)) { return false @@ -319,7 +319,7 @@ export function terminateDescendantSnapshot( snapshot: DescendantSnapshot, deps: TerminateDeps = {} ): void { - const sendSignal = deps.sendSignal ?? defaultSendSignal + const sendSignal = deps.sendSignal ?? sendDescendantSignal const readTable = deps.readTable ?? readProcessTable for (const row of snapshot.descendants) { sendSignal(row.pid, 'SIGTERM')