From 771f164e2675c8f3dd901a279e50e967a1258f36 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 2 May 2026 17:08:07 +0800 Subject: [PATCH] Stream Codex sub-agents as tool calls (#672) --- .../src/server/agent/activity-curator.test.ts | 32 +++ .../src/server/agent/activity-curator.ts | 92 +++++-- .../providers/codex-app-server-agent.test.ts | 97 +++++++ .../agent/providers/codex-app-server-agent.ts | 252 ++++++++++++++++-- .../providers/codex/tool-call-mapper.test.ts | 29 ++ .../agent/providers/codex/tool-call-mapper.ts | 91 ++++++- 6 files changed, 549 insertions(+), 44 deletions(-) diff --git a/packages/server/src/server/agent/activity-curator.test.ts b/packages/server/src/server/agent/activity-curator.test.ts index cceebd458..d4b5990ad 100644 --- a/packages/server/src/server/agent/activity-curator.test.ts +++ b/packages/server/src/server/agent/activity-curator.test.ts @@ -237,3 +237,35 @@ second line'`, expect(curateAgentActivity([])).toBe("No activity to display."); }); }); + +describe("curateAgentActivityActions", () => { + it("converts curated activity lines into sub-agent action entries", async () => { + const { curateAgentActivityActions } = await import("./activity-curator.js"); + const timeline: AgentTimelineItem[] = [ + { type: "assistant_message", text: "Inspecting the repo" }, + toolCallItem({ + callId: "shell-1", + name: "shell", + detail: { + type: "shell", + command: "npm test", + output: "ok", + exitCode: 0, + }, + }), + ]; + + expect(curateAgentActivityActions(timeline)).toEqual([ + { + index: 1, + toolName: "Assistant", + summary: "Inspecting the repo", + }, + { + index: 2, + toolName: "Shell", + summary: "npm test", + }, + ]); + }); +}); diff --git a/packages/server/src/server/agent/activity-curator.ts b/packages/server/src/server/agent/activity-curator.ts index ba48a3877..2172c2e33 100644 --- a/packages/server/src/server/agent/activity-curator.ts +++ b/packages/server/src/server/agent/activity-curator.ts @@ -7,6 +7,16 @@ const DEFAULT_MAX_ITEMS = 0; const MAX_TOOL_INPUT_CHARS = 400; const MAX_TOOL_SUMMARY_CHARS = 200; +interface ActivityAction { + toolName: string; + summary?: string; +} + +interface ActivityEntry { + text: string; + action: ActivityAction; +} + function appendText(buffer: string, text: string): string { const normalized = text.trim(); if (!normalized) { @@ -18,12 +28,24 @@ function appendText(buffer: string, text: string): string { return `${buffer}\n${normalized}`; } -function flushBuffers(lines: string[], buffers: { message: string; thought: string }) { +function activityEntry(text: string, toolName: string, summary?: string): ActivityEntry { + return { + text, + action: { + toolName, + ...(summary ? { summary } : {}), + }, + }; +} + +function flushBuffers(entries: ActivityEntry[], buffers: { message: string; thought: string }) { if (buffers.message.trim()) { - lines.push(buffers.message.trim()); + const text = buffers.message.trim(); + entries.push(activityEntry(text, "Assistant", text)); } if (buffers.thought.trim()) { - lines.push(`[Thought] ${buffers.thought.trim()}`); + const text = buffers.thought.trim(); + entries.push(activityEntry(`[Thought] ${text}`, "Thought", text)); } buffers.message = ""; buffers.thought = ""; @@ -76,15 +98,12 @@ function projectForCuration(items: readonly AgentTimelineItem[]): AgentTimelineI return projectTimelineRows({ rows, mode: "projected" }).map((entry) => entry.item); } -/** - * Convert normalized agent timeline items into a concise text summary. - */ -export function curateAgentActivity( +function curateAgentActivityEntries( timeline: AgentTimelineItem[], options?: { maxItems?: number }, -): string { +): ActivityEntry[] { if (timeline.length === 0) { - return "No activity to display."; + return []; } const collapsed = projectForCuration(timeline); @@ -93,14 +112,14 @@ export function curateAgentActivity( const recentItems = maxItems > 0 && collapsed.length > maxItems ? collapsed.slice(-maxItems) : collapsed; - const lines: string[] = []; + const entries: ActivityEntry[] = []; const buffers = { message: "", thought: "" }; for (const item of recentItems) { switch (item.type) { case "user_message": - flushBuffers(lines, buffers); - lines.push(`[User] ${item.text.trim()}`); + flushBuffers(entries, buffers); + entries.push(activityEntry(`[User] ${item.text.trim()}`, "User", item.text.trim())); break; case "assistant_message": buffers.message = appendText(buffers.message, item.text); @@ -109,7 +128,7 @@ export function curateAgentActivity( buffers.thought = appendText(buffers.thought, item.text); break; case "tool_call": { - flushBuffers(lines, buffers); + flushBuffers(entries, buffers); const inputJson = formatToolInputJson(inputFromUnknownDetail(item.detail)); const display = buildToolCallDisplayModel({ name: item.name, @@ -121,36 +140,59 @@ export function curateAgentActivity( const displayName = display.displayName; const summary = formatToolSummary(display.summary); if (isLikelyExternalToolName(item.name) && inputJson) { - lines.push(`[${displayName}] ${inputJson}`); + entries.push(activityEntry(`[${displayName}] ${inputJson}`, displayName, inputJson)); break; } if (summary) { - lines.push(`[${displayName}] ${summary}`); + entries.push(activityEntry(`[${displayName}] ${summary}`, displayName, summary)); } else { - lines.push(`[${displayName}]`); + entries.push(activityEntry(`[${displayName}]`, displayName)); } break; } case "todo": - flushBuffers(lines, buffers); - lines.push("[Tasks]"); + flushBuffers(entries, buffers); + entries.push(activityEntry("[Tasks]", "Tasks")); for (const entry of item.items) { const checkbox = entry.completed ? "[x]" : "[ ]"; - lines.push(`- ${checkbox} ${entry.text}`); + const text = `- ${checkbox} ${entry.text}`; + entries.push(activityEntry(text, "Assistant", text)); } break; case "error": - flushBuffers(lines, buffers); - lines.push(`[Error] ${item.message}`); + flushBuffers(entries, buffers); + entries.push(activityEntry(`[Error] ${item.message}`, "Error", item.message)); break; case "compaction": - flushBuffers(lines, buffers); - lines.push("[Compacted]"); + flushBuffers(entries, buffers); + entries.push(activityEntry("[Compacted]", "Compacted")); break; } } - flushBuffers(lines, buffers); + flushBuffers(entries, buffers); - return lines.length > 0 ? lines.join("\n") : "No activity to display."; + return entries; +} + +/** + * Convert normalized agent timeline items into a concise text summary. + */ +export function curateAgentActivity( + timeline: AgentTimelineItem[], + options?: { maxItems?: number }, +): string { + const entries = curateAgentActivityEntries(timeline, options); + return entries.length > 0 + ? entries.map((entry) => entry.text).join("\n") + : "No activity to display."; +} + +export function curateAgentActivityActions( + timeline: AgentTimelineItem[], + options?: { maxItems?: number }, +): Array<{ index: number; toolName: string; summary?: string }> { + return curateAgentActivityEntries(timeline, options).map((entry, index) => + Object.assign({ index: index + 1 }, entry.action), + ); } 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 6071b8896..587a61a46 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 @@ -671,6 +671,103 @@ describe("Codex app-server provider", () => { ]); }); + test("converts Codex collab agent notifications through the normal timeline path", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/started", { + threadId: "test-thread", + item: { + type: "collabAgentToolCall", + id: "call-sub-agent-normal-path", + tool: "spawnAgent", + status: "inProgress", + prompt: "Inspect the stream path.", + receiverThreadIds: [], + agentsStates: {}, + }, + }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "tool_call", + callId: "call-sub-agent-normal-path", + name: "Sub-agent", + status: "running", + error: null, + detail: { + type: "sub_agent", + subAgentType: "Sub-agent", + description: "Inspect the stream path.", + log: "", + actions: [], + }, + }, + }, + ]); + }); + + test("folds child-thread Codex activity into the parent sub-agent tool call", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "collabAgentToolCall", + id: "call-sub-agent-child-activity", + tool: "spawnAgent", + status: "completed", + prompt: "Report findings.", + receiverThreadIds: ["child-thread-1"], + agentsStates: { + "child-thread-1": { status: "pendingInit", message: null }, + }, + }, + }); + asInternals(session).handleNotification("item/agentMessage/delta", { + threadId: "child-thread-1", + itemId: "child-message-1", + delta: "Found the path.", + }); + asInternals(session).handleNotification("item/completed", { + threadId: "child-thread-1", + item: { + type: "agentMessage", + id: "child-message-1", + text: "Found the path.", + }, + }); + asInternals(session).handleNotification("turn/completed", { + threadId: "child-thread-1", + turn: { status: "completed" }, + }); + + const timelineEvents = events.filter((event) => event.type === "timeline"); + expect(timelineEvents).toHaveLength(4); + expect(timelineEvents.every((event) => event.item.type === "tool_call")).toBe(true); + const finalItem = timelineEvents.at(-1)?.item; + expect(finalItem).toMatchObject({ + type: "tool_call", + callId: "call-sub-agent-child-activity", + name: "Sub-agent", + status: "completed", + detail: { + type: "sub_agent", + subAgentType: "Sub-agent", + description: "Report findings.", + log: "Found the path.", + actions: [{ index: 1, toolName: "Assistant", summary: "Found the path." }], + }, + }); + }); + test("maps question responses from headers back to question ids and completes the tool call", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; 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 a4fc1352c..c44256bdd 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 @@ -39,6 +39,7 @@ import readline from "node:readline"; import { z } from "zod"; import { loadCodexPersistedTimeline } from "./codex-rollout-timeline.js"; import { renderPromptAttachmentAsText } from "../prompt-attachments.js"; +import { curateAgentActivity, curateAgentActivityActions } from "../activity-curator.js"; import { mapCodexRolloutToolCall, mapCodexToolCallFromThreadItem, @@ -1176,6 +1177,8 @@ function normalizeCodexThreadItemType(rawType: string | undefined): string | und return "mcpToolCall"; case "WebSearch": return "webSearch"; + case "CollabAgentToolCall": + return "collabAgentToolCall"; default: return rawType; } @@ -1547,6 +1550,7 @@ function threadItemToTimeline( case "fileChange": case "mcpToolCall": case "webSearch": + case "collabAgentToolCall": return mapCodexToolCallFromThreadItem(normalizedItem, { cwd }); default: return null; @@ -1608,12 +1612,14 @@ const ThreadStartedNotificationSchema = z const TurnStartedNotificationSchema = z .object({ + threadId: z.string().optional(), turn: z.object({ id: z.string() }).passthrough(), }) .passthrough(); const TurnCompletedNotificationSchema = z .object({ + threadId: z.string().optional(), turn: z .object({ status: z.string(), @@ -1656,6 +1662,7 @@ const ThreadTokenUsageUpdatedNotificationSchema = z const ItemTextDeltaNotificationSchema = z .object({ + threadId: z.string().optional(), itemId: z.string(), delta: z.string(), }) @@ -1663,6 +1670,7 @@ const ItemTextDeltaNotificationSchema = z const ItemLifecycleNotificationSchema = z .object({ + threadId: z.string().optional(), item: z .object({ id: z.string().optional(), @@ -1695,9 +1703,12 @@ const CodexEventTaskCompleteNotificationSchema = z const CodexEventItemLifecycleNotificationSchema = z .object({ + threadId: z.string().optional(), msg: z .object({ type: z.enum(["item_started", "item_completed"]), + threadId: z.string().optional(), + thread_id: z.string().optional(), item: z .object({ id: z.string().optional(), @@ -1827,21 +1838,28 @@ const CodexEventTurnDiffNotificationSchema = z type ParsedCodexNotification = | { kind: "thread_started"; threadId: string } - | { kind: "turn_started"; turnId: string } - | { kind: "turn_completed"; status: string; errorMessage: string | null } + | { kind: "turn_started"; turnId: string; threadId: string | null } + | { + kind: "turn_completed"; + status: string; + errorMessage: string | null; + threadId: string | null; + } | { kind: "plan_updated"; plan: Array<{ step: string | null; status: string | null }> } | { kind: "diff_updated"; diff: string } | { kind: "token_usage_updated"; tokenUsage: unknown } - | { kind: "agent_message_delta"; itemId: string; delta: string } - | { kind: "reasoning_delta"; itemId: string; delta: string } + | { kind: "agent_message_delta"; itemId: string; delta: string; threadId: string | null } + | { kind: "reasoning_delta"; itemId: string; delta: string; threadId: string | null } | { kind: "item_completed"; source: "item" | "codex_event"; + threadId: string | null; item: { id?: string; type?: string; [key: string]: unknown }; } | { kind: "item_started"; source: "item" | "codex_event"; + threadId: string | null; item: { id?: string; type?: string; [key: string]: unknown }; } | { @@ -1910,11 +1928,13 @@ const CodexNotificationSchema = z.union([ params, }), ), - z - .object({ method: z.literal("turn/started"), params: TurnStartedNotificationSchema }) - .transform( - ({ params }): ParsedCodexNotification => ({ kind: "turn_started", turnId: params.turn.id }), - ), + z.object({ method: z.literal("turn/started"), params: TurnStartedNotificationSchema }).transform( + ({ params }): ParsedCodexNotification => ({ + kind: "turn_started", + turnId: params.turn.id, + threadId: params.threadId ?? null, + }), + ), z.object({ method: z.literal("turn/started"), params: z.unknown() }).transform( ({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", @@ -1929,6 +1949,7 @@ const CodexNotificationSchema = z.union([ kind: "turn_completed", status: params.turn.status, errorMessage: params.turn.error?.message ?? null, + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("turn/completed"), params: z.unknown() }).transform( @@ -1996,6 +2017,7 @@ const CodexNotificationSchema = z.union([ kind: "agent_message_delta", itemId: params.itemId, delta: params.delta, + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("item/agentMessage/delta"), params: z.unknown() }).transform( @@ -2015,6 +2037,7 @@ const CodexNotificationSchema = z.union([ kind: "reasoning_delta", itemId: params.itemId, delta: params.delta, + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("item/reasoning/summaryTextDelta"), params: z.unknown() }).transform( @@ -2030,6 +2053,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_completed", source: "item", + threadId: params.threadId ?? null, item: params.item, }), ), @@ -2046,6 +2070,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_started", source: "item", + threadId: params.threadId ?? null, item: params.item, }), ), @@ -2065,6 +2090,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_started", source: "codex_event", + threadId: params.threadId ?? params.msg.threadId ?? params.msg.thread_id ?? null, item: params.msg.item, }), ), @@ -2084,6 +2110,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_completed", source: "codex_event", + threadId: params.threadId ?? params.msg.threadId ?? params.msg.thread_id ?? null, item: params.msg.item, }), ), @@ -2311,6 +2338,7 @@ const CodexNotificationSchema = z.union([ kind: "turn_completed", status: "interrupted", errorMessage: null, + threadId: null, }), ), z.object({ method: z.literal("codex/event/turn_aborted"), params: z.unknown() }).transform( @@ -2330,6 +2358,7 @@ const CodexNotificationSchema = z.union([ kind: "turn_completed", status: "completed", errorMessage: null, + threadId: null, }), ), z.object({ method: z.literal("codex/event/task_complete"), params: z.unknown() }).transform( @@ -2508,6 +2537,13 @@ function buildCodexAppServerInitializeParams(): { }; } +interface CodexSubAgentCallState { + callId: string; + toolCall: ToolCallTimelineItem; + childItemOrder: string[]; + childItems: Map; +} + class CodexAppServerAgentSession implements AgentSession { readonly provider = CODEX_PROVIDER; readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; @@ -2549,6 +2585,8 @@ class CodexAppServerAgentSession implements AgentSession { private emittedExecCommandCompletedCallIds = new Set(); private emittedItemStartedIds = new Set(); private emittedItemCompletedIds = new Set(); + private subAgentCallsByCallId = new Map(); + private subAgentCallIdByChildThreadId = new Map(); private warnedUnknownNotificationMethods = new Set(); private warnedInvalidNotificationPayloads = new Set(); private warnedIncompleteEditToolCallIds = new Set(); @@ -3562,6 +3600,141 @@ class CodexAppServerAgentSession implements AgentSession { } } + private getSubAgentCallIdForThread(threadId: string | null | undefined): string | null { + if (!threadId || threadId === this.currentThreadId) { + return null; + } + return this.subAgentCallIdByChildThreadId.get(threadId) ?? null; + } + + private registerSubAgentToolCall( + timelineItem: ToolCallTimelineItem, + rawItem: { [key: string]: unknown }, + ): void { + if (timelineItem.detail.type !== "sub_agent") { + return; + } + + const existing = this.subAgentCallsByCallId.get(timelineItem.callId); + const state: CodexSubAgentCallState = + existing ?? + ({ + callId: timelineItem.callId, + toolCall: timelineItem, + childItemOrder: [], + childItems: new Map(), + } satisfies CodexSubAgentCallState); + + let actions = timelineItem.detail.actions; + if (actions.length === 0 && state.toolCall.detail.type === "sub_agent") { + actions = state.toolCall.detail.actions; + } + + state.toolCall = { + ...timelineItem, + detail: { + ...timelineItem.detail, + log: + timelineItem.detail.log || + (state.toolCall.detail.type === "sub_agent" ? state.toolCall.detail.log : ""), + actions, + }, + }; + this.subAgentCallsByCallId.set(timelineItem.callId, state); + + const receiverThreadIds = Array.isArray(rawItem.receiverThreadIds) + ? rawItem.receiverThreadIds.filter((value): value is string => typeof value === "string") + : []; + for (const receiverThreadId of receiverThreadIds) { + this.subAgentCallIdByChildThreadId.set(receiverThreadId, timelineItem.callId); + } + } + + private upsertSubAgentChildItem(callId: string, itemId: string, item: AgentTimelineItem): void { + const state = this.subAgentCallsByCallId.get(callId); + if (!state) { + return; + } + if (!state.childItems.has(itemId)) { + state.childItemOrder.push(itemId); + } + state.childItems.set(itemId, item); + } + + private getSubAgentChildTimeline(state: CodexSubAgentCallState): AgentTimelineItem[] { + return state.childItemOrder + .map((itemId) => state.childItems.get(itemId)) + .filter((item): item is AgentTimelineItem => Boolean(item)); + } + + private emitSubAgentActivityUpdate( + callId: string, + status?: ToolCallTimelineItem["status"], + ): void { + const state = this.subAgentCallsByCallId.get(callId); + if (!state || state.toolCall.detail.type !== "sub_agent") { + return; + } + const childTimeline = this.getSubAgentChildTimeline(state); + const log = childTimeline.length > 0 ? curateAgentActivity(childTimeline) : ""; + const actions = childTimeline.length > 0 ? curateAgentActivityActions(childTimeline) : []; + const resolvedStatus = status ?? state.toolCall.status; + const baseToolCall = { + ...state.toolCall, + detail: { + ...state.toolCall.detail, + log, + actions, + }, + }; + const nextToolCall: ToolCallTimelineItem = + resolvedStatus === "failed" + ? { + ...baseToolCall, + status: "failed", + error: state.toolCall.error ?? { message: "Sub-agent failed" }, + } + : { + ...baseToolCall, + status: resolvedStatus, + error: null, + }; + state.toolCall = nextToolCall; + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: nextToolCall }); + } + + private handleSubAgentChildItemCompleted( + callId: string, + itemId: string | undefined, + timelineItem: AgentTimelineItem, + ): void { + this.applyBufferedDeltaTextToTimelineItem(timelineItem, itemId); + if (itemId) { + this.upsertSubAgentChildItem(callId, itemId, timelineItem); + this.pendingAgentMessages.delete(itemId); + this.pendingReasoning.delete(itemId); + this.pendingCommandOutputDeltas.delete(itemId); + this.pendingFileChangeOutputDeltas.delete(itemId); + } + this.emitSubAgentActivityUpdate( + callId, + timelineItem.type === "tool_call" && timelineItem.status === "failed" ? "failed" : "running", + ); + } + + private shouldSkipCompletedThreadItem( + timelineItem: AgentTimelineItem, + normalizedItemType: string | undefined, + itemId: string | undefined, + ): boolean { + // For commandExecution items, codex/event/exec_command_* is authoritative. + if (timelineItem.type === "tool_call" && normalizedItemType === "commandExecution") { + const callId = timelineItem.callId || itemId; + return Boolean(callId && this.emittedExecCommandCompletedCallIds.has(callId)); + } + return Boolean(itemId && this.emittedItemCompletedIds.has(itemId)); + } + private handleCodexDeltaNotification( parsed: Extract< ParsedCodexNotification, @@ -3576,7 +3749,17 @@ class CodexAppServerAgentSession implements AgentSession { ): void { if (parsed.kind === "agent_message_delta") { const prev = this.pendingAgentMessages.get(parsed.itemId) ?? ""; - this.pendingAgentMessages.set(parsed.itemId, prev + parsed.delta); + const text = prev + parsed.delta; + this.pendingAgentMessages.set(parsed.itemId, text); + const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (subAgentCallId) { + this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, { + type: "assistant_message", + text, + }); + this.emitSubAgentActivityUpdate(subAgentCallId, "running"); + return; + } const isFirstDeltaForItem = prev.length === 0; this.emitEvent({ type: "timeline", @@ -3598,6 +3781,15 @@ class CodexAppServerAgentSession implements AgentSession { const prev = this.pendingReasoning.get(parsed.itemId) ?? []; prev.push(parsed.delta); this.pendingReasoning.set(parsed.itemId, prev); + const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (subAgentCallId) { + this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, { + type: "reasoning", + text: prev.join(""), + }); + this.emitSubAgentActivityUpdate(subAgentCallId, "running"); + return; + } this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, @@ -3628,6 +3820,11 @@ class CodexAppServerAgentSession implements AgentSession { private handleTurnStartedNotification( parsed: Extract, ): void { + const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (subAgentCallId) { + this.emitSubAgentActivityUpdate(subAgentCallId, "running"); + return; + } this.currentTurnId = parsed.turnId; this.resetTurnTrackingState(); this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER }); @@ -3636,6 +3833,17 @@ class CodexAppServerAgentSession implements AgentSession { private handleTurnCompletedNotification( parsed: Extract, ): void { + const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (subAgentCallId) { + let status: ToolCallTimelineItem["status"] = "completed"; + if (parsed.status === "failed") { + status = "failed"; + } else if (parsed.status === "interrupted") { + status = "canceled"; + } + this.emitSubAgentActivityUpdate(subAgentCallId, status); + return; + } if (parsed.status === "failed") { this.emitEvent({ type: "turn_failed", @@ -3833,18 +4041,16 @@ class CodexAppServerAgentSession implements AgentSession { if (!timelineItem) { return; } + const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (childSubAgentCallId) { + this.handleSubAgentChildItemCompleted(childSubAgentCallId, parsed.item.id, timelineItem); + return; + } const normalizedItemType = normalizeCodexThreadItemType( typeof parsed.item.type === "string" ? parsed.item.type : undefined, ); const itemId = parsed.item.id; - // For commandExecution items, codex/event/exec_command_* is authoritative. - if (timelineItem.type === "tool_call" && normalizedItemType === "commandExecution") { - const callId = timelineItem.callId || itemId; - if (callId && this.emittedExecCommandCompletedCallIds.has(callId)) { - return; - } - } - if (itemId && this.emittedItemCompletedIds.has(itemId)) { + if (this.shouldSkipCompletedThreadItem(timelineItem, normalizedItemType, itemId)) { return; } if (this.consumeStreamedTextCompletion(timelineItem, itemId)) { @@ -3859,6 +4065,7 @@ class CodexAppServerAgentSession implements AgentSession { } this.applyBufferedDeltaTextToTimelineItem(timelineItem, itemId); if (timelineItem.type === "tool_call") { + this.registerSubAgentToolCall(timelineItem, parsed.item); if (timelineItem.detail.type === "plan") { this.rememberPlanResult(timelineItem); // Codex can surface plans both as turn/plan updates and as completed @@ -3957,6 +4164,14 @@ class CodexAppServerAgentSession implements AgentSession { if (!timelineItem || timelineItem.type !== "tool_call") { return; } + const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (childSubAgentCallId) { + if (parsed.item.id) { + this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem); + } + this.emitSubAgentActivityUpdate(childSubAgentCallId, "running"); + return; + } const normalizedItemType = normalizeCodexThreadItemType( typeof parsed.item.type === "string" ? parsed.item.type : undefined, ); @@ -3971,6 +4186,7 @@ class CodexAppServerAgentSession implements AgentSession { return; } this.warnOnIncompleteEditToolCall(timelineItem, "item_started", parsed.item); + this.registerSubAgentToolCall(timelineItem, parsed.item); this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); if (itemId) { this.emittedItemStartedIds.add(itemId); diff --git a/packages/server/src/server/agent/providers/codex/tool-call-mapper.test.ts b/packages/server/src/server/agent/providers/codex/tool-call-mapper.test.ts index 065934f9b..6871ec04c 100644 --- a/packages/server/src/server/agent/providers/codex/tool-call-mapper.test.ts +++ b/packages/server/src/server/agent/providers/codex/tool-call-mapper.test.ts @@ -150,6 +150,35 @@ describe("codex tool-call mapper", () => { }); }); + it("maps collabAgentToolCall into canonical sub-agent detail", () => { + const item = mapCodexToolCallFromThreadItem({ + type: "collabAgentToolCall", + id: "call-sub-agent-1", + tool: "spawnAgent", + status: "completed", + prompt: "Inspect the Codex stream path.", + receiverThreadIds: ["child-thread-1"], + agentsStates: { + "child-thread-1": { status: "pendingInit", message: null }, + }, + }); + + expect(item).toEqual({ + type: "tool_call", + callId: "call-sub-agent-1", + name: "Sub-agent", + status: "running", + error: null, + detail: { + type: "sub_agent", + subAgentType: "Sub-agent", + description: "Inspect the Codex stream path.", + log: "", + actions: [], + }, + }); + }); + it("maps mcp read_file completion with detail", () => { const item = mapCodexToolCallFromThreadItem( { diff --git a/packages/server/src/server/agent/providers/codex/tool-call-mapper.ts b/packages/server/src/server/agent/providers/codex/tool-call-mapper.ts index 760d757ff..2ac389ba9 100644 --- a/packages/server/src/server/agent/providers/codex/tool-call-mapper.ts +++ b/packages/server/src/server/agent/providers/codex/tool-call-mapper.ts @@ -170,11 +170,32 @@ const CodexWebSearchItemSchema = z }) .passthrough(); +const CodexCollabAgentToolCallItemSchema = z + .object({ + type: z.literal("collabAgentToolCall"), + id: z.string().min(1), + status: z.string().optional(), + error: z.unknown().optional(), + prompt: z.string().optional(), + tool: z.string().optional(), + receiverThreadIds: z.array(z.string()).optional(), + agentsStates: z.record(z.string(), z.unknown()).optional(), + }) + .passthrough(); + +const CodexToolThreadItemSchema = z.discriminatedUnion("type", [ + CodexCommandExecutionItemSchema, + CodexFileChangeItemSchema, + CodexMcpToolCallItemSchema, + CodexWebSearchItemSchema, +]); + const CodexThreadItemSchema = z.discriminatedUnion("type", [ CodexCommandExecutionItemSchema, CodexFileChangeItemSchema, CodexMcpToolCallItemSchema, CodexWebSearchItemSchema, + CodexCollabAgentToolCallItemSchema, ]); function maybeUnwrapShellWrapperCommand(command: string): string { @@ -481,6 +502,38 @@ function hasRenderableEditDetail(detail: ToolCallTimelineItem["detail"]): boolea ); } +function readStatus(value: unknown): string | undefined { + if (!isRecord(value)) { + return undefined; + } + return typeof value.status === "string" ? value.status : undefined; +} + +function resolveCollabAgentStatus( + item: z.infer, +): ToolCallTimelineItem["status"] { + if (item.error !== undefined && item.error !== null) { + return "failed"; + } + + const childStatuses = Object.values(item.agentsStates ?? {}) + .map(readStatus) + .filter((status): status is string => typeof status === "string" && status.trim().length > 0) + .map((status) => normalizeToolCallStatus(status, null, null)); + + if (childStatuses.some((status) => status === "failed")) { + return "failed"; + } + if (childStatuses.some((status) => status === "canceled")) { + return "canceled"; + } + if (childStatuses.length > 0) { + return childStatuses.every((status) => status === "completed") ? "completed" : "running"; + } + + return normalizeToolCallStatus(item.status, item.error ?? null, null); +} + function buildMcpToolName(server: string | undefined, tool: string): string { const trimmedTool = tool.trim(); if (!trimmedTool) { @@ -789,8 +842,41 @@ function mapWebSearchItem( }; } +function mapCollabAgentToolCallItem( + item: z.infer, +): ToolCallTimelineItem { + const status = resolveCollabAgentStatus(item); + const detail: ToolCallTimelineItem["detail"] = { + type: "sub_agent", + subAgentType: "Sub-agent", + ...(item.prompt ? { description: item.prompt } : {}), + log: "", + actions: [], + }; + + if (status === "failed") { + return { + type: "tool_call", + callId: item.id, + name: "Sub-agent", + status, + error: item.error ?? { message: "Sub-agent failed" }, + detail, + }; + } + + return { + type: "tool_call", + callId: item.id, + name: "Sub-agent", + status, + error: null, + detail, + }; +} + function mapThreadItemToNormalizedEnvelope( - item: z.infer, + item: z.infer, options?: CodexMapperOptions, ): CodexNormalizedToolCallEnvelope | null { switch (item.type) { @@ -821,6 +907,9 @@ export function mapCodexToolCallFromThreadItem( if (!parsed.success) { return null; } + if (parsed.data.type === "collabAgentToolCall") { + return mapCollabAgentToolCallItem(parsed.data); + } const envelope = mapThreadItemToNormalizedEnvelope(parsed.data, options); if (!envelope) { return null;