Fix Codex subagent thread isolation

Keep child-thread output, usage, compaction, and completion notifications from mutating the parent ACP session.

Generated-By: PostHog Code
Task-Id: 971044c5-e622-4f07-b1b5-dca9b06d040c

Author: richardsolomou

packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts

@@ -134,6 +134,86 @@ describe("CodexAppServerAgent", () => {
     });
   });
 
+  it("isolates subagent output, usage, compaction, and completion", async () => {
+    const stub = makeStubRpc({
+      initialize: {},
+      "thread/start": { thread: { id: "thr_1" } },
+      "turn/start": { turn: { id: "turn_1", status: "inProgress" } },
+    });
+    const { client, sessionUpdates, extNotifications } = makeFakeClient();
+    const agent = new CodexAppServerAgent(client, {
+      processOptions: { binaryPath: "/bundle/codex" },
+      model: "gpt-5.5",
+      rpcFactory: stub.factory,
+    });
+
+    await agent.initialize(init);
+    await agent.newSession({ cwd: "/repo" } as unknown as NewSessionRequest);
+    const promptDone = agent.prompt({
+      sessionId: "thr_1",
+      prompt: [{ type: "text", text: "delegate this" }],
+    } as unknown as PromptRequest);
+    const sessionUpdateCount = sessionUpdates.length;
+    const extNotificationCount = extNotifications.length;
+
+    stub.emit("item/agentMessage/delta", {
+      threadId: "subagent_1",
+      turnId: "subagent_turn_1",
+      itemId: "subagent_message_1",
+      delta: "subagent prose",
+    });
+    stub.emit("thread/tokenUsage/updated", {
+      threadId: "subagent_1",
+      tokenUsage: {
+        total: { totalTokens: 9000 },
+        modelContextWindow: 10000,
+      },
+    });
+    stub.emit("item/started", {
+      threadId: "subagent_1",
+      turnId: "subagent_turn_1",
+      item: { type: "contextCompaction", id: "subagent_compaction_1" },
+    });
+    stub.emit("item/completed", {
+      threadId: "subagent_1",
+      turnId: "subagent_turn_1",
+      item: { type: "contextCompaction", id: "subagent_compaction_1" },
+    });
+    stub.emit("turn/completed", {
+      threadId: "subagent_1",
+      turn: { id: "subagent_turn_1", status: "completed" },
+    });
+
+    let promptSettled = false;
+    void promptDone.then(() => {
+      promptSettled = true;
+    });
+    await Promise.resolve();
+    expect({
+      extNotifications: extNotifications.length,
+      promptSettled,
+      sessionUpdates: sessionUpdates.length,
+    }).toEqual({
+      extNotifications: extNotificationCount,
+      promptSettled: false,
+      sessionUpdates: sessionUpdateCount,
+    });
+
+    stub.emit("item/agentMessage/delta", {
+      threadId: "thr_1",
+      turnId: "turn_1",
+      itemId: "message_1",
+      delta: "parent response",
+    });
+    stub.emit("turn/completed", {
+      threadId: "thr_1",
+      turn: { id: "turn_1", status: "completed" },
+    });
+
+    await expect(promptDone).resolves.toMatchObject({ stopReason: "end_turn" });
+    expect(JSON.stringify(sessionUpdates)).toContain("parent response");
+  });
+
   it("includes buffered command output when completion omits aggregatedOutput", async () => {
     const stub = makeStubRpc({
       initialize: {},

packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts

@@ -938,32 +938,40 @@ export class CodexAppServerAgent extends BaseAcpAgent {
 
   private handleNotification(method: string, params: unknown): void {
     const mappedParams = this.withBufferedCommandOutput(method, params);
+    const notificationThreadId = readNotificationThreadId(mappedParams);
+    const isMainThread =
+      !notificationThreadId || notificationThreadId === this.threadId;
+
     if (this.sessionId && !this.session.cancelled) {
-      const notification = mapAppServerNotification(
-        this.sessionId,
-        method,
-        mappedParams,
-      );
-      if (notification) {
-        void this.client
-          .sessionUpdate(notification)
-          .catch((err) => this.logger.warn("sessionUpdate failed", err));
-        this.appendNotification(this.sessionId, notification);
+      if (isMainThread) {
+        const notification = mapAppServerNotification(
+          this.sessionId,
+          method,
+          mappedParams,
+        );
+        if (notification) {
+          void this.client
+            .sessionUpdate(notification)
+            .catch((err) => this.logger.warn("sessionUpdate failed", err));
+          this.appendNotification(this.sessionId, notification);
+        }
       }
     }
 
-    if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) {
-      // Capture the active turn id (steer precondition / interrupt target).
-      this.turns.onStarted((params as { turn?: { id?: string } })?.turn?.id);
-    }
-
     if (method === APP_SERVER_NOTIFICATIONS.ITEM_STARTED) {
       this.mcp.capture(params);
     }
     if (method === APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED) {
       this.mcp.release(params);
     }
 
+    if (!isMainThread) return;
+
+    if (method === APP_SERVER_NOTIFICATIONS.TURN_STARTED) {
+      // Capture the active turn id (steer precondition / interrupt target).
+      this.turns.onStarted((params as { turn?: { id?: string } })?.turn?.id);
+    }
+
     // codex auto-compaction surfaces as a contextCompaction item: item/started → in progress,
     // item/completed → boundary (codex emits no separate thread/compacted; that's a guarded
     // fallback). compactionActive dedupes to one boundary per compaction.
@@ -1437,6 +1445,12 @@ function mapTurnStopReason(status: string | undefined): StopReason {
   return "end_turn";
 }
 
+function readNotificationThreadId(params: unknown): string | undefined {
+  if (!params || typeof params !== "object") return undefined;
+  const threadId = (params as { threadId?: unknown }).threadId;
+  return typeof threadId === "string" ? threadId : undefined;
+}
+
 /** The codex thread config override map: folds in MCP servers + makes extra workspace roots writable. Undefined when empty. */
 function buildThreadConfig(
   mcpServers: ReturnType<typeof toCodexMcpServers>,