fix: show Codex subagent activity without streaming child output

packages/agent/README.md

@@ -126,6 +126,8 @@ start()
 
 The two tapping layers are distinct. The inner tap (from `createAcpConnection`) persists to logs. The outer tap (in `AgentServer`) broadcasts to SSE. This means log persistence works for both cloud and local, while SSE broadcast is cloud-only.
 
+Adapters must remove provider notifications that do not belong to the active parent session before writing ACP updates. Persisted ACP updates do not retain enough provider-specific identity to separate a subagent transcript or completion from its parent safely during cloud replay.
+
 ### HTTP endpoints
 
 | Method | Path       | Auth | Description                                              |

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

@@ -134,6 +134,110 @@ 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",
+      _meta: { environment: "cloud", taskRunId: "run_1" },
+    } as unknown as NewSessionRequest);
+    const promptDone = agent.prompt({
+      sessionId: "thr_1",
+      prompt: [{ type: "text", text: "delegate this" }],
+    } as unknown as PromptRequest);
+    stub.emit("item/started", {
+      threadId: "thr_1",
+      turnId: "turn_1",
+      item: {
+        type: "collabAgentToolCall",
+        id: "spawn_1",
+        tool: "spawnAgent",
+        status: "inProgress",
+        senderThreadId: "thr_1",
+        receiverThreadIds: ["subagent_1"],
+        prompt: "Review the implementation",
+      },
+    });
+    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" });
+    const serializedUpdates = JSON.stringify(sessionUpdates);
+    expect(serializedUpdates).toContain("spawn_agent");
+    expect(serializedUpdates).toContain("parent response");
+    expect(serializedUpdates).not.toContain("subagent prose");
+    expect(
+      extNotifications.filter(
+        (notification) => notification.method === "_posthog/turn_complete",
+      ),
+    ).toHaveLength(1);
+  });
+
   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>,

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

@@ -230,6 +230,74 @@ describe("mapAppServerNotification", () => {
     });
   });
 
+  it("maps a spawned Codex agent to an explicit subagent tool call", () => {
+    const result = mapAppServerNotification(
+      "s-1",
+      APP_SERVER_NOTIFICATIONS.ITEM_STARTED,
+      {
+        item: {
+          type: "collabAgentToolCall",
+          id: "spawn-1",
+          tool: "spawnAgent",
+          status: "inProgress",
+          senderThreadId: "main-thread",
+          receiverThreadIds: ["child-thread"],
+          prompt: "Review the authentication changes\nFocus on security.",
+          model: "gpt-5.5",
+          reasoningEffort: "high",
+        },
+      },
+    );
+
+    expect(result).toEqual({
+      sessionId: "s-1",
+      update: {
+        sessionUpdate: "tool_call",
+        toolCallId: "spawn-1",
+        title: "Review the authentication changes",
+        kind: "other",
+        status: "in_progress",
+        rawInput: {
+          prompt: "Review the authentication changes\nFocus on security.",
+          receiverThreadIds: ["child-thread"],
+          model: "gpt-5.5",
+          reasoningEffort: "high",
+        },
+        _meta: { posthog: { toolName: "spawn_agent" } },
+      },
+    });
+  });
+
+  it("keeps a completed spawn tool call active while its subagent is running", () => {
+    const result = mapAppServerNotification(
+      "s-1",
+      APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED,
+      {
+        item: {
+          type: "collabAgentToolCall",
+          id: "spawn-1",
+          tool: "spawnAgent",
+          status: "completed",
+          senderThreadId: "main-thread",
+          receiverThreadIds: ["child-thread"],
+          prompt: "Review the authentication changes",
+          agentsStates: {
+            "child-thread": { status: "running", message: null },
+          },
+        },
+      },
+    );
+
+    expect(result).toEqual({
+      sessionId: "s-1",
+      update: {
+        sessionUpdate: "tool_call_update",
+        toolCallId: "spawn-1",
+        status: "in_progress",
+      },
+    });
+  });
+
   it("drops agent message items (their deltas already streamed)", () => {
     expect(
       mapAppServerNotification("s-1", APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED, {