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 {