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
2026-08-05 06:24:37 +00:00 · 2026-05-12 23:25:41 +08:00
parent af3491e046
commit 088a712f1e
5 changed files with 183 additions and 28 deletions
@@ -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<Record<string, never>> {
-            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']);

@@ -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<string, string>();

    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<void> {
+        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<void> {
        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<string | null> => {
@@ -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');
    }