Stream Codex sub-agents as tool calls (#672)

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",
+      },
+    ]);
+  });
+});

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),
+  );
 }

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[] = [];

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<string, AgentTimelineItem>;
+}
+
 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<string>();
   private emittedItemStartedIds = new Set<string>();
   private emittedItemCompletedIds = new Set<string>();
+  private subAgentCallsByCallId = new Map<string, CodexSubAgentCallState>();
+  private subAgentCallIdByChildThreadId = new Map<string, string>();
   private warnedUnknownNotificationMethods = new Set<string>();
   private warnedInvalidNotificationPayloads = new Set<string>();
   private warnedIncompleteEditToolCallIds = new Set<string>();
@@ -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<string, AgentTimelineItem>(),
+      } 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<ParsedCodexNotification, { kind: "turn_started" }>,
   ): 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<ParsedCodexNotification, { kind: "turn_completed" }>,
   ): 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);

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(
       {

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<typeof CodexCollabAgentToolCallItemSchema>,
+): 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<typeof CodexCollabAgentToolCallItemSchema>,
+): 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<typeof CodexThreadItemSchema>,
+  item: z.infer<typeof CodexToolThreadItemSchema>,
   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;