From 088a712f1eaf149c31bcb8d707a02134d966c596 Mon Sep 17 00:00:00 2001 From: SmallSpider <568442079@qq.com> Date: Tue, 12 May 2026 23:25:41 +0800 Subject: [PATCH] Stop active Codex child agents on abort (#615) * fix(cli): stop active codex child agents * chore: refresh bun lockfile for deploy * fix(web): enable stop for active codex child agents * test(cli): cover aborting active codex child agents --- bun.lock | 2 + cli/src/codex/codexRemoteLauncher.test.ts | 104 +++++++++++++++++++++- cli/src/codex/codexRemoteLauncher.ts | 74 ++++++++++----- web/src/components/SessionChat.tsx | 22 +++++ web/src/lib/assistant-runtime.ts | 9 +- 5 files changed, 183 insertions(+), 28 deletions(-) diff --git a/bun.lock b/bun.lock index 67700f6e..1261eb39 100644 --- a/bun.lock +++ b/bun.lock @@ -1064,6 +1064,8 @@ "@twsxtd/hapi-linux-x64": ["@twsxtd/hapi-linux-x64@0.17.4", "", { "os": "linux", "cpu": "x64", "bin": { "hapi": "bin/hapi" } }, "sha512-+oJ/f6i6rq5S/3Vn4R25WqGpTN0RAjLSL8ALu0Egv1E88AlIuw42AQfx4Ggh37X3ICPmIq1ZyoMzPPvRLTDICQ=="], + "@twsxtd/hapi-win32-x64": ["@twsxtd/hapi-win32-x64@0.17.4", "", { "os": "win32", "cpu": "x64", "bin": { "hapi": "bin/hapi.exe" } }, "sha512-bOttRU1UMKYkCo18ENHAM2Q1Cb8Lr1tU25KxNQDOhvJwW2LQQfZZ/ZMakebDFH28JH23ohZhZpcjlN1Cq8rwdg=="], + "@types/aria-query": ["@types/aria-query@5.0.4", "", {}, "sha512-rfT93uj5s0PRL7EzccGMs3brplhcrghnDoV26NqKhCAS1hVo+WdNsPvE/yb6ilfr5hi2MEk6d5EWJTKdxg8jVw=="], "@types/babel__core": ["@types/babel__core@7.20.5", "", { "dependencies": { "@babel/parser": "^7.20.7", "@babel/types": "^7.20.7", "@types/babel__generator": "*", "@types/babel__template": "*", "@types/babel__traverse": "*" } }, "sha512-qoQprZvz5wQFJwMDqeseRXWv3rqMvhgpbXFfVyWhbx9X47POIA6i/+dXefEmZKoAgOaTdaIgNSMqMIU61yRyzA=="], diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index af9fc23e..2617d6c2 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -34,6 +34,9 @@ const harness = vi.hoisted(() => ({ emitParentSpawnStartWithoutEnd: false, emitParentSendInputFailure: false, emitParentResumeSuccess: false, + emitRunningChildTurnBeforeSuppressedParent: false, + emitCompletedChildTurnBeforeSuppressedParent: false, + emitTurnAbortedOnInterrupt: false, bridgeOptions: [] as unknown[] })); @@ -109,6 +112,33 @@ vi.mock('./codexAppServerClient', () => { return { turn: { id: turnId } }; } + if ( + harness.emitRunningChildTurnBeforeSuppressedParent + || harness.emitCompletedChildTurnBeforeSuppressedParent + ) { + const childStarted = { + msg: { + type: 'task_started', + thread_id: 'child-thread', + turn_id: 'child-turn' + } + }; + harness.notifications.push({ method: 'codex/event/task_started', params: childStarted }); + this.notificationHandler?.('codex/event/task_started', childStarted); + + if (harness.emitCompletedChildTurnBeforeSuppressedParent) { + const childCompleted = { + msg: { + type: 'task_complete', + thread_id: 'child-thread', + turn_id: 'child-turn' + } + }; + harness.notifications.push({ method: 'codex/event/task_complete', params: childCompleted }); + this.notificationHandler?.('codex/event/task_complete', childCompleted); + } + } + if (harness.suppressTurnCompletion) { return { turn: { id: turnId } }; } @@ -565,10 +595,19 @@ vi.mock('./codexAppServerClient', () => { } async interruptTurn(params?: { threadId?: string; turnId?: string }): Promise> { - harness.interruptedTurns.push({ - threadId: params?.threadId ?? 'thread-unknown', - turnId: params?.turnId ?? 'turn-unknown' - }); + const threadId = params?.threadId ?? 'thread-unknown'; + const turnId = params?.turnId ?? 'turn-unknown'; + harness.interruptedTurns.push({ threadId, turnId }); + if (harness.emitTurnAbortedOnInterrupt) { + const interrupted = { + threadId, + turnId, + status: 'interrupted', + turn: { id: turnId } + }; + harness.notifications.push({ method: 'turn/completed', params: interrupted }); + this.notificationHandler?.('turn/completed', interrupted); + } return {}; } @@ -737,6 +776,9 @@ describe('codexRemoteLauncher', () => { harness.emitParentSpawnStartWithoutEnd = false; harness.emitParentSendInputFailure = false; harness.emitParentResumeSuccess = false; + harness.emitRunningChildTurnBeforeSuppressedParent = false; + harness.emitCompletedChildTurnBeforeSuppressedParent = false; + harness.emitTurnAbortedOnInterrupt = false; harness.bridgeOptions = []; }); @@ -1431,6 +1473,60 @@ describe('codexRemoteLauncher', () => { expect(session.thinking).toBe(false); }); + it('interrupts active child agent turns before clearing codex thread state', async () => { + harness.suppressTurnCompletion = true; + harness.emitRunningChildTurnBeforeSuppressedParent = true; + const { session, resetThreadCalls } = createSessionStub(['first message', '/clear']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.interruptedTurns).toEqual([ + { threadId: 'thread-1', turnId: 'turn-1' }, + { threadId: 'child-thread', turnId: 'child-turn' } + ]); + expect(resetThreadCalls).toEqual(['thread-1']); + expect(session.thinking).toBe(false); + }); + + it('interrupts active child agent turns when the abort RPC is invoked', async () => { + harness.suppressTurnCompletion = true; + harness.emitRunningChildTurnBeforeSuppressedParent = true; + harness.emitTurnAbortedOnInterrupt = true; + const { session, rpcHandlers } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.startTurnThreadIds).toEqual(['thread-1']); + expect(rpcHandlers.has('abort')).toBe(true); + }); + + await rpcHandlers.get('abort')?.({}); + const exitReason = await running; + + expect(exitReason).toBe('exit'); + expect(harness.interruptedTurns).toEqual([ + { threadId: 'thread-1', turnId: 'turn-1' }, + { threadId: 'child-thread', turnId: 'child-turn' } + ]); + expect(session.thinking).toBe(false); + }); + + it('does not interrupt completed child agent turns when clearing codex thread state', async () => { + harness.suppressTurnCompletion = true; + harness.emitCompletedChildTurnBeforeSuppressedParent = true; + const { session, resetThreadCalls } = createSessionStub(['first message', '/clear']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.interruptedTurns).toEqual([ + { threadId: 'thread-1', turnId: 'turn-1' } + ]); + expect(resetThreadCalls).toEqual(['thread-1']); + expect(session.thinking).toBe(false); + }); + it('compacts the current thread without starting a turn', async () => { const { session, sessionEvents } = createSessionStub(['first message', '/compact']); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index c8ac8087..1343039a 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -84,6 +84,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { private abortController: AbortController = new AbortController(); private currentThreadId: string | null = null; private currentTurnId: string | null = null; + private readonly activeChildTurns = new Map(); constructor(session: CodexSession) { super(process.env.DEBUG ? session.logPath : undefined); @@ -95,19 +96,50 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return React.createElement(CodexDisplay, context); } + private async interruptActiveTurns(reason: string): Promise { + const turnsToInterrupt = [ + ...(this.currentThreadId && this.currentTurnId + ? [{ threadId: this.currentThreadId, turnId: this.currentTurnId, role: 'parent' as const }] + : []), + ...Array.from(this.activeChildTurns, ([threadId, turnId]) => ({ + threadId, + turnId, + role: 'child' as const + })) + ]; + + if (turnsToInterrupt.length === 0) { + return; + } + + const results = await Promise.allSettled( + turnsToInterrupt.map((target) => this.appServerClient.interruptTurn({ + threadId: target.threadId, + turnId: target.turnId + })) + ); + + results.forEach((result, index) => { + const target = turnsToInterrupt[index]; + if (result.status === 'fulfilled') { + if (target.role === 'child') { + this.activeChildTurns.delete(target.threadId); + } + return; + } + + logger.debug( + `[Codex] Error interrupting ${target.role} app-server turn ` + + `for ${reason}; threadId=${target.threadId} turnId=${target.turnId}:`, + result.reason + ); + }); + } + private async handleAbort(): Promise { logger.debug('[Codex] Abort requested - stopping current task'); try { - if (this.currentThreadId && this.currentTurnId) { - try { - await this.appServerClient.interruptTurn({ - threadId: this.currentThreadId, - turnId: this.currentTurnId - }); - } catch (error) { - logger.debug('[Codex] Error interrupting app-server turn:', error); - } - } + await this.interruptActiveTurns('abort'); this.currentTurnId = null; this.abortController.abort(); @@ -1635,6 +1667,15 @@ class CodexRemoteLauncher extends RemoteLauncherBase { `[Codex] Routing event from non-active thread into agent trace; ` + `type=${msgType}, eventThreadId=${eventThreadId}, activeThread=${this.currentThreadId}` ); + if (msgType === 'task_started') { + if (eventTurnId) { + this.activeChildTurns.set(eventThreadId, eventTurnId); + } else { + logger.debug(`[Codex] Child task_started missing turn id; threadId=${eventThreadId}`); + } + } else if (isTerminalEvent) { + this.activeChildTurns.delete(eventThreadId); + } handleChildCodexEvent(eventThreadId, msg); return; } @@ -2176,17 +2217,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { }; const interruptActiveTurn = async () => { - const threadId = this.currentThreadId; - const turnId = this.currentTurnId; - if (!threadId || !turnId) { - return; - } - - try { - await appServerClient.interruptTurn({ threadId, turnId }); - } catch (error) { - logger.debug('[Codex] Error interrupting app-server turn for slash command:', error); - } + await this.interruptActiveTurns('slash command'); }; const resumeExistingThreadForCompact = async (mode: EnhancedMode): Promise => { @@ -2475,6 +2506,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { this.permissionHandler = null; this.reasoningProcessor = null; this.diffProcessor = null; + this.activeChildTurns.clear(); logger.debug('[codex-remote]: cleanup done'); } diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index e8986828..b36ac351 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -47,6 +47,23 @@ function getOutlineTitle(session: Session): string { return session.id.slice(0, 8) } +function hasAbortableAgentRun(blocks: readonly ChatBlock[]): boolean { + for (const block of blocks) { + if (block.kind === 'tool-call') { + if ( + block.tool.name === 'CodexAgent' + && (block.tool.state === 'running' || block.tool.state === 'pending') + ) { + return true + } + if (hasAbortableAgentRun(block.children)) { + return true + } + } + } + return false +} + export function SessionChat(props: { api: ApiClient session: Session @@ -278,6 +295,10 @@ export function SessionChat(props: { () => reconcileChatBlocks(reduced.blocks, blocksByIdRef.current), [reduced.blocks] ) + const hasRunningChildAgent = useMemo( + () => hasAbortableAgentRun(reduced.blocks), + [reduced.blocks] + ) useEffect(() => { blocksByIdRef.current = reconciled.byId @@ -404,6 +425,7 @@ export function SessionChat(props: { session: props.session, blocks: visibleBlocks, isSending: props.isSending, + isRunning: props.session.thinking || hasRunningChildAgent, onSendMessage: handleSend, onAbort: handleAbort, attachmentAdapter, diff --git a/web/src/lib/assistant-runtime.ts b/web/src/lib/assistant-runtime.ts index 00fb42f6..d6a6b055 100644 --- a/web/src/lib/assistant-runtime.ts +++ b/web/src/lib/assistant-runtime.ts @@ -232,17 +232,20 @@ export function useHappyRuntime(props: { session: Session blocks: readonly VisibleChatBlock[] isSending: boolean + isRunning?: boolean onSendMessage: (text: string, attachments?: AttachmentMetadata[]) => void onAbort: () => Promise attachmentAdapter?: AttachmentAdapter allowSendWhenInactive?: boolean }) { + const isRunning = props.isRunning ?? props.session.thinking + // Use cached message converter for performance optimization // This prevents re-converting all messages on every render const convertedMessages = useExternalMessageConverter({ callback: toThreadMessageLike, messages: props.blocks as VisibleChatBlock[], - isRunning: props.session.thinking, + isRunning, }) const onNew = useCallback(async (message: AppendMessage) => { @@ -259,7 +262,7 @@ export function useHappyRuntime(props: { // useExternalStoreRuntime may use adapter identity for subscriptions const adapter = useMemo(() => ({ isDisabled: props.isSending || (!props.session.active && !props.allowSendWhenInactive), - isRunning: props.session.thinking, + isRunning, messages: convertedMessages, onNew, onCancel, @@ -269,7 +272,7 @@ export function useHappyRuntime(props: { props.session.active, props.isSending, props.allowSendWhenInactive, - props.session.thinking, + isRunning, convertedMessages, onNew, onCancel,