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 9ff1a5b282..42e7513d3d 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 @@ -58,6 +58,7 @@ import { mapCodexToolCallEnvelope, mapCodexToolCallFromThreadItem, normalizeCommandExecutionCommand, + normalizeCollabAgentChildStatus, splitCodexMcpToolResultImages, } from "./codex/tool-call-mapper.js"; import { @@ -1864,7 +1865,12 @@ interface CodexCollabAgentUpdate { reopen: boolean; } -function readCodexCollabAgentUpdates(item: unknown): CodexCollabAgentUpdate[] { +interface CodexCollabAgentControl { + childThreadIds: string[]; + updates: CodexCollabAgentUpdate[]; +} + +function readCodexCollabAgentControl(item: unknown): CodexCollabAgentControl | null { const record = toObjectRecord(item); if ( !record || @@ -1872,34 +1878,33 @@ function readCodexCollabAgentUpdates(item: unknown): CodexCollabAgentUpdate[] { "collabAgentToolCall" || record.tool === "spawnAgent" ) { - return []; + return null; } - return Object.entries(toObjectRecord(record.agentsStates) ?? {}).flatMap(([threadId, value]) => { - const state = toObjectRecord(value); - if (typeof state?.status !== "string") return []; - const mapped = mapCodexToolCallFromThreadItem({ - type: "collabAgentToolCall", - id: record.id, - tool: record.tool, - status: "completed", - agentsStates: { [threadId]: state }, - }); - return mapped - ? [ - { - threadId, - status: - record.tool === "closeAgent" && - record.status === "completed" && - mapped.status === "running" - ? "canceled" - : mapped.status, - shutdown: state.status === "shutdown", - reopen: record.tool === "sendInput" && record.status === "completed", - }, - ] - : []; - }); + const updates = Object.entries(toObjectRecord(record.agentsStates) ?? {}).flatMap( + ([threadId, value]): CodexCollabAgentUpdate[] => { + const state = toObjectRecord(value); + if (typeof state?.status !== "string") return []; + const status = normalizeCollabAgentChildStatus(state.status); + return [ + { + threadId, + status: + record.tool === "closeAgent" && record.status === "completed" && status === "running" + ? "canceled" + : status, + shutdown: state.status === "shutdown", + reopen: record.tool === "sendInput" && record.status === "completed", + }, + ]; + }, + ); + const receivers = Array.isArray(record.receiverThreadIds) + ? record.receiverThreadIds.filter((id): id is string => typeof id === "string") + : []; + return { + childThreadIds: [...new Set([...receivers, ...updates.map((update) => update.threadId)])], + updates, + }; } function resolveCodexCollabAgentUpdate( @@ -2123,8 +2128,9 @@ function updateHistoricalCollabAgentCalls( timelineIndexByThreadId: Map, item: unknown, ): boolean { - let updated = false; - for (const update of readCodexCollabAgentUpdates(item)) { + const control = readCodexCollabAgentControl(item); + if (!control) return false; + for (const update of control.updates) { const existingIndex = timelineIndexByThreadId.get(update.threadId); const existing = existingIndex === undefined ? undefined : timeline[existingIndex]; if (existingIndex === undefined || existing?.item.type !== "tool_call") continue; @@ -2134,9 +2140,11 @@ function updateHistoricalCollabAgentCalls( ? { ...existing.item, status, error: { message: "Sub-agent failed" } } : { ...existing.item, status, error: null }; timeline[existingIndex] = { ...existing, item: toolCall }; - updated = true; } - return updated; + return ( + control.childThreadIds.length > 0 && + control.childThreadIds.every((id) => timelineIndexByThreadId.has(id)) + ); } async function loadCodexThreadHistoryTimeline(params: { @@ -5819,19 +5827,23 @@ export class CodexAppServerAgentSession implements AgentSession { } private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean { - let updatedExistingSubagent = false; - for (const update of readCodexCollabAgentUpdates(rawItem)) { - const callId = this.subAgentCallIdByChildThreadId.get(update.threadId); - if (!callId) continue; - const state = this.subAgentCallsByCallId.get(callId); - if (!state) continue; - const status = resolveCodexCollabAgentUpdate(state.toolCall.status, update); - if (status !== state.toolCall.status) { - this.emitSubAgentActivityUpdate(callId, status, { reopen: update.reopen }); + const control = readCodexCollabAgentControl(rawItem); + if (control) { + for (const update of control.updates) { + const callId = this.subAgentCallIdByChildThreadId.get(update.threadId); + if (!callId) continue; + const state = this.subAgentCallsByCallId.get(callId); + if (!state) continue; + const status = resolveCodexCollabAgentUpdate(state.toolCall.status, update); + if (status !== state.toolCall.status) { + this.emitSubAgentActivityUpdate(callId, status, { reopen: update.reopen }); + } } - updatedExistingSubagent = true; + return ( + control.childThreadIds.length > 0 && + control.childThreadIds.every((id) => this.subAgentCallIdByChildThreadId.has(id)) + ); } - if (updatedExistingSubagent) return true; const activity = readCodexSubAgentActivity(rawItem); if (!activity) { return false; diff --git a/packages/server/src/server/agent/providers/codex/collab-replay.test.ts b/packages/server/src/server/agent/providers/codex/collab-replay.test.ts index 4b9fd1a04d..da86b1eb79 100644 --- a/packages/server/src/server/agent/providers/codex/collab-replay.test.ts +++ b/packages/server/src/server/agent/providers/codex/collab-replay.test.ts @@ -116,12 +116,27 @@ describe("Codex collaboration history", () => { async () => appServer.child, ); const statuses = new Map(); + const timelineCallIds = new Set(); session.subscribe((event) => { if (event.type === "provider_subagent" && event.event.type === "upsert") { statuses.set(event.event.id, event.event.status); } + if (event.type === "timeline" && event.item.type === "tool_call") { + timelineCallIds.add(event.item.callId); + } }); function complete(tool: string, states: Record, id = tool) { + if (tool !== "spawnAgent") { + appServer.child.stdout.write( + `${JSON.stringify({ + method: "item/started", + params: { + threadId: "thread-1", + item: { ...collab(tool, states, id), status: "inProgress", agentsStates: {} }, + }, + })}\n`, + ); + } appServer.child.stdout.write( `${JSON.stringify({ method: "item/completed", @@ -147,6 +162,7 @@ describe("Codex collaboration history", () => { expect(statuses.get("first")).toBe("completed"); complete("sendInput", { first: "running" }); await vi.waitFor(() => expect(statuses.get("first")).toBe("running")); + expect([...timelineCallIds]).toEqual(["spawn-first", "spawn-second"]); } finally { await session.close(); appServer.assertNoErrors(); 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 8147dd9de7..9d02198cfd 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 @@ -545,7 +545,7 @@ function readStatus(value: unknown): string | undefined { return typeof value.status === "string" ? value.status : undefined; } -function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] { +export function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] { const normalized = status.trim().toLowerCase(); switch (normalized) { case "error":