diff --git a/config/scripts/ci-shard-timings.json b/config/scripts/ci-shard-timings.json index 795f6070bafb..9e2fbaedf3fc 100644 --- a/config/scripts/ci-shard-timings.json +++ b/config/scripts/ci-shard-timings.json @@ -1300,7 +1300,6 @@ "src/main/codex/codex-structured-session-shutdown.test.ts": 471, "src/main/codex/codex-structured-thread-goal.test.ts": 73, "src/main/codex/codex-structured-thread-open.test.ts": 300, - "src/main/codex/codex-structured-turn-processes.integration.test.ts": 194, "src/main/codex/codex-subagent-execution-projection.test.ts": 202, "src/main/codex/codex-subagent-executions.test.ts": 71, "src/main/codex/codex-subagent-roster.test.ts": 118, diff --git a/src/main/codex/codex-prompt-registry-retention.test.ts b/src/main/codex/codex-prompt-registry-retention.test.ts index 4757d731dcf7..5a1dbca11bad 100644 --- a/src/main/codex/codex-prompt-registry-retention.test.ts +++ b/src/main/codex/codex-prompt-registry-retention.test.ts @@ -83,7 +83,7 @@ describe('Codex prompt claim lifetime', () => { await collect() expect(old.deref()).toBeUndefined() expect(registry.find(replacement.itemId)).toBe(replacement.prompt) - expect(registry.ownsBoundClaim(claim, replacement.itemId, 'thread', 'new-turn')).toBe(true) + expect(registry.ownsClaim(claim)).toBe(true) registry.clearTurn('thread', 'new-turn') expect(registry.ownsClaim(claim)).toBe(false) }) diff --git a/src/main/codex/codex-prompt-registry.ts b/src/main/codex/codex-prompt-registry.ts index eda1fa0bf306..b77195948864 100644 --- a/src/main/codex/codex-prompt-registry.ts +++ b/src/main/codex/codex-prompt-registry.ts @@ -191,23 +191,6 @@ export class CodexPromptRegistry { return this.claims.get(claim.prompt) === claim && this.find(claim.itemId) === claim.prompt } - ownsBoundClaim( - claim: CodexPromptClaim, - journalItemId: string, - threadId: string, - turnId: string - ): boolean { - return ( - claim.itemId === journalItemId && - this.claims.get(claim.prompt) === claim && - this.journalItemIds.get(journalItemId) === - this.address(claim.prompt.threadId, claim.prompt.promptKey) && - this.boundPrompts.get(journalItemId) === claim.prompt && - claim.prompt.threadId === threadId && - codexPromptMatchesTurn(claim.prompt, turnId) - ) - } - releaseClaim(claim: CodexPromptClaim): void { if (this.claims.get(claim.prompt) === claim) { this.claims.delete(claim.prompt) diff --git a/src/main/codex/codex-structured-dispatch-boundary.test.ts b/src/main/codex/codex-structured-dispatch-boundary.test.ts index 5d4f0912bccb..5ad36626c46b 100644 --- a/src/main/codex/codex-structured-dispatch-boundary.test.ts +++ b/src/main/codex/codex-structured-dispatch-boundary.test.ts @@ -6,41 +6,25 @@ import { fakeCodexAppServer } from './codex-structured-dispatch-test-support' -it('checks continuation authority after process capture and before writing turn/start', async () => { - const capturing = Promise.withResolvers() - const captured = Promise.withResolvers() +it('checks continuation authority before writing turn/start', async () => { const codex = fakeCodexAppServer() - const adapter = await acquiredCodexAdapter({ - codex, - settlements: [], - captureTurnProcesses: () => { - capturing.resolve() - return captured.promise - } - }) - let superseded = false + const adapter = await acquiredCodexAdapter({ codex, settlements: [] }) const beforeDispatch = vi.fn(async () => { - if (superseded) { - throw new AgentSessionPreDispatchError('agent_session_restart_work_superseded') - } + throw new AgentSessionPreDispatchError('agent_session_restart_work_superseded') }) try { - const result = adapter.dispatch({ - sessionId: 'session-1', - clientMessageId: 'continuation', - body: CODEX_TEST_USER_MESSAGE, - fence: 7, - beforeDispatch - }) - const verdict = result.catch((error: unknown) => error) - await capturing.promise - superseded = true - captured.resolve(null) - expect(await verdict).toBeInstanceOf(AgentSessionPreDispatchError) + await expect( + adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'continuation', + body: CODEX_TEST_USER_MESSAGE, + fence: 7, + beforeDispatch + }) + ).rejects.toBeInstanceOf(AgentSessionPreDispatchError) expect(beforeDispatch).toHaveBeenCalledOnce() expect(codex.connections[0]?.calls.some((call) => call.method === 'turn/start')).toBe(false) } finally { - captured.resolve(null) await adapter.closeAll() } }) diff --git a/src/main/codex/codex-structured-dispatch-echo.ts b/src/main/codex/codex-structured-dispatch-echo.ts index 0d3794b5136b..ef84501adcb2 100644 --- a/src/main/codex/codex-structured-dispatch-echo.ts +++ b/src/main/codex/codex-structured-dispatch-echo.ts @@ -43,7 +43,8 @@ export type CodexDispatchEchoes = { answeredUnopenedTurn: (threadId: string, openTurnIds: ReadonlySet) => string | null /** * Records a turn's end and returns the sends bound to it that it settles: all of them unless it - * completed, which echoes its pending input first, so one it never echoed waits for recovery. + * completed, which echoes its pending input first, so one it never echoed waits for recovery. An + * interrupt withdraws an un-echoed send, steered or the turn's own input: neither reached history. */ endTurn: (threadId: string, turnId: string, end: CodexTurnEnd) => string[] /** Submission origin for this exact send, retained until its echo settles it. */ diff --git a/src/main/codex/codex-structured-dispatch-test-support.ts b/src/main/codex/codex-structured-dispatch-test-support.ts index 86f19349ee25..2576c775e214 100644 --- a/src/main/codex/codex-structured-dispatch-test-support.ts +++ b/src/main/codex/codex-structured-dispatch-test-support.ts @@ -84,7 +84,6 @@ export async function acquiredCodexAdapter(input: { codex: ReturnType settlements: LateSettlement[] sink?: StructuredAgentSessionEventSink - captureTurnProcesses?: CodexStructuredSessionAdapterDeps['captureTurnProcesses'] requestTimeoutMs?: number }): Promise { const adapter = new CodexStructuredSessionAdapter({ @@ -97,9 +96,6 @@ export async function acquiredCodexAdapter(input: { }), openConnection: input.codex.openConnection, readProcessStartTime: async () => 1_700_000_000_000, - captureTurnProcesses: input.captureTurnProcesses ?? (async () => null), - // A Stop must never reach for real processes on this machine under the fake pid. - terminateTurnProcesses: async () => true, now: () => 1_700_000_000_500, onDispatchSettledLate: (settlement) => input.settlements.push(settlement), ...(input.requestTimeoutMs === undefined ? {} : { requestTimeoutMs: input.requestTimeoutMs }) diff --git a/src/main/codex/codex-structured-prompt-ownership.test.ts b/src/main/codex/codex-structured-prompt-ownership.test.ts index 7dd048dd08af..ef1f8cc426f8 100644 --- a/src/main/codex/codex-structured-prompt-ownership.test.ts +++ b/src/main/codex/codex-structured-prompt-ownership.test.ts @@ -277,8 +277,7 @@ describe('Codex live prompt ownership', () => { const codex = fakeCodex({ 'turn/interrupt': () => completeTurn(codex, 'thread-child', 'child-turn') }) - const terminateTurnProcesses = vi.fn(async () => true) - const adapter = adapterFor(codex, {}, [], { terminateTurnProcesses }) + const adapter = adapterFor(codex) await adapter.acquire({ identity: identityFor('session-1'), fence: 7, @@ -298,7 +297,6 @@ describe('Codex live prompt ownership', () => { method: 'turn/interrupt', params: { threadId: 'thread-child', turnId: 'child-turn' } }) - expect(terminateTurnProcesses).not.toHaveBeenCalled() expect(codex.connections[0]?.closed).toBe(false) }) @@ -553,7 +551,7 @@ describe('Codex live prompt ownership', () => { expect(commit).not.toHaveBeenCalled() }) - it('does not report success when a deferred provider completion is backpressured', async () => { + it("confirms on Codex's answer and keeps the prompt's claim when the turn's end is backpressured", async () => { const recorded = lifecycleRecorder(true, false) const codex = fakeCodex({ 'turn/interrupt': () => { @@ -580,7 +578,7 @@ describe('Codex live prompt ownership', () => { fence: 7, prompt: { itemId: promptItemId } }) - ).rejects.toThrow(/deferred turn completion lifecycle was not admitted/) + ).resolves.toEqual({ cancelled: true }) const commit = vi.fn(async () => undefined) await expect( adapter.answerPrompt({ @@ -597,7 +595,7 @@ describe('Codex live prompt ownership', () => { await adapter.closeSession('session-1') }) - it('defers only the matching thread and emits its terminal event before cancel resolves', async () => { + it("publishes every thread's turn end as Codex sends it, ahead of the Stop's answer", async () => { const interruptGate = deferred() const events: CodexStructuredSessionEvent[] = [] const codex = fakeCodex({ @@ -610,18 +608,13 @@ describe('Codex live prompt ownership', () => { const adapter = await acquired(codex, {}, events) registerPrompt(adapter, codex, 'child-prompt', 'thread-child') - const cancellation = adapter - .cancelTurn({ - sessionId: 'session-1', - turnId: 'turn-1', - fence: 7, - prompt: { itemId: 'child-prompt' } - }) - .then((result) => { - expect(completionThreads(events)).toEqual([THREAD_ID, 'thread-child']) - return result - }) - await vi.waitFor(() => expect(completionThreads(events)).toEqual([THREAD_ID])) + const cancellation = adapter.cancelTurn({ + sessionId: 'session-1', + turnId: 'turn-1', + fence: 7, + prompt: { itemId: 'child-prompt' } + }) + await vi.waitFor(() => expect(completionThreads(events)).toEqual([THREAD_ID, 'thread-child'])) interruptGate.resolve() await expect(cancellation).resolves.toEqual({ cancelled: true }) diff --git a/src/main/codex/codex-structured-prompt-ownership.ts b/src/main/codex/codex-structured-prompt-ownership.ts index 69c9556bade8..02633874d03b 100644 --- a/src/main/codex/codex-structured-prompt-ownership.ts +++ b/src/main/codex/codex-structured-prompt-ownership.ts @@ -11,7 +11,7 @@ import { type CodexPreparedAnswer } from './codex-structured-prompt-replies' import { requireLiveCodexSession, type CodexSession } from './codex-structured-session-state' -import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation' +import { interruptCodexTurn } from './codex-structured-turn-cancellation' type CancelInput = Parameters[0] type AnswerInput = Parameters[0] @@ -35,27 +35,23 @@ async function cancelCodexConversation( input: Parameters[0], session: CodexSession ): Promise { - const { request, sessions, cancellation } = input + const { request, sessions, requestTimeoutMs } = input const liveTurnId = request.resolveLiveTurnId?.() ?? null // A turn the journal shows that Codex has not started yet (a compaction's) has nothing to stop. const turnId = liveTurnId === null ? ([...(session.activeTurnIds ?? [])].at(-1) ?? (await openedAnsweredTurn(session))) : providerTurnId(session, liveTurnId) - if (!turnId) { + // The wait for the turn to open can outlive the session it began on. + if ( + !turnId || + sessions.get(request.sessionId) !== session || + session.ended || + session.fence !== request.fence + ) { return { cancelled: false } } - const acquisitionGeneration = session.acquisitionGeneration - return cancellation.cancel( - session, - session.threadId, - turnId, - () => - sessions.get(request.sessionId) === session && - !session.ended && - session.fence === request.fence && - session.acquisitionGeneration === acquisitionGeneration - ) + return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }) } /** The turn Codex answered a send into and has not opened yet, once it opens; null when it ends, @@ -74,9 +70,9 @@ async function openedAnsweredTurn(session: CodexSession): Promise export async function cancelCodexStructuredTurn(input: { request: CancelInput sessions: Map - cancellation: CodexStructuredTurnCancellation + requestTimeoutMs?: number }): Promise { - const { request, sessions, cancellation } = input + const { request, sessions, requestTimeoutMs } = input const session = requireLiveCodexSession(sessions, request.sessionId) const prompt = request.prompt const requestedTurnId = request.turnId @@ -88,12 +84,11 @@ export async function cancelCodexStructuredTurn(input: { return { cancelled: false } } if (!prompt) { - return cancellation.cancel(session, session.threadId, turnId) + return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }) } if (session.fence !== request.fence) { return { cancelled: false } } - const acquisitionGeneration = session.acquisitionGeneration const claim = session.prompts.claimBound(prompt.itemId) const promptTurnId = claim?.prompt.turnId if (!claim || !promptTurnId) { @@ -102,25 +97,18 @@ export async function cancelCodexStructuredTurn(input: { } return { cancelled: false } } - const isCurrent = (): boolean => - sessions.get(request.sessionId) === session && - !session.ended && - session.fence === request.fence && - session.acquisitionGeneration === acquisitionGeneration && - providerTurnId(session, requestedTurnId) === turnId && - session.prompts.ownsBoundClaim(claim, prompt.itemId, claim.prompt.threadId, promptTurnId) let interruptConfirmed = false try { - const result = await cancellation.cancel( + const result = await interruptCodexTurn({ session, - claim.prompt.threadId, - promptTurnId, - isCurrent, - () => { + threadId: claim.prompt.threadId, + turnId: promptTurnId, + requestTimeoutMs, + onConfirmed: () => { interruptConfirmed = true return session.translator?.cancelPrompt(prompt.itemId) ?? { accepted: true } } - ) + }) if (!result.cancelled) { session.prompts.releaseClaim(claim) } diff --git a/src/main/codex/codex-structured-provider-events.ts b/src/main/codex/codex-structured-provider-events.ts index 5dc608fbb59b..8b59209a7683 100644 --- a/src/main/codex/codex-structured-provider-events.ts +++ b/src/main/codex/codex-structured-provider-events.ts @@ -4,14 +4,13 @@ import type { CodexJournalTranslationAdmission } from './codex-structured-journa import * as codexRewind from './codex-structured-rewind' import type { CodexSession, CodexStructuredSessionEvent } from './codex-structured-session-state' import { readCodexThreadId } from './codex-structured-thread-facts' -import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation' type EmitCodexEvent = ( session: CodexSession, event: CodexStructuredSessionEvent ) => CodexJournalTranslationAdmission -/** One live notification's journal entry: rewind bookkeeping, cancellation deferral, delivery. */ +/** One live notification's journal entry: rewind bookkeeping, delivery. */ export function translateCodexNotification(input: { sessionId: string session: CodexSession @@ -19,15 +18,11 @@ export function translateCodexNotification(input: { params: unknown observedAt?: number dispatchSequenceAtReceipt?: number - turnCancellation: Pick emit: EmitCodexEvent }): CodexJournalTranslationAdmission { const { sessionId, session, method, params, observedAt, dispatchSequenceAtReceipt } = input codexRewind.observeCodexRewindActivity(session, method, params) session.turnOpenWaits.observe(session.threadId, method, params) - if (input.turnCancellation.handleNotification(sessionId, session, method, params, observedAt)) { - return { accepted: true } - } return deliverCodexNotification( sessionId, session, diff --git a/src/main/codex/codex-structured-session-acquire.ts b/src/main/codex/codex-structured-session-acquire.ts index e765955d53ca..73c1653aaddb 100644 --- a/src/main/codex/codex-structured-session-acquire.ts +++ b/src/main/codex/codex-structured-session-acquire.ts @@ -41,7 +41,6 @@ import { type CodexSession, type CodexStructuredSessionAdapterDeps } from './codex-structured-session-state' -import type { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation' import type { CodexStructuredSessionTeardown } from './codex-structured-session-teardown' import type { CodexStructuredNotificationRetry } from './codex-structured-notification-retry' import type { deliverCodexServerRequest } from './codex-structured-provider-events' @@ -53,7 +52,6 @@ export async function acquireCodexStructuredSession(input: { deps: CodexStructuredSessionAdapterDeps sessions: Map acquisitions: CodexAcquisitionRegistry - turnCancellation: CodexStructuredTurnCancellation notificationRetries: CodexStructuredNotificationRetry deliver: ( acquisition: CodexAcquisitionAttempt['window'], @@ -68,14 +66,7 @@ export async function acquireCodexStructuredSession(input: { handleUnhandledFrame: (sessionId: string, kind: string, payload: unknown) => void forceCloseUnexpected: CodexStructuredSessionTeardown['forceCloseUnexpected'] }): Promise { - const { - input: acquireInput, - deps, - sessions, - acquisitions, - turnCancellation, - notificationRetries - } = input + const { input: acquireInput, deps, sessions, acquisitions, notificationRetries } = input const sessionId = acquireInput.identity.sessionId const { previousAttempt, attempt } = acquisitions.start(sessionId) const acquisition = attempt.window @@ -271,7 +262,6 @@ export async function acquireCodexStructuredSession(input: { ?.supportsFastMode }) } - turnCancellation.register(session) sessions.set(sessionId, session) for (const event of acquisition.drain()) { event() diff --git a/src/main/codex/codex-structured-session-adapter-fixture.ts b/src/main/codex/codex-structured-session-adapter-fixture.ts index 27e6eadb6c57..45c97f000661 100644 --- a/src/main/codex/codex-structured-session-adapter-fixture.ts +++ b/src/main/codex/codex-structured-session-adapter-fixture.ts @@ -11,7 +11,6 @@ import type { import { CodexStructuredSessionAdapter, type CodexStructuredLaunch, - type CodexStructuredSessionAdapterDeps, type CodexStructuredSessionEvent } from './codex-structured-session-adapter' @@ -105,10 +104,7 @@ export function answerWithOpenedTurn( export function adapterFor( codex: ReturnType, launch: Partial = {}, - events: CodexStructuredSessionEvent[] = [], - processControl: Partial< - Pick - > = {} + events: CodexStructuredSessionEvent[] = [] ): CodexStructuredSessionAdapter { let acquisitionGeneration = 0 return new CodexStructuredSessionAdapter({ @@ -123,11 +119,8 @@ export function adapterFor( 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, now: () => 1_700_000_000_500, - mintAcquisitionGeneration: () => `generation-${++acquisitionGeneration}`, - ...processControl + mintAcquisitionGeneration: () => `generation-${++acquisitionGeneration}` }) } diff --git a/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts b/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts index bcfd2f733f06..e3211f320e85 100644 --- a/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts +++ b/src/main/codex/codex-structured-session-adapter-lifecycle.test.ts @@ -14,7 +14,6 @@ import type { StructuredAgentSessionEventSink } from '../native-chat/agent-sessi import { CodexStructuredSessionAdapter, type CodexStructuredLaunch, - type CodexStructuredSessionAdapterDeps, type CodexStructuredSessionEvent } from './codex-structured-session-adapter' @@ -99,10 +98,7 @@ function fakeCodex(routes: Record = {}): { function adapterFor( codex: ReturnType, launch: Partial = {}, - events: CodexStructuredSessionEvent[] = [], - processControl: Partial< - Pick - > = {} + events: CodexStructuredSessionEvent[] = [] ): CodexStructuredSessionAdapter { let acquisitionGeneration = 0 return new CodexStructuredSessionAdapter({ @@ -117,11 +113,8 @@ function adapterFor( 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, now: () => 1_700_000_000_500, - mintAcquisitionGeneration: () => `generation-${++acquisitionGeneration}`, - ...processControl + mintAcquisitionGeneration: () => `generation-${++acquisitionGeneration}` }) } diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 82f927e0e86a..e0b1a93b395a 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -33,7 +33,6 @@ import { deliverCodexUnhandledFrame, translateCodexNotification } from './codex-structured-provider-events' -import { CodexStructuredTurnCancellation } from './codex-structured-turn-cancellation' import { codexDispatchRejection, settleCodexSendsInEndedTurn @@ -55,7 +54,6 @@ export type { export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdapter { private readonly sessions = new Map() private readonly acquisitions = new CodexAcquisitionRegistry() - private readonly turnCancellation: CodexStructuredTurnCancellation private readonly notificationRetries: ReturnType private readonly teardown: CodexStructuredSessionTeardown @@ -70,7 +68,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap params, observedAt, dispatchSequenceAtReceipt, - turnCancellation: this.turnCancellation, emit: (current, event) => this.emit(current, event) }) }) @@ -83,25 +80,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap : {}), forgetNotificationRetries: (sessionId) => this.notificationRetries.clear(sessionId, null) }) - this.turnCancellation = new CodexStructuredTurnCancellation({ - captureTurnProcesses: deps.captureTurnProcesses, - terminateTurnProcesses: deps.terminateTurnProcesses, - requestTimeoutMs: deps.requestTimeoutMs, - emit: (session, event) => { - const admission = this.emit(session, event) - if (!admission.accepted && event.type === 'notification') { - const { sessionId, method, params, observedAt, dispatchSequenceAtReceipt } = event - this.notificationRetries.handle( - sessionId, - method, - params, - observedAt, - dispatchSequenceAtReceipt - ) - } - return admission - } - }) } supportsLocation = (location: Parameters[0]): boolean => @@ -113,7 +91,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap deps: this.deps, sessions: this.sessions, acquisitions: this.acquisitions, - turnCancellation: this.turnCancellation, notificationRetries: this.notificationRetries, deliver: (acquisition, sessionId, event, retainedBytes) => this.deliver(acquisition, sessionId, event, retainedBytes), @@ -222,7 +199,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap const session = this.session(input.sessionId) session.dispatchPending = true try { - await this.turnCancellation.captureBaseline(session) await input.beforeDispatch?.() return await dispatchCodexTurn(session, input, this.deps.requestTimeoutMs) } finally { @@ -234,7 +210,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap cancelCodexStructuredTurn({ request, sessions: this.sessions, - cancellation: this.turnCancellation + requestTimeoutMs: this.deps.requestTimeoutMs }) rewindSupport: NonNullable = (sessionId) => @@ -253,7 +229,6 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap const session = this.session(input.sessionId) session.translator?.beginCommand(input.command) try { - await this.turnCancellation.captureBaseline(session) await session.connection.request( 'thread/compact/start', { threadId: session.threadId }, diff --git a/src/main/codex/codex-structured-session-cancel.test.ts b/src/main/codex/codex-structured-session-cancel.test.ts index 2d6c50adae46..cd30bfa2e6d1 100644 --- a/src/main/codex/codex-structured-session-cancel.test.ts +++ b/src/main/codex/codex-structured-session-cancel.test.ts @@ -1,4 +1,35 @@ import { describe, expect, it, vi } from 'vitest' + +// Every way Orca reads or kills processes, never settling: a Stop that reaches for one hangs. +const processWork = vi.hoisted(() => { + const never = (): Promise => new Promise(() => {}) + return { + captureDescendantSnapshot: vi.fn(async () => ({ + rootPgid: null, + descendants: [], + capturedAtMs: 0 + })), + terminateDescendantSnapshotAndWait: vi.fn(never), + queryWindowsProcessDescendants: vi.fn(never), + terminateWindowsProcessTree: vi.fn(never) + } +}) +vi.mock('../pty-descendant-termination', async (importOriginal) => ({ + ...(await importOriginal>()), + captureDescendantSnapshot: processWork.captureDescendantSnapshot +})) +vi.mock('../pty-descendant-exit-verification', async (importOriginal) => ({ + ...(await importOriginal>()), + terminateDescendantSnapshotAndWait: processWork.terminateDescendantSnapshotAndWait +})) +vi.mock('../providers/windows-foreground-process-rows', async (importOriginal) => ({ + ...(await importOriginal>()), + queryWindowsProcessDescendants: processWork.queryWindowsProcessDescendants +})) +vi.mock('../windows-process-tree-kill', async (importOriginal) => ({ + ...(await importOriginal>()), + terminateWindowsProcessTree: processWork.terminateWindowsProcessTree +})) import type { AgentJournalMessageItem, AgentSessionJournalIdentity @@ -79,12 +110,7 @@ function fakeCodex(): { async function acquired( codex: ReturnType, events: CodexStructuredSessionEvent[] = [], - processControl: Partial< - Pick< - CodexStructuredSessionAdapterDeps, - 'captureTurnProcesses' | 'terminateTurnProcesses' | 'now' - > - > = {} + overrides: Partial> = {} ): Promise { const adapter = new CodexStructuredSessionAdapter({ resolveLaunch: async () => ({ @@ -97,9 +123,7 @@ async function acquired( 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 + ...overrides }) await adapter.acquire({ identity: identity(), fence: 7, spawnToken: 'spawn-9' }) return adapter @@ -167,71 +191,6 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => { ).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, unconfirmed: true }) - expect(events).toContainEqual(expect.objectContaining({ method: 'turn/completed' })) - }) - it('accepts an immediate resend after verified interruption', async () => { let nextTurn = 0 const codex = fakeCodex() @@ -270,46 +229,79 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => { }) }) - it('keeps the receipt time of a completion deferred behind physical termination', async () => { + it("publishes the turn's end when the interrupt receipt fails", async () => { const events: CodexStructuredSessionEvent[] = [] - let clock = 5_000 - let finishTermination!: (terminated: boolean) => void - const termination = new Promise((resolve) => { - finishTermination = resolve - }) const codex = fakeCodex() codex.routes['turn/interrupt'] = () => { completeTurn(codex) - return {} + throw new Error('interrupt receipt lost') } - const adapter = await acquired(codex, events, { - terminateTurnProcesses: async () => termination, - now: () => clock - }) + const adapter = await acquired(codex, events) - 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')) - clock = 9_000 - finishTermination(true) - await expect(pending).resolves.toEqual({ cancelled: true }) - - expect(events.at(-1)).toMatchObject({ method: 'turn/completed', observedAt: 5_000 }) + await expect( + adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + ).rejects.toThrow('interrupt receipt lost') + expect(events).toContainEqual(expect.objectContaining({ method: 'turn/completed' })) }) +}) - it('does not strand a deferred completion when the interrupt receipt fails', async () => { +// Codex keeps a turn's background terminals alive across an interrupt and kills its one-shot +// commands itself, so a Stop is the interrupt alone. +describe('a Codex Stop is the interrupt alone', () => { + async function stopAfterASend() { const events: CodexStructuredSessionEvent[] = [] + const answer = Promise.withResolvers() const codex = fakeCodex() - codex.routes['turn/interrupt'] = () => { - completeTurn(codex) - throw new Error('interrupt receipt lost') + codex.routes['turn/start'] = () => { + codex.connections[0].handlers.onNotification?.('turn/started', { + threadId: THREAD_ID, + turn: { id: 'turn-1' } + }) + return { turn: { id: 'turn-1' } } } - const adapter = await acquired(codex, events, { - terminateTurnProcesses: async () => true + codex.routes['turn/interrupt'] = () => answer.promise + const adapter = await acquired(codex, events) + await adapter.dispatch({ + sessionId: 'session-1', + clientMessageId: 'client-1', + body: USER_MESSAGE, + fence: 7 }) + const stopped = adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) + await vi.waitFor(() => expect(codex.connections[0].calls.at(-1)?.method).toBe('turn/interrupt')) + // Codex answers the interrupt, then sends the turn's end; one read can carry both. + const answerThenEnd = (): void => { + answer.resolve({}) + completeTurn(codex) + } + return { events, stopped, answerThenEnd } + } - await expect( - adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).rejects.toThrow('interrupt receipt lost') - expect(events).toContainEqual(expect.objectContaining({ method: 'turn/completed' })) + it('publishes the interrupted end the moment Codex sends it', async () => { + const { events, stopped, answerThenEnd } = await stopAfterASend() + + answerThenEnd() + + expect(events).toContainEqual( + expect.objectContaining({ + method: 'turn/completed', + params: expect.objectContaining({ turn: { id: 'turn-1', status: 'interrupted' } }) + }) + ) + await expect(stopped).resolves.toEqual({ cancelled: true }) + }) + + it('reads and kills no processes, from the send through the Stop', async () => { + const { stopped, answerThenEnd } = await stopAfterASend() + + answerThenEnd() + + for (const work of Object.values(processWork)) { + expect(work).not.toHaveBeenCalled() + } + await expect(stopped).resolves.toEqual({ cancelled: true }) + for (const work of Object.values(processWork)) { + expect(work).not.toHaveBeenCalled() + } }) }) diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 627e33dbfa9f..fcfb120fe821 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -19,7 +19,6 @@ import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-work-evidence' import type { CodexBackgroundTaskTracker } from './codex-background-task-tracker' import type { CodexJournalTranslator } from './codex-structured-journal-translation' -import type { CodexTurnProcessSnapshot } from './codex-structured-turn-processes' import type { StructuredAgentSessionEndedEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexStructuredPermissionPolicy } from './codex-structured-permission-policy' import type { @@ -102,11 +101,6 @@ export type CodexStructuredSessionAdapterDeps = { mintAcquisitionGeneration?: () => string now?: () => number requestTimeoutMs?: number - captureTurnProcesses?: (rootPid: number) => Promise - terminateTurnProcesses?: ( - rootPid: number, - baseline: CodexTurnProcessSnapshot | null - ) => Promise /** Host model catalog; sessions write their listings through and read back. */ modelCatalog?: AgentModelCatalogStore } diff --git a/src/main/codex/codex-structured-turn-cancellation.ts b/src/main/codex/codex-structured-turn-cancellation.ts index 524ab252e4f8..99af63f9d6f3 100644 --- a/src/main/codex/codex-structured-turn-cancellation.ts +++ b/src/main/codex/codex-structured-turn-cancellation.ts @@ -1,203 +1,41 @@ -import { - isCodexAppServerRequestError, - type CodexAppServerConnection -} from './codex-app-server-connection' +import { isCodexAppServerRequestError } from './codex-app-server-connection' import { isCodexAppServerUnsupportedError } from './codex-app-server-session' import { providerDiagnosticOf } from '../../shared/agent-session-failure' import type { AgentSessionCancelOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import type { - CodexSession, - CodexStructuredSessionAdapterDeps, - CodexStructuredSessionEvent -} from './codex-structured-session-state' +import type { CodexSession } from './codex-structured-session-state' import type { CodexJournalTranslationAdmission } from './codex-structured-journal-contracts' -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 -} - -function turnKey(threadId: string, turnId: string): string { - return JSON.stringify([threadId, turnId]) -} - -type TurnCancellationDeps = Pick< - CodexStructuredSessionAdapterDeps, - 'captureTurnProcesses' | 'requestTimeoutMs' | 'terminateTurnProcesses' -> & { - emit: ( - session: CodexSession, - event: CodexStructuredSessionEvent - ) => CodexJournalTranslationAdmission -} - -const ADMITTED: CodexJournalTranslationAdmission = { accepted: true } - -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, - observedAt?: number - ): boolean { - const threadId = readCodexThreadId(params) ?? session.threadId - if (method !== 'turn/completed') { - return false - } - const turnId = readCodexTurnId(params) - const state = this.state(session) - const key = turnId ? turnKey(threadId, turnId) : null - if (!key || !state.blockedCompletions.has(key)) { - return false - } - const event = { - type: 'notification' as const, - sessionId, - threadId, - method, - params, - ...(observedAt !== undefined ? { observedAt } : {}) - } - state.deferredCompletions.set(key, event) - return true - } - - async cancel( - session: CodexSession, - threadId: string, - turnId: string, - isCurrent: () => boolean = () => true, - onConfirmed?: () => CodexJournalTranslationAdmission - ): Promise { - const state = this.state(session) - const key = turnKey(threadId, turnId) - state.blockedCompletions.add(key) - const targetsPrimaryTurn = threadId === session.threadId - const baseline = targetsPrimaryTurn ? await state.baseline : null - if (!isCurrent()) { - this.releaseCompletion(session, key) - return { cancelled: false } - } - let requestError: unknown - const interruptReceipt = session.connection - .request('turn/interrupt', { threadId, turnId }, { timeoutMs: this.deps.requestTimeoutMs }) - .then( - () => true, - (error: unknown) => { - requestError = error - return false - } - ) - const [acknowledged, terminated] = await Promise.all([ - interruptReceipt, - targetsPrimaryTurn ? this.terminate(session.connection, baseline) : Promise.resolve(true) - ]) - if (terminated && acknowledged) { - const completion = state.deferredCompletions.get(key) - let confirmationError: unknown - let promptAdmission = ADMITTED - try { - promptAdmission = onConfirmed?.() ?? ADMITTED - } catch (error) { - confirmationError = error - } - const completionAdmission = this.releaseCompletion(session, key, completion) - if (confirmationError) { - throw confirmationError - } - if (!promptAdmission.accepted) { - throw new Error( - `Codex prompt cancellation lifecycle was not admitted (${promptAdmission.reason})` - ) - } - if (onConfirmed && completion && !completionAdmission.accepted) { - throw new Error( - `Codex deferred turn completion lifecycle was not admitted (${completionAdmission.reason})` - ) - } - return { cancelled: true } - } - if ( - requestError && - !isCodexAppServerRequestError(requestError) && - !isCodexAppServerUnsupportedError(requestError) - ) { - this.releaseCompletion(session, key) - throw requestError - } - // A failed cancellation must not permanently divert the provider's later - // completion for this turn. Let the normal completion path settle it. - this.releaseCompletion(session, key) - if (acknowledged) { - return { cancelled: false, unconfirmed: true } - } - if (!requestError) { - return { cancelled: false } - } - const detail = providerDiagnosticOf(requestError) +/** + * Codex answers a turn's interrupt only as that turn ends, so the answer is what confirms the + * Stop. The interrupt is the whole Stop: Codex kills the turn's one-shot commands itself and keeps + * its background terminals running until the thread ends. + */ +export async function interruptCodexTurn(input: { + session: CodexSession + threadId: string + turnId: string + requestTimeoutMs?: number + onConfirmed?: () => CodexJournalTranslationAdmission +}): Promise { + const { session, threadId, turnId } = input + try { + await session.connection.request( + 'turn/interrupt', + { threadId, turnId }, + { timeoutMs: input.requestTimeoutMs } + ) + } catch (error) { + if (!isCodexAppServerRequestError(error) && !isCodexAppServerUnsupportedError(error)) { + throw error + } + const detail = providerDiagnosticOf(error) return { cancelled: false, refusal: detail ? { detail } : {} } } - - 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, - key: string, - completion = this.state(session).deferredCompletions.get(key) - ): CodexJournalTranslationAdmission { - const state = this.state(session) - state.blockedCompletions.delete(key) - state.deferredCompletions.delete(key) - return completion ? this.deps.emit(session, completion) : ADMITTED - } - - private state(session: CodexSession): TurnProcessState { - const state = this.states.get(session) - if (!state) { - throw new Error('codex turn process state is unavailable') - } - return state + const promptAdmission = input.onConfirmed?.() + if (promptAdmission && !promptAdmission.accepted) { + throw new Error( + `Codex prompt cancellation lifecycle was not admitted (${promptAdmission.reason})` + ) } + return { cancelled: true } } diff --git a/src/main/codex/codex-structured-turn-end-settlement.ts b/src/main/codex/codex-structured-turn-end-settlement.ts index 5187aabf45aa..657006449331 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.ts @@ -1,6 +1,8 @@ -// A send Codex answered into a turn that then ended without echoing it. Codex clears -// a turn's pending input when it is interrupted, so that send never reached the model -// and is withdrawn, as a Stop's host-side withdrawal is. Any other end records pending +// A send Codex answered into a turn that then ended without echoing it. On an interrupt, a +// steered send waits in Codex's pending input, which the interrupt clears; a send that opened the +// turn is saved as the turn cancels, but Codex hard-aborts that save after 100 ms, so it is lost +// or reaches only the model's context, and the thread history shows no user message either way. +// So it is withdrawn, as a Stop's host-side withdrawal is. Any other end records pending // input before `turn/completed`, a failed turn after its `error` frame, so only that // frame settles: a failed turn that never echoed the send refused it, in Codex's words, // and a completed one leaves it pending for the journal's recovery on exit. diff --git a/src/main/codex/codex-structured-turn-processes.integration.test.ts b/src/main/codex/codex-structured-turn-processes.integration.test.ts deleted file mode 100644 index 2a63c8099eaa..000000000000 --- a/src/main/codex/codex-structured-turn-processes.integration.test.ts +++ /dev/null @@ -1,103 +0,0 @@ -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 deleted file mode 100644 index 163cbb098fbd..000000000000 --- a/src/main/codex/codex-structured-turn-processes.ts +++ /dev/null @@ -1,89 +0,0 @@ -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)) - // Added roots come from a table walk, not a spawn, so a refused tree walk has - // no handle to fall back to: the row stays in `remaining` and this reports false. - await Promise.all( - roots.map((row) => terminateWindowsProcessTree(row.pid, { site: 'codex-turn-added-roots' })) - ) - 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/crash-reporting/self-initiated-tree-kill-log.test.ts b/src/main/crash-reporting/self-initiated-tree-kill-log.test.ts index e958dac42f8a..c9a2a5cdf03c 100644 --- a/src/main/crash-reporting/self-initiated-tree-kill-log.test.ts +++ b/src/main/crash-reporting/self-initiated-tree-kill-log.test.ts @@ -124,7 +124,7 @@ describe('self-initiated tree kill breadcrumb', () => { ;(done as () => void)() return undefined as never }) as never, - site: 'codex-turn-added-roots' + site: 'codex-app-server-teardown' }) expect(getCrashBreadcrumbSnapshot()).toEqual([ @@ -132,7 +132,7 @@ describe('self-initiated tree kill breadcrumb', () => { name: 'self_tree_kill', data: expect.objectContaining({ pid: 777, - site: 'codex-turn-added-roots', + site: 'codex-app-server-teardown', scope: 'win-taskkill-tree' }) }) diff --git a/src/main/crash-reporting/self-initiated-tree-kill-log.ts b/src/main/crash-reporting/self-initiated-tree-kill-log.ts index 701cb2d646e2..d684bd874ef3 100644 --- a/src/main/crash-reporting/self-initiated-tree-kill-log.ts +++ b/src/main/crash-reporting/self-initiated-tree-kill-log.ts @@ -60,8 +60,8 @@ export type SelfInitiatedTreeKill = { at: number } -// Why 32 and not the sibling ring's 16: one teardown fans out over every root of -// a codex turn, so a single incident can spend a dozen entries on its own. +// Why 32 and not the sibling ring's 16: a window close tears down every session at once, +// a burst of 30+ group kills (see the eviction note below). const MAX_TRACKED_SELF_KILLS = 32 // Why asymmetric: a kill older than this cannot plausibly explain the death, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index df81e8ec1a08..6620139bd26d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -236,12 +236,10 @@ export type StructuredAgentSessionSetOptionInput = { fence: number } -/** `refusal`: the provider answered the Stop and declined it, in its own words when it gave any. - * `unconfirmed`: the provider took the Stop, but Orca could not confirm the turn's work ended. */ +/** `refusal`: the provider answered the Stop and declined it, in its own words when it gave any. */ export type AgentSessionCancelOutcome = { cancelled: boolean refusal?: { detail?: ProviderDiagnostic } - unconfirmed?: true } export type StructuredAgentSessionAdapter = { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts new file mode 100644 index 000000000000..28b4dde82e07 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts @@ -0,0 +1,138 @@ +// What a Stop's row says when Codex took the interrupt, against the real host, journal and Codex +// adapter. Codex answers a turn's interrupt as the turn aborts and sends the turn's end right after +// the answer, so the end reaches Orca a moment after the Stop has its answer. + +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { + THREAD_ID as THREAD, + adapterFor, + fakeCodex +} from '../../codex/codex-structured-session-adapter-fixture' +import { codexTurnLifecycleFake } from '../../codex/codex-turn-lifecycle-fake' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestMessage, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' + +const CALLER = { callerKey: 'client-1' } + +let root: string +let host: StructuredAgentSessionHost +let turns: ReturnType + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-codex-stop-row-')) + resetHostTestOperationIds() + const codex = fakeCodex() + const notify = (method: string, params: unknown): void => + codex.connections.at(-1)?.handlers.onNotification?.(method, params) + turns = codexTurnLifecycleFake(THREAD, () => notify) + codex.routes['turn/start'] = turns.routes['turn/start'] + codex.routes['turn/interrupt'] = () => { + const turnId = turns.turnId + // The answer first, then the turn's end on a later read of Codex's output. + setTimeout( + () => + notify('turn/completed', { threadId: THREAD, turn: { id: turnId, status: 'interrupted' } }), + 4 + ) + return {} + } + const store = await AgentSessionRecordStore.open({ + directory: join(root, 'store'), + hostId: 'local' + }) + host = new StructuredAgentSessionHost({ + store, + adapter: Object.assign(adapterFor(codex), { supportsCreate: () => true }), + journalDatabase: openTestJournalHostDatabase(root), + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-1', + now: () => NOW + }) + expect(await host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) +}) + +afterEach(async () => { + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +function send(text: string) { + const body = hostTestMessage(text) + return host.send(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields: { body } + }) + }, + body + }) +} + +function stop(turnId?: string) { + const fields = turnId === undefined ? {} : { turnId } + return host.cancel(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: SESSION, + fields + }) + }, + ...fields + }) +} + +async function runningTurn(): Promise { + expect(await send('count to 40')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(turns.turnId).toBe('turn-1')) + turns.start() + await vi.waitFor(async () => expect((await journalRows()).turns).toEqual(['running'])) +} + +async function journalRows() { + const items = (await host.journalSnapshot(SESSION)).items + return { + statuses: items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])), + turns: items.flatMap((item) => (item.body.kind === 'turn' ? [item.body.state] : [])) + } +} + +describe('a Codex Stop that Codex answered', () => { + it.each([ + ['names no turn', undefined], + ['names its turn, as an older client sends it', 'turn-1'] + ] as const)( + 'reads as requested, with the turn interrupted, when it %s', + async (_case, turnId) => { + await runningTurn() + + const stopped = await stop(turnId) + await vi.waitFor(async () => expect((await journalRows()).turns).toEqual(['interrupted'])) + await host.flushStreamedEvents(SESSION) + + expect((await journalRows()).statuses).toEqual(['Cancellation requested.']) + expect(stopped).toMatchObject({ ok: true, value: { cancelled: true } }) + } + ) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index 2079b25c1bb0..7bb05480f774 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -217,11 +217,12 @@ describe('a Stop that names no turn', () => { expect(await statusRows()).toEqual(["Codex didn't stop: no active turn to interrupt."]) }) - it('says the Stop is unconfirmed, not that nothing ran, when the provider took it', async () => { + it('says the Stop is unconfirmed, not that nothing ran, when the provider never answered it', async () => { const { id, result } = send('hello') await result await eventually(async () => expect((await submission(id))?.handedOverAt).toBeDefined()) - cancelTurn.mockResolvedValueOnce({ cancelled: false, unconfirmed: true }) + // Codex answers an interrupt as the turn ends, so a turn that never ends leaves it unanswered. + cancelTurn.mockRejectedValueOnce(new Error('codex app-server turn/interrupt exceeded 30000ms')) expect(await stop()).toMatchObject({ ok: true, value: { cancelled: false } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts index 4bd5b412300f..3fe70a0f4a30 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts @@ -75,13 +75,6 @@ export async function performCancel( note = { kind: 'status', text: 'The provider had already finished this turn.' } } else if (!cancelled && input.prompt) { note = null - } else if (!cancelled && outcome.unconfirmed) { - note = { - kind: 'status', - ...agentSessionFailureWords(agentSessionFailureFact('cancelUnconfirmed'), { - surface: 'row' - }) - } } else if (!cancelled) { // Sent only while the chat reads working, so a Stop that ended nothing must say why. const detail = outcome.refusal?.detail diff --git a/src/main/refused-tree-kill-root-termination.test.ts b/src/main/refused-tree-kill-root-termination.test.ts index dbd3a10f1912..f6bb531165af 100644 --- a/src/main/refused-tree-kill-root-termination.test.ts +++ b/src/main/refused-tree-kill-root-termination.test.ts @@ -1,9 +1,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { spawnMock, execFileMock, queryWindowsProcessDescendantsMock } = vi.hoisted(() => ({ +const { spawnMock, execFileMock } = vi.hoisted(() => ({ spawnMock: vi.fn(), - execFileMock: vi.fn(), - queryWindowsProcessDescendantsMock: vi.fn() + execFileMock: vi.fn() })) vi.mock('node:child_process', async (importOriginal) => ({ @@ -12,9 +11,6 @@ vi.mock('node:child_process', async (importOriginal) => ({ execFile: execFileMock })) vi.mock('electron', () => ({ ipcMain: { handle: vi.fn(), on: vi.fn() } })) -vi.mock('./providers/windows-foreground-process-rows', () => ({ - queryWindowsProcessDescendants: queryWindowsProcessDescendantsMock -})) import { getAppEnvironment, @@ -36,7 +32,6 @@ import { killSpawnedCommandTree } from './git/command-runner/spawned-command-tre import { killCodexAppServerProcessTree } from './codex/codex-app-server-process-tree-kill' import { signalProcessTree } from '../shared/child-process/process-tree-termination' import { killSourceControlAgentProcess } from './text-generation/source-control-local-process' -import { terminateCodexTurnProcesses } from './codex/codex-structured-turn-processes' /** A pid Electron reports as one of ours: every gate below must refuse it. */ const RENDERER_PID = 1001 @@ -72,7 +67,6 @@ beforeEach(() => { installMainProcessTreeKillGate() spawnMock.mockReset() execFileMock.mockReset() - queryWindowsProcessDescendantsMock.mockReset() spawnMock.mockReturnValue({ on: vi.fn(), once: vi.fn(), unref: vi.fn(), kill: vi.fn() }) }) @@ -175,35 +169,3 @@ describe('a refused tree-kill still terminates the root it owns', () => { processKill.mockRestore() }) }) - -/** - * The one gated site with nothing to fall back to: the roots it kills are found - * by a process-table walk, not spawned here, so there is no child handle. A - * refusal must then be visible — the refusal crumb is written and the turn is - * reported as not cancelled — rather than resolving as if the tree had gone. - */ -describe('a refused tree-kill with no handle to fall back to', () => { - it('reports the codex turn as not cancelled and records the refused added root', async () => { - const appServerPid = 500 - const addedRoot = { - pid: RENDERER_PID, - ppid: appServerPid, - name: 'node.exe', - command: 'node', - depth: 1 - } - queryWindowsProcessDescendantsMock.mockResolvedValue([addedRoot]) - - await expect( - terminateCodexTurnProcesses(appServerPid, { platform: 'win32', identities: new Map() }) - ).resolves.toBe(false) - - expect(execFileMock).not.toHaveBeenCalled() - expect(getCrashBreadcrumbSnapshot()).toEqual([ - expect.objectContaining({ - name: 'self_tree_kill_refused_own_chromium', - data: expect.objectContaining({ pid: RENDERER_PID, site: 'codex-turn-added-roots' }) - }) - ]) - }) -}) diff --git a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts index c86f7c852420..990c33850efe 100644 --- a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts +++ b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts @@ -33,12 +33,6 @@ import { stopStructuredAgentSessionRuntime } from './structured-agent-session-runtime' -// A Stop must never reach for real processes on this machine under a made-up pid. -vi.mock('../codex/codex-structured-turn-processes', () => ({ - captureCodexTurnProcesses: async () => null, - terminateCodexTurnProcesses: async () => true -})) - const CALLER = { callerKey: 'codex-turn-end-test' } const MODEL = { model: 'gpt-test',