diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 2aecd18e21..cc667b2494 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -150,6 +150,15 @@ function isCodexAlreadyUnarchivedError(error: unknown, threadId: string): boolea const TURN_START_TIMEOUT_MS = 90 * 1000; const INTERRUPT_TIMEOUT_MS = 2_000; +// `turn/interrupt` ends a turn but leaves the shell commands it started running. +// Each terminate is bounded by the first constant. A turn start waits for the +// interrupted turn's commands to end, at most the interrupt's own bound plus one +// terminate's and a margin, so it gives up only when a request outlives its timeout. +const INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS = 2_000; +const INTERRUPTED_TURN_CLEANUP_WAIT_MS = + INTERRUPT_TIMEOUT_MS + INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS + 500; +// Turns whose commands are remembered; older turns are forgotten first. +const INTERRUPTED_TURN_TRACKED_TURNS = 16; const CODEX_PROVIDER = "codex" as const; // These native provider identities belong to Codex, not to an older custom // Paseo provider. Never include them in a custom provider's legacy import scan. @@ -3476,6 +3485,14 @@ export class CodexAppServerAgentSession implements AgentSession { private pendingFileChangeOutputDeltas = new Map(); private pendingAssistantMessageBoundary = false; private terminalCommandByProcessId = new Map(); + // The root thread's still-running command process ids, keyed by the native turn that + // started them. Codex's `thread/backgroundTerminals/list` carries no turn, so + // `item/started` is the only record of which turn owns a process. + private terminalProcessIdsByTurnId = new Map< + string, + { processIds: Set; terminatedProcessIds: Set; interruptAccepted: boolean } + >(); + private interruptedTurnCleanup: Promise | null = null; private pendingUnlabeledTerminalInteractions = new Map< string, Array<{ callId: string; stdin: string | null }> @@ -4431,6 +4448,9 @@ export class CodexAppServerAgentSession implements AgentSession { hasDeveloperInstructions: turnStart.hasDeveloperInstructions, hasCodexConfig: turnStart.hasCodexConfig, }); + // The interrupted turn's commands end before this turn begins, so none of them + // outlives it into the new turn. + await this.waitForInterruptedTurnCleanup(); if (pendingStart.cancelRequested) { throw new Error("Codex turn start was interrupted before reaching Codex"); } @@ -5036,15 +5056,27 @@ export class CodexAppServerAgentSession implements AgentSession { if (!turnId || (foregroundTurnId && this.activeForegroundTurnId !== foregroundTurnId)) { throw new Error("Cannot interrupt Codex before turn/started identifies the active turn"); } + const client = this.client; + const threadId = this.currentThreadId; + const interruptRequest = client.request( + "turn/interrupt", + { + threadId, + turnId, + }, + INTERRUPT_TIMEOUT_MS, + ); + // Once Codex accepts the interrupt, end the commands this turn started. Registered + // before the await, so a turn start racing this interrupt waits for it. An already-idle + // turn was not interrupted, and what it left running stays. + this.trackInterruptedTurnCleanup( + interruptRequest.then( + () => this.endInterruptedTurnProcesses(client, threadId, turnId), + () => undefined, + ), + ); try { - await this.client.request( - "turn/interrupt", - { - threadId: this.currentThreadId, - turnId, - }, - INTERRUPT_TIMEOUT_MS, - ); + await interruptRequest; } catch (error) { if (!isCodexAlreadyIdleInterrupt(error)) { throw error; @@ -5085,6 +5117,137 @@ export class CodexAppServerAgentSession implements AgentSession { return !this.client.hasPendingRequests(); } + // Remember which turn started a command, so an interrupt ends that turn's commands and + // no other turn's. Only the root thread's: a subagent's commands live on its own thread. + // A command that starts after its turn's interrupt was accepted is ended as soon as it + // is seen. + private recordTurnTerminalProcess( + parsed: Extract, + ): void { + const processId = parsed.item.processId; + if ( + parsed.source !== "item" || + !parsed.turnId || + !parsed.threadId || + parsed.threadId !== this.currentThreadId || + typeof processId !== "string" || + !processId || + normalizeCodexThreadItemType( + typeof parsed.item.type === "string" ? parsed.item.type : undefined, + ) !== "commandExecution" + ) { + return; + } + const record = this.terminalProcessIdsByTurnId.get(parsed.turnId); + if (parsed.kind === "item_completed") { + // Only a completed command has exited. A failed one can still be running: + // Codex reports an interrupted turn's live command as failed. + if (parsed.item.status === "completed") record?.processIds.delete(processId); + return; + } + const current = record ?? this.getTurnTerminalRecord(parsed.turnId); + current.processIds.add(processId); + if (current.interruptAccepted && this.client) { + this.trackInterruptedTurnCleanup( + this.endInterruptedTurnProcesses(this.client, parsed.threadId, parsed.turnId), + ); + } + } + + // An accepted interrupt needs a record even before any of its turn's commands is seen: + // an `item/started` can still be queued behind an earlier notification's conversion. + private getTurnTerminalRecord(turnId: string) { + const existing = this.terminalProcessIdsByTurnId.get(turnId); + if (existing) return existing; + const record = { + processIds: new Set(), + terminatedProcessIds: new Set(), + interruptAccepted: false, + }; + this.terminalProcessIdsByTurnId.set(turnId, record); + for (const oldestTurnId of this.terminalProcessIdsByTurnId.keys()) { + if (this.terminalProcessIdsByTurnId.size <= INTERRUPTED_TURN_TRACKED_TURNS) break; + this.terminalProcessIdsByTurnId.delete(oldestTurnId); + } + return record; + } + + // A later sweep joins the pending one, so a turn start waits for both. + private trackInterruptedTurnCleanup(cleanup: Promise): void { + const tracked = Promise.all([this.interruptedTurnCleanup, cleanup]) + .then(() => undefined) + .finally(() => { + if (this.interruptedTurnCleanup === tracked) this.interruptedTurnCleanup = null; + }); + this.interruptedTurnCleanup = tracked; + } + + // Terminate each process the interrupted turn started, by its own id, once each. + // `thread/backgroundTerminals/clean` is never used: it ends every terminal on the thread, + // an earlier turn's included. A process already gone answers `terminated: false`. Never + // rejects; failures are logged. + private async endInterruptedTurnProcesses( + client: CodexAppServerClient, + threadId: string | null, + turnId: string, + ): Promise { + if (!threadId) return; + const record = this.getTurnTerminalRecord(turnId); + record.interruptAccepted = true; + const processIds = [...record.processIds].filter( + (processId) => !record.terminatedProcessIds.has(processId), + ); + for (const processId of processIds) record.terminatedProcessIds.add(processId); + const logContext = { agentId: this.agentId, provider: CODEX_PROVIDER, threadId, turnId }; + await Promise.all( + processIds.map(async (processId) => { + try { + const response = await client.request( + "thread/backgroundTerminals/terminate", + { threadId, processId }, + INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS, + ); + this.logger.info( + { ...logContext, processId, terminated: toObjectRecord(response)?.terminated ?? null }, + "provider.codex.interrupt.terminal_terminated", + ); + } catch (error) { + this.logger.warn( + { ...logContext, processId, err: error }, + "provider.codex.interrupt.terminal_terminate_failed", + ); + } + }), + ); + } + + // A new turn starts only after every pending sweep of interrupted turns' commands has + // ended, or after the bound, whichever is first. + private async waitForInterruptedTurnCleanup(): Promise { + if (!this.interruptedTurnCleanup) return; + let timer: ReturnType | undefined; + const timedOut = await Promise.race([ + (async () => { + while (this.interruptedTurnCleanup) await this.interruptedTurnCleanup; + return false; + })(), + new Promise((resolve) => { + timer = setTimeout(() => resolve(true), INTERRUPTED_TURN_CLEANUP_WAIT_MS); + }), + ]); + clearTimeout(timer); + if (timedOut) { + this.logger.warn( + { + agentId: this.agentId, + provider: CODEX_PROVIDER, + waitMs: INTERRUPTED_TURN_CLEANUP_WAIT_MS, + }, + "provider.codex.interrupt.terminal_cleanup_timed_out", + ); + } + } + async close(): Promise { this.closed = true; this.clearPendingPermissions(); @@ -6209,6 +6372,11 @@ export class CodexAppServerAgentSession implements AgentSession { return; } this.completePendingRootCompactions(); + // What a turn that was not interrupted leaves running, it left on purpose; no later + // interrupt may end it. + if (parsed.status !== "interrupted" && this.currentTurnId) { + this.terminalProcessIdsByTurnId.delete(this.currentTurnId); + } if (parsed.status === "failed") { this.emitEvent({ type: "turn_failed", @@ -6613,6 +6781,7 @@ export class CodexAppServerAgentSession implements AgentSession { private async handleItemCompletedNotification( parsed: Extract, ): Promise { + this.recordTurnTerminalProcess(parsed); // Codex emits mirrored lifecycle notifications via both `codex/event/item_*` // and canonical `item/*`. Render ordinary items only from the canonical // channel, but accept a legacy-only child announcement so it can establish @@ -6775,6 +6944,7 @@ export class CodexAppServerAgentSession implements AgentSession { private async handleItemStartedNotification( parsed: Extract, ): Promise { + this.recordTurnTerminalProcess(parsed); if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) { return; } diff --git a/packages/server/src/server/agent/providers/codex-interrupted-turn-terminals.test.ts b/packages/server/src/server/agent/providers/codex-interrupted-turn-terminals.test.ts new file mode 100644 index 0000000000..3d2ea64f9a --- /dev/null +++ b/packages/server/src/server/agent/providers/codex-interrupted-turn-terminals.test.ts @@ -0,0 +1,374 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; + +import type { AgentSession } from "../agent-sdk-types.js"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { CodexAppServerAgentSession } from "./codex-app-server-agent.js"; +import { + createFakeCodexAppServer, + type FakeCodexAppServer, + waitForNextEvent, + waitForTimelineToolCall, +} from "./codex/test-utils/fake-app-server.js"; + +const THREAD = "thread-1"; +const TERMINATE = "thread/backgroundTerminals/terminate"; + +type Handler = (params: unknown) => unknown; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function paramsOf(value: unknown): Record { + return value && typeof value === "object" ? (value as Record) : {}; +} + +let sessions: AgentSession[] = []; + +afterEach(async () => { + for (const session of sessions) await session.close().catch(() => undefined); + sessions = []; +}); + +/** + * A Codex app-server whose interrupt ends the turn the way Codex does: an `interrupted` + * turn completion, then the response. `terminate` answers per process. + */ +function createHarness( + options: { + terminate?: (processId: string) => unknown; + interruptResponse?: (completeTurn: () => void) => unknown; + } = {}, +) { + const handlers: Record = { + "turn/interrupt": () => { + const completeTurn = () => + appServer.completeTurn({ threadId: THREAD, status: "interrupted" }); + if (options.interruptResponse) return options.interruptResponse(completeTurn); + completeTurn(); + return {}; + }, + [TERMINATE]: (params) => + options.terminate?.(String(paramsOf(params).processId)) ?? { terminated: true }, + }; + const appServer: FakeCodexAppServer = createFakeCodexAppServer(handlers); + const session = new CodexAppServerAgentSession( + { + provider: "codex", + cwd: "/workspace/project", + modeId: "auto", + model: "gpt-5.4", + }, + null, + createTestLogger(), + async () => appServer.child, + ); + sessions.push(session); + + function requests(method: string): Record[] { + return appServer + .requests() + .filter((message) => message.method === method && "id" in message) + .map((message) => paramsOf(message.params)); + } + + function terminated(): string[] { + return requests(TERMINATE) + .map((params) => String(params.processId)) + .sort(); + } + + // Starts a turn and lets Codex identify it as `nativeTurnId`. + async function startTurn(prompt: string, nativeTurnId: string): Promise { + const before = requests("turn/start").length; + const started = session.startTurn(prompt); + await vi.waitFor(() => expect(requests("turn/start")).toHaveLength(before + 1), { + timeout: 5_000, + }); + const turnStarted = waitForNextEvent(session, "turn_started"); + appServer.startsTurn({ threadId: THREAD, turnId: nativeTurnId }); + await started; + await turnStarted; + } + + async function runsCommand(params: { turnId: string; processId: string }): Promise { + const itemId = `exec-${params.processId}`; + const shown = waitForTimelineToolCall(session, itemId); + appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: params.turnId, + itemId, + processId: params.processId, + command: `sleep ${params.processId}`, + }); + await shown; + } + + async function completesTurn(): Promise { + const completed = waitForNextEvent(session, "turn_completed"); + appServer.completeTurn({ threadId: THREAD }); + await completed; + } + + // Interrupts the running turn and waits for the session to report it canceled, as a + // caller does before it starts the next turn: Codex notifications are handled in order + // but asynchronously, so the interrupt's answer can arrive before the turn has ended. + async function interrupts(): Promise { + const canceled = waitForNextEvent(session, "turn_canceled"); + await session.interrupt(); + await canceled; + } + + return { + appServer, + session, + requests, + terminated, + startTurn, + runsCommand, + completesTurn, + interrupts, + }; +} + +// Turn A leaves a server running on purpose and completes. Turn B starts two +// commands, and a subagent on its own thread starts a third, then B is interrupted. +async function interruptTurnB(harness: ReturnType): Promise { + await harness.startTurn("start a server", "native-A"); + await harness.runsCommand({ turnId: "native-A", processId: "101" }); + await harness.completesTurn(); + + await harness.startTurn("run two commands", "native-B"); + await harness.runsCommand({ turnId: "native-B", processId: "201" }); + // A subagent thread's command shows no row here; the next command's row shows it + // was read, because notifications are read in order. + harness.appServer.startsTerminalCommand({ + threadId: "child-thread", + turnId: "native-B", + itemId: "exec-209", + processId: "209", + command: "sleep 209", + }); + await harness.runsCommand({ turnId: "native-B", processId: "202" }); + await harness.interrupts(); +} + +describe("an interrupted Codex turn's commands end before the next turn", () => { + test("terminates exactly the processes the interrupted turn started, by id, and never cleans", async () => { + const harness = createHarness(); + await interruptTurnB(harness); + await harness.startTurn("correction", "native-C"); + + expect(harness.terminated()).toEqual(["201", "202"]); + expect(harness.requests(TERMINATE).every((params) => params.threadId === THREAD)).toBe(true); + // `clean` would end every terminal on the thread, an earlier turn's included. + expect( + harness.appServer + .requests() + .filter((message) => String(message.method).startsWith("thread/backgroundTerminals/")) + .map((message) => message.method), + ).toEqual([TERMINATE, TERMINATE]); + harness.appServer.assertNoErrors(); + }); + + test("leaves the next turn's commands running when Codex starts that turn before acknowledging the interrupt", async () => { + const acknowledge = deferred(); + const harness = createHarness({ + // Codex ends B, starts a turn of its own (a goal continuation), and only then + // answers the interrupt. + interruptResponse: (completeTurn) => { + completeTurn(); + return acknowledge.promise.then(() => ({})); + }, + }); + await harness.startTurn("run two commands", "native-B"); + await harness.runsCommand({ turnId: "native-B", processId: "201" }); + await harness.runsCommand({ turnId: "native-B", processId: "202" }); + const interrupting = harness.session.interrupt(); + await vi.waitFor(() => expect(harness.requests("turn/interrupt")).toHaveLength(1)); + const autonomousTurn = waitForNextEvent(harness.session, "turn_started"); + harness.appServer.startsTurn({ threadId: THREAD, turnId: "native-C" }); + await autonomousTurn; + await harness.runsCommand({ turnId: "native-C", processId: "301" }); + + acknowledge.resolve(); + await interrupting; + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"])); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(harness.terminated()).toEqual(["201", "202"]); + }); + + test("sends the next turn/start only after the terminates are answered", async () => { + const answer = deferred(); + const harness = createHarness({ + terminate: () => answer.promise.then(() => ({ terminated: true })), + }); + await interruptTurnB(harness); + const next = harness.startTurn("correction", "native-C"); + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"])); + await new Promise((resolve) => setTimeout(resolve, 200)); + expect(harness.requests("turn/start")).toHaveLength(2); + + answer.resolve(); + await next; + expect(harness.requests("turn/start")).toHaveLength(3); + }); + + test.each(["initial", "late"] as const)( + "tracks both cleanup sweeps when the %s answer arrives first", + async (answeredFirst) => { + const first = deferred(); + const second = deferred(); + const harness = createHarness({ + terminate: (processId) => + (processId === "201" ? first : second).promise.then(() => ({ terminated: true })), + }); + await harness.startTurn("run commands", "native-B"); + await harness.runsCommand({ turnId: "native-B", processId: "201" }); + await harness.interrupts(); + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201"])); + const next = harness.startTurn("correction", "native-C"); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(harness.requests("turn/start")).toHaveLength(1); + // The next turn is already waiting when another process is announced. + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-B", + itemId: "late-202", + processId: "202", + command: "sleep 202", + }); + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-B", + itemId: "duplicate-202", + processId: "202", + command: "sleep 202", + }); + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"])); + (answeredFirst === "initial" ? first : second).resolve(); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(harness.requests("turn/start")).toHaveLength(1); + (answeredFirst === "initial" ? second : first).resolve(); + await next; + expect(harness.terminated()).toEqual(["201", "202"]); + // Accepted state survives the completed sweep and a newer foreground turn. + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-B", + itemId: "late-203", + processId: "203", + command: "sleep 203", + }); + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202", "203"])); + harness.appServer.assertNoErrors(); + }, + ); + + test("keeps the first sweep pending when a late sweep answers before the next start", async () => { + const first = deferred(); + const harness = createHarness({ + terminate: (processId) => + processId === "201" + ? first.promise.then(() => ({ terminated: true })) + : { terminated: true }, + }); + await harness.startTurn("run commands", "native-B"); + await harness.runsCommand({ turnId: "native-B", processId: "201" }); + await harness.interrupts(); + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-B", + itemId: "late-202", + processId: "202", + command: "sleep 202", + }); + await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"])); + const next = harness.startTurn("correction", "native-C"); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(harness.requests("turn/start")).toHaveLength(1); + first.resolve(); + await next; + harness.appServer.assertNoErrors(); + }); + + test("forgets completed commands but retains failed commands at interrupt", async () => { + const harness = createHarness(); + await harness.startTurn("run commands", "native-B"); + for (const processId of ["201", "202"]) { + await harness.runsCommand({ turnId: "native-B", processId }); + harness.appServer.child.stdout.write( + JSON.stringify({ + method: "item/completed", + params: { + threadId: THREAD, + turnId: "native-B", + item: { + id: `exec-${processId}`, + type: "commandExecution", + processId, + command: `sleep ${processId}`, + status: processId === "201" ? "completed" : "failed", + exitCode: processId === "201" ? 0 : null, + }, + }, + }) + "\n", + ); + } + // A following command's row is a barrier for both completion handlers. + await harness.runsCommand({ turnId: "native-B", processId: "203" }); + await harness.interrupts(); + await harness.startTurn("correction", "native-C"); + expect(harness.terminated()).toEqual(["202", "203"]); + harness.appServer.assertNoErrors(); + }); + + test("retains accepted interrupts for only the latest sixteen turns", async () => { + const harness = createHarness(); + for (let index = 0; index < 17; index++) { + await harness.startTurn("run a command", `native-${index}`); + await harness.interrupts(); + } + await harness.startTurn("correction", "native-current"); + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-0", + itemId: "evicted-201", + processId: "201", + command: "sleep 201", + }); + harness.appServer.startsTerminalCommand({ + threadId: THREAD, + turnId: "native-16", + itemId: "retained-202", + processId: "202", + command: "sleep 202", + }); + await harness.runsCommand({ turnId: "native-current", processId: "301" }); + expect(harness.terminated()).toEqual(["202"]); + harness.appServer.assertNoErrors(); + }); + + test("a failing or unanswered terminate does not block or fail the next turn", async () => { + const harness = createHarness({ + // 201 is refused while the next turn waits; 202 is never answered. + terminate: (processId) => + processId === "201" + ? new Promise((resolve) => setTimeout(resolve, 300)).then(() => ({ + __jsonRpcError: { code: -32600, message: "invalid background terminal process id" }, + })) + : new Promise(() => {}), + }); + await interruptTurnB(harness); + const startedAt = Date.now(); + await harness.startTurn("correction", "native-C"); + + expect(Date.now() - startedAt).toBeLessThan(5_000); + expect(harness.requests("turn/start")).toHaveLength(3); + expect(harness.terminated()).toEqual(["201", "202"]); + }, 15_000); +});