From 17b0264fde2246887d1deddf2511c3da7889195b Mon Sep 17 00:00:00 2001 From: Matt Pua Date: Thu, 16 Jul 2026 14:37:24 -0400 Subject: [PATCH 1/3] fix(agent): show Codex subagent activity Generated-By: PostHog Code Task-Id: d8f51f0f-0f93-47ef-9c93-4b70efc02672 --- .../codex-app-server-agent.test.ts | 25 ++-- .../codex-app-server-agent.ts | 115 ++++++++++++++++-- packages/shared/src/index.ts | 1 + packages/shared/src/tool-meta.test.ts | 20 +++ packages/shared/src/tool-meta.ts | 10 +- .../components/buildConversationItems.ts | 6 +- .../incrementalConversationItems.test.ts | 30 +++++ 7 files changed, 183 insertions(+), 24 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 23355aee71..34dec2522c 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -143,7 +143,7 @@ describe("CodexAppServerAgent", () => { }); }); - it("isolates subagent output, usage, compaction, and completion", async () => { + it("surfaces subagent activity while isolating its lifecycle state", async () => { const stub = makeStubRpc({ initialize: {}, "thread/start": { thread: { id: "thr_1" } }, @@ -191,7 +191,6 @@ describe("CodexAppServerAgent", () => { prompt: "Review the implementation", }, }); - const sessionUpdateCount = sessionUpdates.length; const extNotificationCount = extNotifications.length; stub.emit("item/agentMessage/delta", { @@ -215,6 +214,16 @@ describe("CodexAppServerAgent", () => { text: '{"source":"child"}', }, }); + stub.emit("item/started", { + threadId: "subagent_1", + turnId: "subagent_turn_1", + item: { + type: "commandExecution", + id: "shared_command_id", + command: "echo child", + status: "inProgress", + }, + }); stub.emit("item/commandExecution/outputDelta", { threadId: "subagent_1", turnId: "subagent_turn_1", @@ -251,11 +260,9 @@ describe("CodexAppServerAgent", () => { expect({ extNotifications: extNotifications.length, promptSettled, - sessionUpdates: sessionUpdates.length, }).toEqual({ extNotifications: extNotificationCount, promptSettled: false, - sessionUpdates: sessionUpdateCount, }); stub.emit("item/agentMessage/delta", { @@ -292,10 +299,14 @@ describe("CodexAppServerAgent", () => { await expect(promptDone).resolves.toMatchObject({ stopReason: "end_turn" }); const serializedUpdates = JSON.stringify(sessionUpdates); expect(serializedUpdates).toContain("spawn_agent"); + expect(serializedUpdates).toContain("subagent prose"); + expect(serializedUpdates).toContain("subagent reasoning"); + expect(serializedUpdates).toContain("child command output"); + expect(serializedUpdates).toContain( + "subagent:subagent_1:shared_command_id", + ); + expect(serializedUpdates).toContain('"parentToolCallId":"spawn_1"'); expect(serializedUpdates).toContain("parent response"); - expect(serializedUpdates).not.toContain("subagent prose"); - expect(serializedUpdates).not.toContain("subagent reasoning"); - expect(serializedUpdates).not.toContain("child command output"); expect(structuredOutputs).toEqual([{ source: "parent" }]); expect( extNotifications.filter( diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 8b1ce61b89..7a4f2e48c6 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -15,6 +15,7 @@ import type { RequestPermissionResponse, ResumeSessionRequest, ResumeSessionResponse, + SessionNotification, SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, StopReason, @@ -236,6 +237,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { /** Deployment environment; on "cloud" a non-danger sandbox would panic, so we skip the override. */ private environment?: "local" | "cloud"; private readonly commandOutputs = new Map(); + private readonly subagentParents = new Map(); /** Extra writable roots for this session, folded into workspaceWrite sandbox turns. */ private additionalDirectories?: string[]; /** The session workspace stays writable when extra roots are applied per turn. */ @@ -460,6 +462,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { ): Promise<{ threadId: string; thread: AppServerThread | undefined }> { this.cancelNextGoalTurn = false; this.nativeGoalTurnId = undefined; + this.subagentParents.clear(); this.jsonSchema = params.meta?.jsonSchema ?? undefined; this.taskRunId = params.meta?.taskRunId; this.environment = params.meta?.environment; @@ -1159,6 +1162,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { async closeSession(): Promise { this.commandOutputs.clear(); + this.subagentParents.clear(); this.nativeGoalTurnId = undefined; this.session.abortController.abort(); this.session.cancelled = true; @@ -1178,23 +1182,25 @@ export class CodexAppServerAgent extends BaseAcpAgent { const notificationThreadId = readNotificationThreadId(params); const isMainThread = !notificationThreadId || notificationThreadId === this.threadId; + this.captureSubagentRelationship(method, params, notificationThreadId); const mappedParams = isMainThread ? this.withBufferedCommandOutput(method, params) : params; if (this.sessionId && !this.session.cancelled) { - 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); - } + const notification = mapAppServerNotification( + this.sessionId, + method, + mappedParams, + ); + const visibleNotification = isMainThread + ? notification + : this.mapSubagentNotification(notification, notificationThreadId); + if (visibleNotification) { + void this.client + .sessionUpdate(visibleNotification) + .catch((err) => this.logger.warn("sessionUpdate failed", err)); + this.appendNotification(this.sessionId, visibleNotification); } } @@ -1307,6 +1313,87 @@ export class CodexAppServerAgent extends BaseAcpAgent { } } + private captureSubagentRelationship( + method: string, + params: unknown, + senderThreadId: string | undefined, + ): void { + if ( + method !== APP_SERVER_NOTIFICATIONS.ITEM_STARTED && + method !== APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED + ) { + return; + } + const item = (params as { item?: AppServerItem })?.item; + if ( + item?.type !== "collabAgentToolCall" || + item.tool !== "spawnAgent" || + !item.id || + !item.receiverThreadIds?.length + ) { + return; + } + const parentToolCallId = + senderThreadId && senderThreadId !== this.threadId + ? subagentToolCallId(senderThreadId, item.id) + : item.id; + for (const receiverThreadId of item.receiverThreadIds) { + this.subagentParents.set(receiverThreadId, parentToolCallId); + } + } + + private mapSubagentNotification( + notification: SessionNotification | null, + threadId: string | undefined, + ): SessionNotification | null { + if (!notification || !threadId) return null; + const parentToolCallId = this.subagentParents.get(threadId); + if (!parentToolCallId) return null; + const update = notification.update as SessionNotification["update"] & { + _meta?: Record; + toolCallId?: string; + }; + if ( + update.sessionUpdate !== "agent_message_chunk" && + update.sessionUpdate !== "agent_thought_chunk" && + update.sessionUpdate !== "tool_call" && + update.sessionUpdate !== "tool_call_update" + ) { + return null; + } + const toolCallId = update.toolCallId + ? subagentToolCallId(threadId, update.toolCallId) + : undefined; + if (update.sessionUpdate === "tool_call_update") { + return { + ...notification, + update: { ...update, ...(toolCallId ? { toolCallId } : {}) }, + } as SessionNotification; + } + const existingPosthog = (update._meta?.posthog ?? {}) as Record< + string, + unknown + >; + return { + ...notification, + update: { + ...update, + ...(toolCallId ? { toolCallId } : {}), + _meta: { + ...update._meta, + posthog: { + toolName: + typeof existingPosthog.toolName === "string" + ? existingPosthog.toolName + : "subagent_activity", + ...existingPosthog, + parentToolCallId, + }, + }, + }, + } as SessionNotification; + } + private withBufferedCommandOutput(method: string, params: unknown): unknown { if (!params || typeof params !== "object") { return params; @@ -1718,6 +1805,10 @@ function readNotificationThreadId(params: unknown): string | undefined { return typeof threadId === "string" ? threadId : undefined; } +function subagentToolCallId(threadId: string, toolCallId: string): string { + return `subagent:${threadId}:${toolCallId}`; +} + /** The codex thread config override map: folds in MCP servers + makes extra workspace roots writable. Undefined when empty. */ function buildThreadConfig( mcpServers: ReturnType, diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index fe66950795..37a4640262 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -275,6 +275,7 @@ export { readAgentToolName, readMcpToolDescriptor, readMcpToolName, + readParentToolCallId, } from "./tool-meta"; export { TypedEventEmitter } from "./typed-event-emitter"; export { isSafeExternalUrl } from "./url"; diff --git a/packages/shared/src/tool-meta.test.ts b/packages/shared/src/tool-meta.test.ts index 8e718d8616..cf3fbd61ef 100644 --- a/packages/shared/src/tool-meta.test.ts +++ b/packages/shared/src/tool-meta.test.ts @@ -4,6 +4,7 @@ import { readAgentToolName, readMcpToolDescriptor, readMcpToolName, + readParentToolCallId, } from "./tool-meta"; describe("parseMcpToolName", () => { @@ -49,6 +50,25 @@ describe("readAgentToolName", () => { }); }); +describe("readParentToolCallId", () => { + it("prefers the posthog channel over the legacy claudeCode fallback", () => { + expect( + readParentToolCallId({ + posthog: { toolName: "Bash", parentToolCallId: "parent-1" }, + claudeCode: { parentToolCallId: "stale" }, + }), + ).toBe("parent-1"); + }); + + it("falls back to claudeCode when posthog is absent", () => { + expect( + readParentToolCallId({ + claudeCode: { parentToolCallId: "parent-2" }, + }), + ).toBe("parent-2"); + }); +}); + describe("readMcpToolDescriptor / readMcpToolName", () => { it("uses the structured mcp descriptor when present (no name parsing)", () => { const meta = { diff --git a/packages/shared/src/tool-meta.ts b/packages/shared/src/tool-meta.ts index 8ef62cc324..ef0d7bfef8 100644 --- a/packages/shared/src/tool-meta.ts +++ b/packages/shared/src/tool-meta.ts @@ -12,6 +12,8 @@ export interface PosthogToolMeta { toolName: string; /** Set only for MCP tool calls — the originating server + tool. */ mcp?: { server: string; tool: string }; + /** Parent subagent tool call for nested activity. */ + parentToolCallId?: string; } /** `_meta` fragment for adapters to spread onto a tool_call update. */ @@ -45,7 +47,7 @@ export function parseMcpToolName( interface ToolCallMeta { posthog?: PosthogToolMeta; /** Legacy Claude-adapter channel, read only as a fallback. */ - claudeCode?: { toolName?: string }; + claudeCode?: { toolName?: string; parentToolCallId?: string }; } function asToolCallMeta(meta: unknown): ToolCallMeta | undefined { @@ -58,6 +60,12 @@ export function readAgentToolName(meta: unknown): string | undefined { return m?.posthog?.toolName ?? m?.claudeCode?.toolName; } +/** Parent subagent tool call: neutral channel first, legacy fallback. */ +export function readParentToolCallId(meta: unknown): string | undefined { + const m = asToolCallMeta(meta); + return m?.posthog?.parentToolCallId ?? m?.claudeCode?.parentToolCallId; +} + /** * The MCP `{ server, tool }` descriptor for a tool call, or undefined for a * non-MCP call. Prefers the structured channel, else parses the legacy diff --git a/packages/ui/src/features/sessions/components/buildConversationItems.ts b/packages/ui/src/features/sessions/components/buildConversationItems.ts index d4386b921e..5747935a04 100644 --- a/packages/ui/src/features/sessions/components/buildConversationItems.ts +++ b/packages/ui/src/features/sessions/components/buildConversationItems.ts @@ -12,6 +12,7 @@ import { isJsonRpcNotification, isJsonRpcRequest, isJsonRpcResponse, + readParentToolCallId, type UserShellExecuteParams, } from "@posthog/shared"; import { @@ -747,10 +748,7 @@ function extractUserPrompt(params: unknown): { } function getParentToolCallId(update: SessionUpdate): string | undefined { - const meta = (update as Record)?._meta as - | { claudeCode?: { parentToolCallId?: string } } - | undefined; - return meta?.claudeCode?.parentToolCallId; + return readParentToolCallId((update as Record)._meta); } function pushChildItem(b: ItemBuilder, parentId: string, update: RenderItem) { diff --git a/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts b/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts index d793141e2c..d90366e7ec 100644 --- a/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts +++ b/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts @@ -475,4 +475,34 @@ describe("createIncrementalConversationBuilder", () => { expect(row2.turnContext.childItems).not.toBe(row1.turnContext.childItems); expect(row2.turnContext.childItems.get("agent1")?.length).toBe(1); }); + + it("groups canonical PostHog child metadata under its subagent", () => { + const inc = createIncrementalConversationBuilder(); + const messages = [ + userPromptMsg(1, 1, "go"), + toolCallMsg(2, "agent1", { + _meta: { posthog: { toolName: "spawn_agent" } }, + }), + updateMsg(3, { + sessionUpdate: "tool_call", + toolCallId: "child1", + kind: "read", + status: "pending", + title: "child1", + _meta: { + posthog: { + toolName: "subagent_activity", + parentToolCallId: "agent1", + }, + }, + }), + ]; + + const result = inc.update(messages, true); + const row = result.items.find((item) => item.type === "session_update"); + if (row?.type !== "session_update") { + throw new Error("expected agent session_update row"); + } + expect(row.turnContext.childItems.get("agent1")?.length).toBe(1); + }); }); From 7b8a5838973c1d99a25bc20beac95ccfa2889b36 Mon Sep 17 00:00:00 2001 From: Matt Pua Date: Thu, 16 Jul 2026 14:52:50 -0400 Subject: [PATCH 2/3] fix(agent): preserve resumed subagent activity Generated-By: PostHog Code Task-Id: 38f87de0-3672-400f-af61-378f74c17f2b --- .../codex-app-server-agent.test.ts | 127 ++++++++++++++++++ .../codex-app-server-agent.ts | 22 ++- packages/shared/src/tool-meta.test.ts | 18 +++ packages/shared/src/tool-meta.ts | 5 +- .../components/buildConversationItems.ts | 17 +++ .../incrementalConversationItems.test.ts | 40 ++++++ 6 files changed, 227 insertions(+), 2 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 34dec2522c..9b69db94d8 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -315,6 +315,86 @@ describe("CodexAppServerAgent", () => { ).toHaveLength(1); }); + it.each(["resumeAgent", "sendInput"])( + "attaches child activity to the current %s call", + async (collaborationTool) => { + let turnNumber = 0; + const stub = makeStubRpc({ + initialize: {}, + "thread/start": { thread: { id: "thr_1" } }, + "turn/start": () => ({ + turn: { + id: `turn_${++turnNumber}`, + status: "inProgress", + }, + }), + }); + const { client, sessionUpdates } = 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 firstPrompt = agent.prompt({ + sessionId: "thr_1", + prompt: [{ type: "text", text: "spawn" }], + } as unknown as PromptRequest); + stub.emit("item/started", { + threadId: "thr_1", + turnId: "turn_1", + item: { + type: "collabAgentToolCall", + id: "spawn_1", + tool: "spawnAgent", + receiverThreadIds: ["subagent_1"], + status: "inProgress", + }, + }); + stub.emit("turn/completed", { + threadId: "thr_1", + turn: { id: "turn_1", status: "completed" }, + }); + await firstPrompt; + + const secondPrompt = agent.prompt({ + sessionId: "thr_1", + prompt: [{ type: "text", text: "continue" }], + } as unknown as PromptRequest); + const currentCallId = `${collaborationTool}_1`; + stub.emit("item/started", { + threadId: "thr_1", + turnId: "turn_2", + item: { + type: "collabAgentToolCall", + id: currentCallId, + tool: collaborationTool, + receiverThreadIds: ["subagent_1"], + status: "inProgress", + }, + }); + stub.emit("item/agentMessage/delta", { + threadId: "subagent_1", + turnId: "subagent_turn_2", + itemId: "message_2", + delta: "continued work", + }); + + expect(JSON.stringify(sessionUpdates)).toContain( + `"parentToolCallId":"${currentCallId}"`, + ); + + stub.emit("turn/completed", { + threadId: "thr_1", + turn: { id: "turn_2", status: "completed" }, + }); + await secondPrompt; + }, + ); + it.each([ { label: "reads an empty goal", @@ -2477,6 +2557,53 @@ describe("CodexAppServerAgent", () => { }); }); + it("restores subagent relationships from resumed thread history", async () => { + const stub = makeStubRpc({ + initialize: {}, + "thread/resume": { + thread: { + id: "t1", + turns: [ + { + items: [ + { + type: "collabAgentToolCall", + id: "spawn_1", + tool: "spawnAgent", + receiverThreadIds: ["subagent_1"], + status: "completed", + }, + ], + }, + ], + }, + }, + }); + const { client, sessionUpdates } = makeFakeClient(); + const agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/x/codex" }, + model: "gpt-5.5", + rpcFactory: stub.factory, + }); + await agent.initialize(init); + await agent.resumeSession({ + sessionId: "t1", + cwd: "/r", + mcpServers: [], + } as unknown as Parameters[0]); + + stub.emit("item/agentMessage/delta", { + threadId: "subagent_1", + turnId: "subagent_turn_1", + itemId: "message_1", + delta: "still working", + }); + + expect(JSON.stringify(sessionUpdates)).toContain( + '"parentToolCallId":"spawn_1"', + ); + }); + it("forwards additionalDirectories to thread/start as writable_roots", async () => { const stub = makeStubRpc({ "thread/start": { thread: { id: "t" } } }); const { client } = makeFakeClient(); diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 7a4f2e48c6..4ce2ee0c30 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -518,6 +518,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { } this.threadId = threadId; this.sessionId = threadId; + this.restoreSubagentRelationships(thread); if (method === APP_SERVER_METHODS.THREAD_START && params.meta?.nativeGoal) { await this.restoreGoal(params.meta.nativeGoal); } @@ -1325,9 +1326,28 @@ export class CodexAppServerAgent extends BaseAcpAgent { return; } const item = (params as { item?: AppServerItem })?.item; + this.captureSubagentRelationshipItem(item, senderThreadId); + } + + private restoreSubagentRelationships( + thread: AppServerThread | undefined, + ): void { + for (const turn of thread?.turns ?? []) { + for (const item of turn.items ?? []) { + this.captureSubagentRelationshipItem(item, item.senderThreadId); + } + } + } + + private captureSubagentRelationshipItem( + item: AppServerItem | undefined, + senderThreadId: string | undefined, + ): void { if ( item?.type !== "collabAgentToolCall" || - item.tool !== "spawnAgent" || + (item.tool !== "spawnAgent" && + item.tool !== "resumeAgent" && + item.tool !== "sendInput") || !item.id || !item.receiverThreadIds?.length ) { diff --git a/packages/shared/src/tool-meta.test.ts b/packages/shared/src/tool-meta.test.ts index cf3fbd61ef..77e0bdd02b 100644 --- a/packages/shared/src/tool-meta.test.ts +++ b/packages/shared/src/tool-meta.test.ts @@ -67,6 +67,24 @@ describe("readParentToolCallId", () => { }), ).toBe("parent-2"); }); + + it("ignores malformed canonical metadata and uses a valid legacy fallback", () => { + expect( + readParentToolCallId({ + posthog: { toolName: "Bash", parentToolCallId: {} }, + claudeCode: { parentToolCallId: "parent-3" }, + }), + ).toBe("parent-3"); + }); + + it("returns undefined for empty or non-string parent ids", () => { + expect( + readParentToolCallId({ posthog: { parentToolCallId: "" } }), + ).toBeUndefined(); + expect( + readParentToolCallId({ claudeCode: { parentToolCallId: 123 } }), + ).toBeUndefined(); + }); }); describe("readMcpToolDescriptor / readMcpToolName", () => { diff --git a/packages/shared/src/tool-meta.ts b/packages/shared/src/tool-meta.ts index ef0d7bfef8..afbfde8049 100644 --- a/packages/shared/src/tool-meta.ts +++ b/packages/shared/src/tool-meta.ts @@ -63,7 +63,10 @@ export function readAgentToolName(meta: unknown): string | undefined { /** Parent subagent tool call: neutral channel first, legacy fallback. */ export function readParentToolCallId(meta: unknown): string | undefined { const m = asToolCallMeta(meta); - return m?.posthog?.parentToolCallId ?? m?.claudeCode?.parentToolCallId; + const canonical = m?.posthog?.parentToolCallId; + if (typeof canonical === "string" && canonical.length > 0) return canonical; + const legacy = m?.claudeCode?.parentToolCallId; + return typeof legacy === "string" && legacy.length > 0 ? legacy : undefined; } /** diff --git a/packages/ui/src/features/sessions/components/buildConversationItems.ts b/packages/ui/src/features/sessions/components/buildConversationItems.ts index 5747935a04..60cf5f5dd9 100644 --- a/packages/ui/src/features/sessions/components/buildConversationItems.ts +++ b/packages/ui/src/features/sessions/components/buildConversationItems.ts @@ -174,7 +174,17 @@ function isThoughtItem( } export function markThoughtCompletion(items: ConversationItem[]) { + markThoughtCompletionInItems(items, new Set()); +} + +function markThoughtCompletionInItems( + items: ConversationItem[], + visited: Set, +) { + if (visited.has(items)) return; + visited.add(items); const seenContexts = new Set(); + const itemContexts = new Set(); for (let i = items.length - 1; i >= 0; i--) { const item = items[i]; @@ -186,6 +196,13 @@ export function markThoughtCompletion(items: ConversationItem[]) { if (item.type === "session_update") { seenContexts.add(item.turnContext); + itemContexts.add(item.turnContext); + } + } + + for (const context of itemContexts) { + for (const children of context.childItems.values()) { + markThoughtCompletionInItems(children, visited); } } } diff --git a/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts b/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts index d90366e7ec..ee98a728e7 100644 --- a/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts +++ b/packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts @@ -143,6 +143,22 @@ const childToolCallMsg = ( _meta: { claudeCode: { parentToolCallId } }, }); +const childThoughtChunk = ( + ts: number, + text: string, + parentToolCallId: string, +) => + updateMsg(ts, { + sessionUpdate: "agent_thought_chunk", + content: { type: "text", text }, + _meta: { + posthog: { + toolName: "subagent_activity", + parentToolCallId, + }, + }, + }); + // --- normalization (cycle-free, Map-resolved) ----------------------------- function normContext(ctx: TurnContext) { @@ -505,4 +521,28 @@ describe("createIncrementalConversationBuilder", () => { } expect(row.turnContext.childItems.get("agent1")?.length).toBe(1); }); + + it("marks nested subagent thoughts complete when the turn finishes", () => { + const inc = createIncrementalConversationBuilder(); + const messages = [ + userPromptMsg(1, 1, "go"), + toolCallMsg(2, "agent1", { + _meta: { posthog: { toolName: "spawn_agent" } }, + }), + childThoughtChunk(3, "investigating", "agent1"), + promptResponseMsg(4, 1), + ]; + + const result = inc.update(messages, false); + const row = result.items.find((item) => item.type === "session_update"); + if (row?.type !== "session_update") { + throw new Error("expected agent session_update row"); + } + const thought = row.turnContext.childItems.get("agent1")?.[0]; + expect(thought).toMatchObject({ + type: "session_update", + thoughtComplete: true, + update: { sessionUpdate: "agent_thought_chunk" }, + }); + }); }); From a1b390c24da71f49019813f195c684198a991b1d Mon Sep 17 00:00:00 2001 From: Matt Pua Date: Thu, 16 Jul 2026 15:22:11 -0400 Subject: [PATCH 3/3] fix(agent): buffer early Codex subagent activity Preserve child-thread messages, reasoning, and tool activity that arrive before the parent collaboration call is observed. Generated-By: PostHog Code Task-Id: acb27526-5140-4f34-9024-1df8d5c241a8 --- .../codex-app-server-agent.test.ts | 56 +++++++++++ .../codex-app-server-agent.ts | 94 ++++++++++++++----- 2 files changed, 128 insertions(+), 22 deletions(-) diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 9b69db94d8..597a324d45 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -395,6 +395,62 @@ describe("CodexAppServerAgent", () => { }, ); + it("buffers child activity until its parent tool call arrives", async () => { + const stub = makeStubRpc({ + initialize: {}, + "thread/start": { thread: { id: "thr_1" } }, + "turn/start": { turn: { id: "turn_1", status: "inProgress" } }, + }); + const { client, sessionUpdates } = 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" }], + } as unknown as PromptRequest); + + stub.emit("item/agentMessage/delta", { + threadId: "subagent_1", + turnId: "subagent_turn_1", + itemId: "message_1", + delta: "early child activity", + }); + expect(JSON.stringify(sessionUpdates)).not.toContain( + "early child activity", + ); + + stub.emit("item/started", { + threadId: "thr_1", + turnId: "turn_1", + item: { + type: "collabAgentToolCall", + id: "spawn_1", + tool: "spawnAgent", + receiverThreadIds: ["subagent_1"], + status: "inProgress", + }, + }); + + const serializedUpdates = JSON.stringify(sessionUpdates); + expect(serializedUpdates).toContain("early child activity"); + expect(serializedUpdates).toContain('"parentToolCallId":"spawn_1"'); + expect(serializedUpdates.indexOf("spawn_agent")).toBeLessThan( + serializedUpdates.indexOf("early child activity"), + ); + + stub.emit("turn/completed", { + threadId: "thr_1", + turn: { id: "turn_1", status: "completed" }, + }); + await promptDone; + }); + it.each([ { label: "reads an empty goal", diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 4ce2ee0c30..9670c5d698 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -238,6 +238,10 @@ export class CodexAppServerAgent extends BaseAcpAgent { private environment?: "local" | "cloud"; private readonly commandOutputs = new Map(); private readonly subagentParents = new Map(); + private readonly pendingSubagentNotifications = new Map< + string, + SessionNotification[] + >(); /** Extra writable roots for this session, folded into workspaceWrite sandbox turns. */ private additionalDirectories?: string[]; /** The session workspace stays writable when extra roots are applied per turn. */ @@ -463,6 +467,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { this.cancelNextGoalTurn = false; this.nativeGoalTurnId = undefined; this.subagentParents.clear(); + this.pendingSubagentNotifications.clear(); this.jsonSchema = params.meta?.jsonSchema ?? undefined; this.taskRunId = params.meta?.taskRunId; this.environment = params.meta?.environment; @@ -1164,6 +1169,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { async closeSession(): Promise { this.commandOutputs.clear(); this.subagentParents.clear(); + this.pendingSubagentNotifications.clear(); this.nativeGoalTurnId = undefined; this.session.abortController.abort(); this.session.cancelled = true; @@ -1183,7 +1189,11 @@ export class CodexAppServerAgent extends BaseAcpAgent { const notificationThreadId = readNotificationThreadId(params); const isMainThread = !notificationThreadId || notificationThreadId === this.threadId; - this.captureSubagentRelationship(method, params, notificationThreadId); + const relatedSubagentThreadIds = this.captureSubagentRelationship( + method, + params, + notificationThreadId, + ); const mappedParams = isMainThread ? this.withBufferedCommandOutput(method, params) : params; @@ -1198,10 +1208,19 @@ export class CodexAppServerAgent extends BaseAcpAgent { ? notification : this.mapSubagentNotification(notification, notificationThreadId); if (visibleNotification) { - void this.client - .sessionUpdate(visibleNotification) - .catch((err) => this.logger.warn("sessionUpdate failed", err)); - this.appendNotification(this.sessionId, visibleNotification); + this.emitSessionNotification(visibleNotification); + } else if ( + notification && + notificationThreadId && + isSubagentActivityNotification(notification) + ) { + const pending = + this.pendingSubagentNotifications.get(notificationThreadId) ?? []; + pending.push(notification); + this.pendingSubagentNotifications.set(notificationThreadId, pending); + } + for (const threadId of relatedSubagentThreadIds) { + this.flushSubagentNotifications(threadId); } } @@ -1318,15 +1337,15 @@ export class CodexAppServerAgent extends BaseAcpAgent { method: string, params: unknown, senderThreadId: string | undefined, - ): void { + ): string[] { if ( method !== APP_SERVER_NOTIFICATIONS.ITEM_STARTED && method !== APP_SERVER_NOTIFICATIONS.ITEM_COMPLETED ) { - return; + return []; } const item = (params as { item?: AppServerItem })?.item; - this.captureSubagentRelationshipItem(item, senderThreadId); + return this.captureSubagentRelationshipItem(item, senderThreadId); } private restoreSubagentRelationships( @@ -1342,7 +1361,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { private captureSubagentRelationshipItem( item: AppServerItem | undefined, senderThreadId: string | undefined, - ): void { + ): string[] { if ( item?.type !== "collabAgentToolCall" || (item.tool !== "spawnAgent" && @@ -1351,7 +1370,7 @@ export class CodexAppServerAgent extends BaseAcpAgent { !item.id || !item.receiverThreadIds?.length ) { - return; + return []; } const parentToolCallId = senderThreadId && senderThreadId !== this.threadId @@ -1360,6 +1379,30 @@ export class CodexAppServerAgent extends BaseAcpAgent { for (const receiverThreadId of item.receiverThreadIds) { this.subagentParents.set(receiverThreadId, parentToolCallId); } + return item.receiverThreadIds; + } + + private emitSessionNotification(notification: SessionNotification): void { + if (!this.sessionId) return; + void this.client + .sessionUpdate(notification) + .catch((err) => this.logger.warn("sessionUpdate failed", err)); + this.appendNotification(this.sessionId, notification); + } + + private flushSubagentNotifications(threadId: string): void { + const pending = this.pendingSubagentNotifications.get(threadId); + if (!pending) return; + this.pendingSubagentNotifications.delete(threadId); + for (const notification of pending) { + const visibleNotification = this.mapSubagentNotification( + notification, + threadId, + ); + if (visibleNotification) { + this.emitSessionNotification(visibleNotification); + } + } } private mapSubagentNotification( @@ -1369,18 +1412,8 @@ export class CodexAppServerAgent extends BaseAcpAgent { if (!notification || !threadId) return null; const parentToolCallId = this.subagentParents.get(threadId); if (!parentToolCallId) return null; - const update = notification.update as SessionNotification["update"] & { - _meta?: Record; - toolCallId?: string; - }; - if ( - update.sessionUpdate !== "agent_message_chunk" && - update.sessionUpdate !== "agent_thought_chunk" && - update.sessionUpdate !== "tool_call" && - update.sessionUpdate !== "tool_call_update" - ) { - return null; - } + if (!isSubagentActivityNotification(notification)) return null; + const update = notification.update; const toolCallId = update.toolCallId ? subagentToolCallId(threadId, update.toolCallId) : undefined; @@ -1829,6 +1862,23 @@ function subagentToolCallId(threadId: string, toolCallId: string): string { return `subagent:${threadId}:${toolCallId}`; } +function isSubagentActivityNotification( + notification: SessionNotification, +): notification is SessionNotification & { + update: SessionNotification["update"] & { + _meta?: Record; + toolCallId?: string; + }; +} { + const { sessionUpdate } = notification.update; + return ( + sessionUpdate === "agent_message_chunk" || + sessionUpdate === "agent_thought_chunk" || + sessionUpdate === "tool_call" || + sessionUpdate === "tool_call_update" + ); +} + /** The codex thread config override map: folds in MCP servers + makes extra workspace roots writable. Undefined when empty. */ function buildThreadConfig( mcpServers: ReturnType,