From 175f36d64ada373a52d89bc115d7e64707aadf94 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 14 Nov 2026 12:13:07 +0700 Subject: [PATCH] Keep foreground Codex commands out of background tasks (#9635) * fix(codex): keep foreground commands out of background tasks Codex lists live terminals before their initial tool call returns. Exclude commands owned by an active turn so ordinary shell calls do not flash the background-work pill, while preserving processes left running by earlier turns. * fix(codex): preserve foreground ownership during resume Start terminal inventory ownership before installing notification handlers so autonomous commands emitted during thread resume remain foreground until their turn settles. Use named fields for the new inventory inputs. --- .../providers/codex-app-server-agent.test.ts | 135 +++++++++++++++++- .../agent/providers/codex-app-server-agent.ts | 32 ++++- .../providers/codex/background-inventory.ts | 56 +++++++- .../codex/test-utils/fake-app-server.ts | 2 +- 4 files changed, 214 insertions(+), 11 deletions(-) diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 667c0965fb..0b8ca8ea5a 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -2706,11 +2706,14 @@ describe("Codex app-server provider", () => { turnId: "turn-next", itemId: "new", processId: "12", - command: "echo hello", + command: "sleep 99", }); await vi.waitFor(() => expect(hasBackgroundListRequest(appServer, "forked-thread")).toBe(true), ); + await new Promise((resolve) => setImmediate(resolve)); + expect(latestBackgroundWork(events)).toMatchObject({ items: [{ command: "sleep 180" }] }); + appServer.completeTurn({ threadId: "forked-thread" }); await vi.waitFor(() => expect(latestBackgroundWork(events)).toMatchObject({ items: [{ command: "sleep 180" }, { command: "sleep 99" }], @@ -8221,6 +8224,49 @@ describe("Codex denied plan approvals", () => { }); describe("Codex interactive resume startup", () => { + test("keeps an autonomous command started during resume out of background work until settlement", async () => { + const appServer = createFakeCodexAppServer({ + "thread/resume": () => { + appServer.startsTurn({ threadId: "saved-thread", turnId: "autonomous-turn" }); + appServer.startsTerminalCommand({ + threadId: "saved-thread", + turnId: "autonomous-turn", + itemId: "resumed-command", + processId: "123", + command: "sleep 120", + }); + return { thread: { id: "saved-thread" } }; + }, + "thread/read": () => ({ thread: { turns: [] } }), + "thread/backgroundTerminals/list": () => ({ + data: [{ itemId: "resumed-command", processId: "123", command: "sleep 120" }], + nextCursor: null, + }), + }); + const session = new CodexAppServerAgentSession( + createConfig({ modeId: "full-access" }), + { sessionId: "saved-thread" }, + createTestLogger(), + async () => appServer.child, + ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + await session.connect(); + await expect.poll(() => latestBackgroundWork(events)?.items).toEqual([]); + appServer.completeTurn({ threadId: "saved-thread" }); + await expect + .poll(() => latestBackgroundWork(events)?.items) + .toMatchObject([{ command: "sleep 120" }]); + expect( + events.filter((event) => event.type === "background_work" && event.items?.length), + ).toHaveLength(1); + } finally { + await session.close(); + appServer.assertNoErrors(); + } + }); + test("delivers an autonomous turn that starts during resume to the first subscriber", async () => { const appServer = createFakeCodexAppServer({ "turn/interrupt": () => ({}), @@ -8312,6 +8358,93 @@ describe("Codex interactive resume startup", () => { ); }); +test.each([true, false])( + "Codex background work excludes a live turn's commands (inventory item IDs: %s)", + async (itemIds) => { + let terminals = [{ itemId: "earlier", processId: "1", command: "sleep 180" }]; + const appServer = createFakeCodexAppServer({ + "thread/backgroundTerminals/list": () => ({ + data: terminals.map(({ itemId, ...row }) => (itemIds ? { ...row, itemId } : row)), + nextCursor: null, + }), + }); + const session = new CodexAppServerAgentSession( + createConfig(), + null, + createTestLogger(), + async () => appServer.child, + ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + await session.startTurn("first"); + await appServer.waitForTurnStart(); + appServer.completeTurn(); + await expect + .poll(() => latestBackgroundWork(events)?.items) + .toMatchObject([{ command: "sleep 180" }]); + await session.startTurn("git status"); + appServer.startsTurn({ threadId: "thread-1", turnId: "second" }); + terminals = [...terminals, { itemId: "status", processId: "2", command: "git status" }]; + const before = appServer.requests().length; + appServer.startsTerminalCommand({ + threadId: "thread-1", + turnId: "second", + itemId: "status", + processId: itemIds ? undefined : "2", + command: "git status", + }); + await expect + .poll(() => + appServer + .requests() + .slice(before) + .some((request) => request.method === "thread/backgroundTerminals/list"), + ) + .toBe(true); + await new Promise((resolve) => setImmediate(resolve)); + expect(latestBackgroundWork(events)?.items).toMatchObject([{ command: "sleep 180" }]); + terminals = terminals.filter((row) => row.itemId !== "status"); + appServer.completesCommand({ + threadId: "thread-1", + callId: "status", + command: "git status", + output: "clean", + }); + terminals.push({ itemId: "server", processId: "3", command: "npm run dev" }); + appServer.startsTerminalCommand({ + threadId: "thread-1", + turnId: "second", + itemId: "server", + processId: itemIds ? undefined : "3", + command: "npm run dev", + }); + // A process surviving settlement becomes background work even if its item has not completed. + appServer.completeTurn(); + await expect + .poll(() => latestBackgroundWork(events)?.items) + .toMatchObject([{ command: "sleep 180" }, { command: "npm run dev" }]); + expect( + events + .filter((event) => event.type === "background_work") + .flatMap((event) => event.items ?? []) + .some((item) => item.command === "git status"), + ).toBe(false); + terminals = []; + appServer.completesCommand({ + threadId: "thread-1", + callId: "server", + command: "npm run dev", + output: "exited", + }); + await expect.poll(() => latestBackgroundWork(events)?.items).toEqual([]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }, +); + test("Codex runtime safety queries live terminals and retains unsupported app servers", async () => { let response: unknown = { data: [], nextCursor: null }; const appServer = createFakeCodexAppServer({ "thread/backgroundTerminals/list": () => response }); 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 269801fd73..bb7faab9e0 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 @@ -3827,6 +3827,9 @@ export class CodexAppServerAgentSession implements AgentSession { throw this.createClosedError(); } this.client = client; + // Resume can emit autonomous commands before connection setup finishes. Their + // ownership must live for the whole connection, including startup notifications. + this.background.start({ client, thread: null }); client.setUnexpectedTerminationHandler((error) => { this.pendingNotifications = this.queueUnexpectedTermination(error); }); @@ -3854,7 +3857,7 @@ export class CodexAppServerAgentSession implements AgentSession { throw this.createClosedError(); } this.connectionState = "connected"; - this.background.start({ client, thread: this.currentThreadId }); + this.background.changed(this.currentThreadId); } catch (error) { try { if (this.client === client) { @@ -5944,15 +5947,34 @@ export class CodexAppServerAgentSession implements AgentSession { ) { this.settleStreamRetry({ status: "completed" }); } - if ( - parsed.kind === "turn_completed" || - parsed.kind === "exec_command_started" || + this.observeBackgroundWork(parsed); + await this.dispatchParsedNotification(parsed); + } + + private observeBackgroundWork(parsed: ParsedCodexNotification): void { + const backgroundThread = getCodexNotificationThreadId(parsed) ?? this.currentThreadId; + if (backgroundThread && parsed.kind === "turn_started") + this.background.turnStarted(backgroundThread); + else if (backgroundThread && parsed.kind === "turn_completed") + this.background.turnSettled(backgroundThread); + else if (backgroundThread && parsed.kind === "exec_command_started") + this.background.commandStarted({ thread: backgroundThread, itemId: parsed.callId }); + else if ( + backgroundThread && + parsed.kind === "item_started" && + normalizeCodexThreadItemType(parsed.item.type) === "commandExecution" + ) + this.background.commandStarted({ + thread: backgroundThread, + itemId: parsed.item.id ?? null, + processId: typeof parsed.item.processId === "string" ? parsed.item.processId : undefined, + }); + else if ( parsed.kind === "exec_command_completed" || ((parsed.kind === "item_started" || parsed.kind === "item_completed") && normalizeCodexThreadItemType(parsed.item.type) === "commandExecution") ) this.background.changed(this.currentThreadId); - await this.dispatchParsedNotification(parsed); } private async dispatchSubAgentNotification( diff --git a/packages/server/src/server/agent/providers/codex/background-inventory.ts b/packages/server/src/server/agent/providers/codex/background-inventory.ts index 856634dbcb..91f6fcf11c 100644 --- a/packages/server/src/server/agent/providers/codex/background-inventory.ts +++ b/packages/server/src/server/agent/providers/codex/background-inventory.ts @@ -6,12 +6,27 @@ type InventoryEvent = Extract; interface Client { request(method: string, params?: unknown, timeoutMs?: number): Promise; } +interface ForegroundCommands { + items: Set; + processes: Set; +} +interface CommandStartedInput { + thread: string; + itemId: string | null; + processId?: string; +} +interface ReadInventoriesInput { + client: Client; + threads: readonly string[]; + foreground: ReadonlyMap; +} export class CodexBackgroundInventory { private generation = 0; private client: Client | null = null; private thread: string | null = null; private readonly threads = new Set(); + private readonly foreground = new Map(); private origins = new Map(); private pending = false; private dirty = false; @@ -85,6 +100,7 @@ export class CodexBackgroundInventory { this.client = null; this.thread = null; this.threads.clear(); + this.foreground.clear(); this.origins.clear(); this.pending = false; this.dirty = false; @@ -101,6 +117,24 @@ export class CodexBackgroundInventory { this.dirty = true; if (!this.retrying) this.refresh(); } + turnStarted(thread: string) { + if (!this.foreground.has(thread)) + this.foreground.set(thread, { items: new Set(), processes: new Set() }); + } + commandStarted({ thread, itemId, processId }: CommandStartedInput) { + let turn = this.foreground.get(thread); + if (!turn) { + turn = { items: new Set(), processes: new Set() }; + this.foreground.set(thread, turn); + } + if (itemId) turn.items.add(itemId); + if (processId) turn.processes.add(processId); + this.changed(thread); + } + turnSettled(thread: string) { + this.foreground.delete(thread); + this.changed(thread); + } private refresh() { if (this.pending || !this.client || !this.thread || this.unsupported) return; if (this.timer) clearTimeout(this.timer); @@ -113,7 +147,11 @@ export class CodexBackgroundInventory { void (async () => { let failed = false; try { - const { items, origins, activeThreads } = await readInventories(client, threads); + const { items, origins, activeThreads } = await readInventories({ + client, + threads, + foreground: this.foreground, + }); if (generation !== this.generation || this.dirty) return; this.origins = origins; this.threads.clear(); @@ -149,7 +187,7 @@ export class CodexBackgroundInventory { } } -async function readInventories(client: Client, threads: string[]) { +async function readInventories({ client, threads, foreground }: ReadInventoriesInput) { const items: NonNullable = []; const origins = new Map(); const activeThreads = new Set(); @@ -158,7 +196,9 @@ async function readInventories(client: Client, threads: string[]) { do { const page = z .object({ - data: z.array(z.object({ processId: z.string(), command: z.string() })), + data: z.array( + z.object({ itemId: z.string().optional(), processId: z.string(), command: z.string() }), + ), nextCursor: z.string().nullable(), }) .parse( @@ -169,9 +209,17 @@ async function readInventories(client: Client, threads: string[]) { ), ); for (const row of page.data) { + activeThreads.add(threadId); + // Codex registers processes before the initial tool call returns. Only commands + // surviving their owning turn belong in the background-work presentation. + const turn = foreground.get(threadId); + if ( + turn && + ((row.itemId && turn.items.has(row.itemId)) || turn.processes.has(row.processId)) + ) + continue; const id = JSON.stringify([threadId, row.processId]); origins.set(id, { threadId, processId: row.processId }); - activeThreads.add(threadId); items.push({ id, kind: "shell", diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index ccd8a60ac6..2aee26894c 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -33,7 +33,7 @@ interface FakeTerminalCommand { threadId: string; turnId: string; itemId: string; - processId: string; + processId?: string; command: string; } interface FakeTerminalInput {