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.

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<void>((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<void>((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 });

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(

packages/server/src/server/agent/providers/codex/background-inventory.ts

@@ -6,12 +6,27 @@ type InventoryEvent = Extract<AgentStreamEvent, { type: "background_work" }>;
 interface Client {
   request(method: string, params?: unknown, timeoutMs?: number): Promise<unknown>;
 }
+interface ForegroundCommands {
+  items: Set<string>;
+  processes: Set<string>;
+}
+interface CommandStartedInput {
+  thread: string;
+  itemId: string | null;
+  processId?: string;
+}
+interface ReadInventoriesInput {
+  client: Client;
+  threads: readonly string[];
+  foreground: ReadonlyMap<string, ForegroundCommands>;
+}
 
 export class CodexBackgroundInventory {
   private generation = 0;
   private client: Client | null = null;
   private thread: string | null = null;
   private readonly threads = new Set<string>();
+  private readonly foreground = new Map<string, ForegroundCommands>();
   private origins = new Map<string, { threadId: string; processId: string }>();
   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<InventoryEvent["items"]> = [];
   const origins = new Map<string, { threadId: string; processId: string }>();
   const activeThreads = new Set<string>();
@@ -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",

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 {