From a689ccb0af80819381fbaa74da87db78455e7a69 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 10 Jul 2026 09:41:32 +0200 Subject: [PATCH] Fix Codex status and streaming during sub-agent work (#1967) * fix(codex): keep parent turns active during sub-agent work Codex multi-agent activity is thread-scoped. Treating child events as root activity could end the foreground turn early and splice child output into parent messages. Route legacy and current activity shapes through stable child identities and preserve message boundaries across live and resumed turns. * fix(codex): preserve sub-agent details across stream chunks Keep empty activity paths explicit and normalize the synthetic assistant boundary for every chunk in the active message. * fix(codex): route legacy child events by thread * fix(codex): stabilize sub-agent activity lifecycles * fix(codex): retain child tool activity and terminal state --- docs/testing.md | 2 + .../src/server/agent/activity-curator.test.ts | 40 + .../src/server/agent/activity-curator.ts | 3 + .../agent/agent-stream-coalescer.test.ts | 27 +- .../server/agent/agent-stream-coalescer.ts | 12 +- .../codex-app-server-agent.real.e2e.test.ts | 113 ++- .../providers/codex-app-server-agent.test.ts | 765 ++++++++++++++++++ .../agent/providers/codex-app-server-agent.ts | 632 +++++++++++++-- .../codex/test-utils/fake-app-server.ts | 154 ++++ .../providers/codex/tool-call-mapper.test.ts | 43 + .../agent/providers/codex/tool-call-mapper.ts | 33 + 11 files changed, 1740 insertions(+), 84 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index 6c7e24109..7bb773482 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -115,6 +115,8 @@ App-level Playwright browser E2E lives in `packages/app/e2e/*.spec.ts` and runs Live provider smoke tests belong in `*.real.e2e.test.ts`, not `*.test.ts`, even when guarded by environment variables. Default unit suites must use deterministic provider adapters/fakes so missing credits, auth outages, and upstream model drift do not block normal CI. +Codex MultiAgentV2 real tests use local Codex authentication rather than the OpenRouter-compatible test provider. OpenRouter does not accept Codex collaboration-history items on the parent follow-up request, so it cannot verify a complete native sub-agent turn. + ### Test setup - Server: `packages/server/src/test-utils/vitest-setup.ts` loads `.env.test`, sets `PASEO_SUPERVISED=0`, and disables Git/SSH prompts. Add new global env shims here, not in individual tests. diff --git a/packages/server/src/server/agent/activity-curator.test.ts b/packages/server/src/server/agent/activity-curator.test.ts index d7500ac73..75ae75fe7 100644 --- a/packages/server/src/server/agent/activity-curator.test.ts +++ b/packages/server/src/server/agent/activity-curator.test.ts @@ -185,6 +185,46 @@ second line'`, ]); }); + it("keeps nested sub-agent logs beside the sub-agent that produced them", () => { + const timeline: AgentTimelineItem[] = [ + { type: "assistant_message", text: "Before first child." }, + toolCallItem({ + callId: "child-1", + name: "Sub-agent", + detail: { + type: "sub_agent", + subAgentType: "Child one", + description: "First investigation", + log: "[Assistant] First child result.", + }, + }), + { type: "assistant_message", text: "Between children." }, + toolCallItem({ + callId: "child-2", + name: "Sub-agent", + detail: { + type: "sub_agent", + subAgentType: "Child two", + description: "Second investigation", + log: "[Assistant] Second child result.", + }, + }), + { type: "assistant_message", text: "After second child." }, + ]; + + const result = curateAgentActivity(timeline, { labelAssistantMessages: true }); + + expect(result.split("\n")).toEqual([ + "[Assistant] Before first child.", + "[Child one] First investigation", + "[Assistant] First child result.", + "[Assistant] Between children.", + "[Child two] Second investigation", + "[Assistant] Second child result.", + "[Assistant] After second child.", + ]); + }); + it("renders todo/error/compaction entries", () => { const timeline: AgentTimelineItem[] = [ { diff --git a/packages/server/src/server/agent/activity-curator.ts b/packages/server/src/server/agent/activity-curator.ts index cb248cb62..359fb4280 100644 --- a/packages/server/src/server/agent/activity-curator.ts +++ b/packages/server/src/server/agent/activity-curator.ts @@ -165,6 +165,9 @@ function curateProjectedActivityEntries( case "tool_call": { flushBuffers(entries, buffers, options); entries.push(formatToolCallEntry(item, options)); + if (item.detail.type === "sub_agent" && item.detail.log.trim()) { + entries.push(activityEntry(item.detail.log.trim())); + } break; } case "todo": diff --git a/packages/server/src/server/agent/agent-stream-coalescer.test.ts b/packages/server/src/server/agent/agent-stream-coalescer.test.ts index 0983712e2..64b348d8e 100644 --- a/packages/server/src/server/agent/agent-stream-coalescer.test.ts +++ b/packages/server/src/server/agent/agent-stream-coalescer.test.ts @@ -45,9 +45,17 @@ function assistant( options?: { provider?: AgentProvider; turnId?: string; + messageId?: string; }, ): Extract { - return timeline({ type: "assistant_message", text }, options); + return timeline( + { + type: "assistant_message", + text, + ...(options?.messageId !== undefined ? { messageId: options.messageId } : {}), + }, + options, + ); } function reasoning( @@ -284,6 +292,23 @@ describe("AgentStreamCoalescer", () => { ]); }); + test("does not splice concurrent assistant message ids together", async () => { + const { coalescer, flushes } = createHarness(); + + coalescer.handle("agent-1", assistant("The", { turnId: "turn-1", messageId: "message-a" })); + coalescer.handle("agent-1", assistant("0po", { turnId: "turn-1", messageId: "message-b" })); + coalescer.handle("agent-1", assistant(" exact", { turnId: "turn-1", messageId: "message-a" })); + coalescer.handle("agent-1", assistant("7/fr", { turnId: "turn-1", messageId: "message-b" })); + + await vi.advanceTimersByTimeAsync(60); + expect(flushes.map((flush) => flush.item)).toEqual([ + { type: "assistant_message", messageId: "message-a", text: "The" }, + { type: "assistant_message", messageId: "message-b", text: "0po" }, + { type: "assistant_message", messageId: "message-a", text: " exact" }, + { type: "assistant_message", messageId: "message-b", text: "7/fr" }, + ]); + }); + test("drops empty text chunks", async () => { const { coalescer, flushes } = createHarness(); diff --git a/packages/server/src/server/agent/agent-stream-coalescer.ts b/packages/server/src/server/agent/agent-stream-coalescer.ts index 3f179d58c..46008fcb5 100644 --- a/packages/server/src/server/agent/agent-stream-coalescer.ts +++ b/packages/server/src/server/agent/agent-stream-coalescer.ts @@ -73,6 +73,16 @@ function isTerminalToolCall(item: CoalescableTimelineItem): boolean { ); } +function isSameTextStream(previous: PendingTextEntry, next: PendingTextEntry): boolean { + if (previous.item.type !== next.item.type) { + return false; + } + if (previous.item.type === "assistant_message" && next.item.type === "assistant_message") { + return previous.item.messageId === next.item.messageId; + } + return true; +} + export class AgentStreamCoalescer { private readonly buffers = new Map(); private readonly onFlush: (payload: AgentStreamCoalescerFlush) => void; @@ -241,7 +251,7 @@ export class AgentStreamCoalescer { previous && previous.kind === "text" && entry.kind === "text" && - previous.item.type === entry.item.type && + isSameTextStream(previous, entry) && previous.provider === entry.provider && previous.turnId === entry.turnId ) { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.real.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.real.e2e.test.ts index 798cfffcf..ab01f5aff 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.real.e2e.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.real.e2e.test.ts @@ -1,47 +1,120 @@ import { mkdtempSync, rmSync } from "node:fs"; import os from "node:os"; import path from "node:path"; -import { beforeAll, beforeEach, describe, expect, test } from "vitest"; +import { beforeAll, describe, expect, test } from "vitest"; +import type { AgentStreamEvent } from "../agent-sdk-types.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { canRunRealProvider, createRealProviderClient, getRealProviderConfig, } from "../../daemon-e2e/real-provider-test-config.js"; +import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; describe("Codex app-server provider (real)", () => { - let canRun = false; + let canRunOpenRouter = false; beforeAll(async () => { - canRun = await canRunRealProvider("codex"); + canRunOpenRouter = await canRunRealProvider("codex"); }); - beforeEach((context) => { - if (!canRun) { + test("lists models and runs a simple prompt", async (context) => { + if (!canRunOpenRouter) { context.skip(); } - }); - - test("lists models and runs a simple prompt", async () => { const client = createRealProviderClient("codex", createTestLogger()); const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-app-server-e2e-")); - const { models } = await client.fetchCatalog({ scope: "workspace", cwd, force: false }); - expect(models.length).toBeGreaterThan(0); - - const session = await client.createSession({ - ...getRealProviderConfig("codex"), - cwd, - modeId: "auto", - }); try { - expect(session.features?.some((feature) => feature.id === "plan_mode")).toBe(true); + const { models } = await client.fetchCatalog({ scope: "workspace", cwd, force: false }); + expect(models.length).toBeGreaterThan(0); + const session = await client.createSession({ + ...getRealProviderConfig("codex"), + cwd, + modeId: "auto", + }); + try { + expect(session.features?.some((feature) => feature.id === "plan_mode")).toBe(true); - const result = await session.run("Say hello in one sentence."); - expect(result.finalText.length).toBeGreaterThan(0); + const result = await session.run("Say hello in one sentence."); + expect(result.finalText.length).toBeGreaterThan(0); + } finally { + await session.close(); + } } finally { - await session.close(); rmSync(cwd, { recursive: true, force: true }); } }, 30_000); + + test("keeps a real MultiAgentV2 child inside its parent turn", async () => { + const client = new CodexAppServerAgentClient(createTestLogger()); + const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-multi-agent-v2-e2e-")); + try { + const { models } = await client.fetchCatalog({ scope: "workspace", cwd, force: false }); + const model = models.find((candidate) => candidate.isDefault) ?? models[0]; + if (!model) { + throw new Error("Native Codex app-server returned no models"); + } + const session = await client.createSession({ + provider: "codex", + modeId: "full-access", + model: model.id, + cwd, + thinkingOptionId: "medium", + extra: { codex: { features: { multi_agent_v2: true } } }, + }); + const events: AgentStreamEvent[] = []; + const unsubscribe = session.subscribe((event) => events.push(event)); + + try { + const result = await session.run(` +Use collaboration.spawn_agent exactly once with task_name "sentinel_child", fork_turns "none", +and this task: "Reply with exactly CHILD_SENTINEL and do nothing else." +Wait for that child to finish with collaboration.wait_agent. Do not emit any assistant text before +the child finishes. After it finishes, reply with exactly ROOT_SENTINEL. Never repeat +CHILD_SENTINEL in your own response. +`); + + const completedChildIndex = events.findIndex( + (event) => + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.status === "completed" && + event.item.detail.type === "sub_agent" && + event.item.detail.log.includes("CHILD_SENTINEL"), + ); + const rootMessageIndexes = events.flatMap((event, index) => + event.type === "timeline" && event.item.type === "assistant_message" ? [index] : [], + ); + const terminalIndexes = events.flatMap((event, index) => + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ? [index] + : [], + ); + + expect(result.finalText.trim()).toBe("ROOT_SENTINEL"); + expect( + result.timeline.findLast( + (item) => item.type === "tool_call" && item.detail.type === "sub_agent", + ), + ).toMatchObject({ status: "completed" }); + const topLevelAssistantText = result.timeline + .filter((item) => item.type === "assistant_message") + .map((item) => item.text) + .join(""); + expect(topLevelAssistantText).not.toContain("CHILD_SENTINEL"); + expect(completedChildIndex).toBeGreaterThanOrEqual(0); + expect(rootMessageIndexes[0]).toBeGreaterThan(completedChildIndex); + expect(terminalIndexes).toHaveLength(1); + expect(terminalIndexes[0]).toBeGreaterThan(rootMessageIndexes.at(-1) ?? -1); + } finally { + unsubscribe(); + await session.close(); + } + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, 300_000); }); 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 835d979a2..56ec1fc3b 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 @@ -1527,6 +1527,599 @@ describe("Codex app-server provider", () => { }); }); + test("keeps the parent running when a MultiAgentV2 sub-agent finishes", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation, then report the result."); + await appServer.waitForTurnStart(); + + appServer.startsSubAgent({ + callId: "spawn-child-1", + threadId: "child-thread-1", + agentPath: "/root/child", + }); + appServer.says({ + threadId: "child-thread-1", + itemId: "child-message-1", + text: "Child findings.", + }); + appServer.completeTurn({ threadId: "child-thread-1" }); + appServer.says({ + threadId: "thread-1", + itemId: "parent-message-1", + text: "Parent report.", + chunks: ["Parent ", "report."], + }); + appServer.completeTurn(); + + const result = await resultPromise; + expect(result.finalText).toBe("Parent report."); + const assistantMessages = result.timeline.filter((item) => item.type === "assistant_message"); + expect(assistantMessages.map((item) => item.messageId)).toEqual([ + "parent-message-1", + "parent-message-1", + ]); + expect(assistantMessages.map((item) => item.text).join("")).toBe("Parent report."); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("returns only the latest assistant item without its visual boundary", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Report twice, then finish."); + await appServer.waitForTurnStart(); + + appServer.says({ + threadId: "thread-1", + itemId: "first-parent-message", + text: "First report.", + }); + appServer.says({ + threadId: "thread-1", + itemId: "second-parent-message", + text: "Second report.", + chunks: ["", "Second report."], + }); + appServer.completeTurn(); + + const result = await resultPromise; + expect(result.finalText).toBe("Second report."); + expect(result.finalText).not.toContain("---"); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("returns only the latest id-less assistant item", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Report twice, then finish."); + await appServer.waitForTurnStart(); + + appServer.says({ threadId: "thread-1", text: "First report." }); + appServer.says({ threadId: "thread-1", text: "Second report." }); + appServer.completeTurn(); + + const result = await resultPromise; + expect(result.finalText).toBe("Second report."); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("replays MultiAgentV2 child activity that arrives before its parent mapping", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation, then report the result."); + await appServer.waitForTurnStart(); + + appServer.startsTurn({ threadId: "child-thread-early" }); + appServer.says({ + threadId: "child-thread-early", + itemId: "child-message-early", + text: "Early child findings.", + }); + appServer.completeTurn({ threadId: "child-thread-early" }); + appServer.startsSubAgent({ + callId: "spawn-child-early", + threadId: "child-thread-early", + agentPath: "/root/early-child", + }); + appServer.says({ + threadId: "thread-1", + itemId: "parent-message-after-early-child", + text: "Parent report after replay.", + }); + appServer.completeTurn(); + + const result = await resultPromise; + expect(result.finalText).toBe("Parent report after replay."); + expect(result.timeline.filter((item) => item.type === "assistant_message")).toEqual([ + { + type: "assistant_message", + messageId: "parent-message-after-early-child", + text: "Parent report after replay.", + }, + ]); + expect(result.timeline.findLast((item) => item.type === "tool_call")).toMatchObject({ + type: "tool_call", + callId: "spawn-child-early", + status: "completed", + detail: { + type: "sub_agent", + log: "[Assistant] Early child findings.", + }, + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("keeps MultiAgentV2 interaction and interruption on the original child card", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation."); + await appServer.waitForTurnStart(); + + appServer.startsSubAgent({ + callId: "spawn-child-stable", + threadId: "child-thread-stable", + agentPath: "/root/stable-child", + }); + appServer.beginsSubAgentActivity({ + callId: "message-child-stable", + threadId: "child-thread-stable", + agentPath: "/root/stable-child", + kind: "interacted", + }); + appServer.completesSubAgentActivity({ + callId: "message-child-stable", + threadId: "child-thread-stable", + agentPath: "/root/stable-child", + kind: "interacted", + }); + appServer.says({ + threadId: "child-thread-stable", + itemId: "stable-child-message", + text: "Still on the same card.", + }); + appServer.beginsSubAgentActivity({ + callId: "interrupt-child-stable", + threadId: "child-thread-stable", + agentPath: "/root/stable-child", + kind: "interrupted", + }); + appServer.completesSubAgentActivity({ + callId: "interrupt-child-stable", + threadId: "child-thread-stable", + agentPath: "/root/stable-child", + kind: "interrupted", + }); + appServer.completeTurn(); + + const result = await resultPromise; + const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); + expect(new Set(toolCalls.map((item) => item.callId))).toEqual( + new Set(["spawn-child-stable"]), + ); + expect(toolCalls.at(-1)).toMatchObject({ + callId: "spawn-child-stable", + status: "canceled", + detail: { + type: "sub_agent", + log: "[Assistant] Still on the same card.", + }, + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("does not reopen a completed MultiAgentV2 child on activity completion", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation."); + await appServer.waitForTurnStart(); + + appServer.completeTurn({ threadId: "child-thread-fast" }); + const activity = { + callId: "spawn-child-fast", + threadId: "child-thread-fast", + agentPath: "/root/fast-child", + kind: "started" as const, + }; + appServer.beginsSubAgentActivity(activity); + appServer.completesSubAgentActivity(activity); + appServer.completeTurn(); + + const result = await resultPromise; + const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); + expect(toolCalls.map((item) => item.status)).toEqual(["running", "completed"]); + expect(toolCalls.at(-1)).toMatchObject({ + callId: "spawn-child-fast", + status: "completed", + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("preserves a completed child status when replaying a late compaction", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the investigation."); + await appServer.waitForTurnStart(); + + appServer.completeTurn({ threadId: "child-late-compaction" }); + appServer.completesCompaction({ + threadId: "child-late-compaction", + itemId: "late-child-compaction", + }); + appServer.startsSubAgent({ + callId: "spawn-child-late-compaction", + threadId: "child-late-compaction", + agentPath: "/root/late-compaction", + }); + appServer.completeTurn(); + + const result = await resultPromise; + const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); + expect(toolCalls.map((item) => item.status)).toEqual(["running", "completed", "completed"]); + expect(toolCalls.at(-1)).toMatchObject({ + callId: "spawn-child-late-compaction", + status: "completed", + detail: { type: "sub_agent", log: "[Compacted]" }, + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("projects legacy child tools into one stable sub-agent log", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + + try { + const resultPromise = session.run("Delegate the implementation."); + await appServer.waitForTurnStart(); + + appServer.startsSubAgent({ + callId: "spawn-legacy-tool-child", + threadId: "legacy-tool-child", + agentPath: "/root/legacy-tool-child", + }); + const command = { + threadId: "legacy-tool-child", + callId: "legacy-child-command", + command: "printf child", + output: "child output", + }; + appServer.runsLegacyCommand(command); + appServer.completesCommand(command); + appServer.appliesLegacyPatch({ + threadId: "legacy-tool-child", + callId: "legacy-child-patch", + path: "/workspace/project/src/child.ts", + diff: "@@\n-old\n+new\n", + }); + appServer.completeTurn({ threadId: "legacy-tool-child" }); + appServer.completeTurn(); + + const result = await resultPromise; + const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); + expect(new Set(toolCalls.map((item) => item.callId))).toEqual( + new Set(["spawn-legacy-tool-child"]), + ); + const finalToolCall = toolCalls.at(-1); + expect(finalToolCall).toMatchObject({ + callId: "spawn-legacy-tool-child", + status: "completed", + detail: { type: "sub_agent" }, + }); + if (finalToolCall?.detail.type === "sub_agent") { + expect(finalToolCall.detail.log.match(/\[Shell\]/g)).toHaveLength(1); + expect(finalToolCall.detail.log).toContain("[Edit]"); + } + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("keeps nested MultiAgentV2 output inside the root sub-agent card", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-child-root", + kind: "started", + agentThreadId: "child-thread-root", + agentPath: "/root/child", + }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "child-thread-root", + item: { + type: "subAgentActivity", + id: "spawn-grandchild", + kind: "started", + agentThreadId: "grandchild-thread", + agentPath: "/root/child/grandchild", + }, + }); + asInternals(session).handleNotification("item/agentMessage/delta", { + threadId: "grandchild-thread", + itemId: "grandchild-message", + delta: "Grandchild findings.", + }); + asInternals(session).handleNotification("turn/completed", { + threadId: "grandchild-thread", + turn: { status: "completed" }, + }); + + const beforeParentCompletes = events + .filter((event) => event.type === "timeline" && event.item.type === "tool_call") + .map((event) => event.item); + expect(new Set(beforeParentCompletes.map((item) => item.callId))).toEqual( + new Set(["spawn-child-root"]), + ); + expect(beforeParentCompletes.at(-1)).toMatchObject({ + callId: "spawn-child-root", + status: "running", + detail: { type: "sub_agent", log: expect.stringContaining("Grandchild findings.") }, + }); + + asInternals(session).handleNotification("turn/completed", { + threadId: "child-thread-root", + turn: { status: "completed" }, + }); + expect(events.at(-1)).toMatchObject({ + type: "timeline", + item: { callId: "spawn-child-root", status: "completed" }, + }); + }); + + test("never treats an unmapped foreign terminal as the root terminal", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("turn/completed", { + threadId: "unmapped-child-thread", + turn: { status: "completed" }, + }); + expect(events).toEqual([]); + + asInternals(session).handleNotification("turn/completed", { + threadId: "test-thread", + turn: { status: "completed" }, + }); + expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); + + asInternals(session).handleNotification("turn/started", { + threadId: "test-thread", + turn: { id: "next-root-turn" }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-reused-foreign-thread", + kind: "started", + agentThreadId: "unmapped-child-thread", + agentPath: "/root/reused-child", + }, + }); + expect(events.at(-1)).toMatchObject({ + type: "timeline", + item: { callId: "spawn-reused-foreign-thread", status: "running" }, + }); + }); + + test("routes msg-scoped legacy Codex events to their child thread", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-legacy-envelope-child", + kind: "started", + agentThreadId: "legacy-envelope-child", + agentPath: "/root/legacy-envelope-child", + }, + }); + asInternals(session).handleNotification("codex/event/exec_command_begin", { + msg: { + type: "exec_command_begin", + threadId: "legacy-envelope-child", + call_id: "child-command", + command: "pwd", + }, + }); + asInternals(session).handleNotification("codex/event/task_complete", { + msg: { + type: "task_complete", + thread_id: "legacy-envelope-child", + }, + }); + + expect( + events.some( + (event) => + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.callId === "child-command", + ), + ).toBe(false); + expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(0); + expect(events.at(-1)).toMatchObject({ + type: "timeline", + item: { + callId: "spawn-legacy-envelope-child", + status: "completed", + }, + }); + + asInternals(session).handleNotification("codex/event/task_complete", { + msg: { type: "task_complete" }, + }); + expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); + }); + + test("never replaces the root identity with an early child thread start", () => { + const session = createSession(); + + asInternals(session).handleNotification("thread/started", { + thread: { id: "child-thread-started-early" }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-child-thread-started-early", + kind: "started", + agentThreadId: "child-thread-started-early", + agentPath: "/root/early-thread", + }, + }); + + expect(session.currentThreadId).toBe("test-thread"); + }); + + test("does not leak aggregate child telemetry into the root timeline", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-child-telemetry", + kind: "started", + agentThreadId: "child-thread-telemetry", + agentPath: "/root/telemetry-child", + }, + }); + const eventCountAfterSpawn = events.length; + + asInternals(session).handleNotification("turn/plan/updated", { + threadId: "child-thread-telemetry", + plan: [{ step: "Child-only plan", status: "inProgress" }], + }); + + expect(events).toHaveLength(eventCountAfterSpawn); + }); + + test("keeps child context compaction inside the child card", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "spawn-child-compaction", + kind: "started", + agentThreadId: "child-thread-compaction", + agentPath: "/root/compacting-child", + }, + }); + asInternals(session).handleNotification("item/started", { + threadId: "child-thread-compaction", + item: { type: "contextCompaction", id: "child-compaction" }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "child-thread-compaction", + item: { type: "contextCompaction", id: "child-compaction" }, + }); + + const timelineItems = events.flatMap((event) => + event.type === "timeline" ? [event.item] : [], + ); + expect(timelineItems.every((item) => item.type === "tool_call")).toBe(true); + expect( + timelineItems.every( + (item) => item.type === "tool_call" && item.callId === "spawn-child-compaction", + ), + ).toBe(true); + expect(timelineItems.at(-1)).toMatchObject({ + type: "tool_call", + detail: { type: "sub_agent", log: "[Compacted]" }, + }); + }); + test("keeps the parent sub-agent running when a child command fails during the child turn", () => { const session = createSession(); const events: AgentStreamEvent[] = []; @@ -1673,6 +2266,178 @@ describe("Codex app-server provider", () => { ]); }); + test("loads mixed legacy and MultiAgentV2 sub-agent history", async () => { + const session = createSession(); + session.client = { + request: vi.fn(async (method: string) => { + if (method !== "thread/read") { + return {}; + } + return { + thread: { + turns: [ + { + items: [ + { + type: "collabAgentToolCall", + id: "legacy-spawn-history", + tool: "spawnAgent", + status: "completed", + prompt: "Legacy child", + receiverThreadIds: ["legacy-child-thread"], + agentsStates: { "legacy-child-thread": { status: "completed" } }, + }, + { + type: "subAgentActivity", + id: "v2-spawn-history", + kind: "started", + agentThreadId: "v2-child-thread", + agentPath: "/root/v2-child", + }, + ], + }, + ], + }, + }; + }), + }; + + await asInternals(session).loadPersistedHistory(); + + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) { + history.push(event); + } + expect( + history + .filter((event) => event.type === "timeline" && event.item.type === "tool_call") + .map((event) => event.item), + ).toMatchObject([ + { + callId: "legacy-spawn-history", + status: "completed", + detail: { type: "sub_agent", description: "Legacy child" }, + }, + { + callId: "v2-spawn-history", + status: "completed", + detail: { type: "sub_agent", description: "/root/v2-child" }, + }, + ]); + + const liveEvents: AgentStreamEvent[] = []; + session.subscribe((event) => liveEvents.push(event)); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "subAgentActivity", + id: "v2-interaction-after-resume", + kind: "interacted", + agentThreadId: "v2-child-thread", + agentPath: "/root/v2-child", + }, + }); + asInternals(session).handleNotification("item/agentMessage/delta", { + threadId: "v2-child-thread", + itemId: "v2-child-message-after-resume", + delta: "More findings after resume.", + }); + + const liveToolCalls = liveEvents.flatMap((event) => + event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [], + ); + expect(new Set(liveToolCalls.map((item) => item.callId))).toEqual( + new Set(["v2-spawn-history"]), + ); + expect(liveToolCalls.at(-1)).toMatchObject({ + status: "running", + detail: { type: "sub_agent", log: "[Assistant] More findings after resume." }, + }); + + liveEvents.length = 0; + asInternals(session).handleNotification("item/agentMessage/delta", { + threadId: "legacy-child-thread", + itemId: "legacy-child-message-after-resume", + delta: "Legacy findings after resume.", + }); + expect(liveEvents.at(-1)).toMatchObject({ + type: "timeline", + item: { + callId: "legacy-spawn-history", + status: "running", + detail: { type: "sub_agent", log: "[Assistant] Legacy findings after resume." }, + }, + }); + }); + + test("coalesces persisted MultiAgentV2 activity for one child into one terminal card", async () => { + const session = createSession(); + session.client = { + request: vi.fn(async (method: string) => { + if (method !== "thread/read") { + return {}; + } + return { + thread: { + turns: [ + { + items: [ + { + type: "subAgentActivity", + id: "child-started-history", + kind: "started", + agentThreadId: "history-child-thread", + agentPath: "/root/history-child", + timestamp: "2026-07-09T10:00:00.000Z", + }, + { + type: "subAgentActivity", + id: "child-interacted-history", + kind: "interacted", + agentThreadId: "history-child-thread", + agentPath: "/root/history-child", + timestamp: "2026-07-09T10:01:00.000Z", + }, + { + type: "subAgentActivity", + id: "child-interrupted-history", + kind: "interrupted", + agentThreadId: "history-child-thread", + agentPath: "/root/history-child", + timestamp: "2026-07-09T10:02:00.000Z", + }, + ], + }, + ], + }, + }; + }), + }; + + await asInternals(session).loadPersistedHistory(); + + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) { + history.push(event); + } + expect(history).toEqual([ + { + type: "timeline", + provider: "codex", + timestamp: "2026-07-09T10:00:00.000Z", + item: expect.objectContaining({ + type: "tool_call", + callId: "child-started-history", + status: "canceled", + detail: expect.objectContaining({ + type: "sub_agent", + description: "/root/history-child", + }), + }), + }, + ]); + }); + test("uses Codex turn timestamps for timestamp-less persisted history items", async () => { const session = createSession(); session.client = { 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 4681703c2..cd140fce6 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 @@ -91,7 +91,7 @@ import { buildCommandResolutionDiagnosticRows, resolveBinaryVersion, } from "./diagnostic-utils.js"; -import { runProviderTurn } from "./provider-runner.js"; +import { appendOrReplaceGrowingAssistantMessage, runProviderTurn } from "./provider-runner.js"; import { SETTING_APPLIES_NEXT_TURN_NOTICE } from "../provider-notices.js"; import type { WorkspaceGitService } from "../../workspace-git-service.js"; @@ -124,12 +124,17 @@ const CODEX_NON_ORIGINATING_APP_SERVER_CLIENT_INFO = { version: "0.0.0", } as const; const ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n"; +const MAX_PENDING_SUB_AGENT_THREADS = 32; +const MAX_PENDING_SUB_AGENT_NOTIFICATIONS_PER_THREAD = 128; +// COMPAT(codexLegacyCollabAgentToolCall): Codex <0.143 emits this shape. Added in +// Paseo v0.1.105; remove after 2027-01-09 once the supported Codex floor is >=0.143. const CODEX_TOOL_THREAD_ITEM_TYPES = new Set([ "commandExecution", "fileChange", "mcpToolCall", "webSearch", "collabAgentToolCall", + "subAgentActivity", ]); const CODEX_CONTEXT_COMPACTION_TYPE = "contextCompaction"; const CODEX_PLAN_IMPLEMENTATION_PROMPT_PREFIX = @@ -402,6 +407,16 @@ interface PersistedTimelineEntry { timestamp?: string; } +interface PersistedSubAgentRoute { + childThreadId: string; + toolCall: ToolCallTimelineItem; +} + +interface CodexThreadHistoryProjection { + timeline: PersistedTimelineEntry[]; + subAgentRoutes: PersistedSubAgentRoute[]; +} + function mergeCodexConfiguredDefaults( primary: CodexConfiguredDefaults, fallback: CodexConfiguredDefaults, @@ -1205,6 +1220,8 @@ function normalizeCodexThreadItemType(rawType: string | undefined): string | und return "webSearch"; case "CollabAgentToolCall": return "collabAgentToolCall"; + case "SubAgentActivity": + return "subAgentActivity"; case "ImageView": return "imageView"; case "ImageGeneration": @@ -1603,6 +1620,79 @@ function readCodexTurnHistoryTimestamp( return completedAt ?? startedAt; } +interface CodexSubAgentActivity { + id: string | null; + agentThreadId: string; + kind: "started" | "interacted" | "interrupted"; +} + +function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null { + const record = toObjectRecord(item); + if (!record) { + return null; + } + const normalizedType = normalizeCodexThreadItemType( + typeof record.type === "string" ? record.type : undefined, + ); + if ( + normalizedType !== "subAgentActivity" || + typeof record.agentThreadId !== "string" || + (record.kind !== "started" && record.kind !== "interacted" && record.kind !== "interrupted") + ) { + return null; + } + return { + id: nonEmptyString(record.id) ?? null, + agentThreadId: record.agentThreadId, + kind: record.kind, + }; +} + +function settleHistoricalSubAgentActivity( + item: ToolCallTimelineItem, + kind: CodexSubAgentActivity["kind"], +): ToolCallTimelineItem { + // thread/read returns completed parent items, not a live child snapshot. + // Only an explicit interruption remains non-completed when replayed. + return { + ...item, + status: kind === "interrupted" ? "canceled" : "completed", + error: null, + }; +} + +function updateHistoricalSubAgentActivity( + timeline: PersistedTimelineEntry[], + index: number, + kind: CodexSubAgentActivity["kind"], +): void { + const existing = timeline[index]; + if (existing?.item.type !== "tool_call") { + return; + } + timeline[index] = { + ...existing, + item: settleHistoricalSubAgentActivity(existing.item, kind), + }; +} + +function readCodexHistoricalSubAgentThreadIds(item: unknown): string[] { + const activity = readCodexSubAgentActivity(item); + if (activity) { + return [activity.agentThreadId]; + } + const record = toObjectRecord(item); + const normalizedType = normalizeCodexThreadItemType( + typeof record?.type === "string" ? record.type : undefined, + ); + if (normalizedType !== "collabAgentToolCall" || !Array.isArray(record?.receiverThreadIds)) { + return []; + } + return record.receiverThreadIds.filter( + (threadId): threadId is string => typeof threadId === "string" && threadId.length > 0, + ); +} + function codexImageOutputFromResult(result: unknown): ProviderImageOutput | null { if (typeof result === "string") { const trimmed = result.trim(); @@ -1764,22 +1854,52 @@ async function loadCodexThreadHistoryTimeline(params: { threadId: string; cwd: string | null; requestThread: CodexThreadReadRequest; -}): Promise { +}): Promise { const response = await requestCodexThreadHistory(params.requestThread, params.threadId); const timeline: PersistedTimelineEntry[] = []; + const subAgentTimelineIndexByThreadId = new Map(); for (const turn of response.thread.turns) { for (const item of turn.items) { + const historicalSubAgentActivity = readCodexSubAgentActivity(item); + if (historicalSubAgentActivity) { + const existingIndex = subAgentTimelineIndexByThreadId.get( + historicalSubAgentActivity.agentThreadId, + ); + if (existingIndex !== undefined) { + updateHistoricalSubAgentActivity( + timeline, + existingIndex, + historicalSubAgentActivity.kind, + ); + continue; + } + } for (const timelineItem of threadItemToTimelineEntries(item, { cwd: params.cwd })) { const timestamp = readCodexHistoryTimestamp(item) ?? readCodexTurnHistoryTimestamp(turn, timelineItem); + const settledTimelineItem = + historicalSubAgentActivity && timelineItem.type === "tool_call" + ? settleHistoricalSubAgentActivity(timelineItem, historicalSubAgentActivity.kind) + : timelineItem; timeline.push({ - item: timelineItem, + item: settledTimelineItem, timestamp: timestamp ?? undefined, }); + for (const childThreadId of readCodexHistoricalSubAgentThreadIds(item)) { + subAgentTimelineIndexByThreadId.set(childThreadId, timeline.length - 1); + } } } } - return timeline; + const subAgentRoutes = Array.from(subAgentTimelineIndexByThreadId.entries()).flatMap( + ([childThreadId, timelineIndex]): PersistedSubAgentRoute[] => { + const item = timeline[timelineIndex]?.item; + return item?.type === "tool_call" && item.detail.type === "sub_agent" + ? [{ childThreadId, toolCall: item }] + : []; + }, + ); + return { timeline, subAgentRoutes }; } function readCodexThread(client: CodexAppServerClientLike, threadId: string): Promise { @@ -1855,6 +1975,7 @@ const TurnCompletedNotificationSchema = z const TurnPlanUpdatedNotificationSchema = z .object({ + threadId: z.string().optional(), plan: z.array( z .object({ @@ -1868,12 +1989,14 @@ const TurnPlanUpdatedNotificationSchema = z const TurnDiffUpdatedNotificationSchema = z .object({ + threadId: z.string().optional(), diff: z.string(), }) .passthrough(); const ThreadTokenUsageUpdatedNotificationSchema = z .object({ + threadId: z.string().optional(), tokenUsage: z.unknown(), }) .passthrough(); @@ -1905,10 +2028,25 @@ const ContextCompactedNotificationSchema = z }) .passthrough(); +const CodexEventThreadIdFields = { + threadId: z.string().optional(), + thread_id: z.string().optional(), +}; + +function getCodexEventThreadId(params: { + threadId?: string; + thread_id?: string; + msg: { threadId?: string; thread_id?: string }; +}): string | null { + return params.threadId ?? params.thread_id ?? params.msg.threadId ?? params.msg.thread_id ?? null; +} + const CodexEventTurnAbortedNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("turn_aborted"), reason: z.string().optional(), }) @@ -1918,8 +2056,10 @@ const CodexEventTurnAbortedNotificationSchema = z const CodexEventTaskCompleteNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("task_complete"), }) .passthrough(), @@ -1928,12 +2068,11 @@ const CodexEventTaskCompleteNotificationSchema = z const CodexEventItemLifecycleNotificationSchema = z .object({ - threadId: z.string().optional(), + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.enum(["item_started", "item_completed"]), - threadId: z.string().optional(), - thread_id: z.string().optional(), item: z .object({ id: z.string().optional(), @@ -1947,8 +2086,10 @@ const CodexEventItemLifecycleNotificationSchema = z const CodexEventExecCommandBeginNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("exec_command_begin"), call_id: z.string().optional(), command: z.unknown().optional(), @@ -1960,8 +2101,10 @@ const CodexEventExecCommandBeginNotificationSchema = z const CodexEventExecCommandEndNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("exec_command_end"), call_id: z.string().optional(), command: z.unknown().optional(), @@ -1981,8 +2124,10 @@ const CodexEventExecCommandEndNotificationSchema = z const CodexEventExecCommandOutputDeltaNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("exec_command_output_delta"), call_id: z.string().optional(), stream: z.string().optional(), @@ -1995,8 +2140,10 @@ const CodexEventExecCommandOutputDeltaNotificationSchema = z const CodexEventTerminalInteractionNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("terminal_interaction"), call_id: z.string().optional(), process_id: z.union([z.string(), z.number()]).optional(), @@ -2008,6 +2155,7 @@ const CodexEventTerminalInteractionNotificationSchema = z const ItemCommandExecutionTerminalInteractionNotificationSchema = z .object({ + threadId: z.string().optional(), itemId: z.string().optional(), processId: z.union([z.string(), z.number()]).optional(), stdin: z.string().optional(), @@ -2016,8 +2164,10 @@ const ItemCommandExecutionTerminalInteractionNotificationSchema = z const CodexEventPatchApplyBeginNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("patch_apply_begin"), call_id: z.string().optional(), changes: z.unknown().optional(), @@ -2028,8 +2178,10 @@ const CodexEventPatchApplyBeginNotificationSchema = z const CodexEventPatchApplyEndNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("patch_apply_end"), call_id: z.string().optional(), changes: z.unknown().optional(), @@ -2043,6 +2195,7 @@ const CodexEventPatchApplyEndNotificationSchema = z const ItemFileChangeOutputDeltaNotificationSchema = z .object({ + threadId: z.string().optional(), itemId: z.string(), delta: z.string().optional(), chunk: z.string().optional(), @@ -2051,8 +2204,10 @@ const ItemFileChangeOutputDeltaNotificationSchema = z const CodexEventTurnDiffNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("turn_diff"), unified_diff: z.string().optional(), diff: z.string().optional(), @@ -2063,8 +2218,10 @@ const CodexEventTurnDiffNotificationSchema = z const CodexEventThreadRolledBackNotificationSchema = z .object({ + ...CodexEventThreadIdFields, msg: z .object({ + ...CodexEventThreadIdFields, type: z.literal("thread_rolled_back"), num_turns: z.number().int().nonnegative().optional(), numTurns: z.number().int().nonnegative().optional(), @@ -2082,9 +2239,13 @@ type ParsedCodexNotification = 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: "plan_updated"; + plan: Array<{ step: string | null; status: string | null }>; + threadId: string | null; + } + | { kind: "diff_updated"; diff: string; threadId: string | null } + | { kind: "token_usage_updated"; tokenUsage: unknown; threadId: string | null } | { kind: "agent_message_delta"; itemId: string; delta: string; threadId: string | null } | { kind: "reasoning_delta"; itemId: string; delta: string; threadId: string | null } | { @@ -2104,6 +2265,7 @@ type ParsedCodexNotification = callId: string | null; command: unknown; cwd: string | null; + threadId: string | null; } | { kind: "exec_command_completed"; @@ -2114,12 +2276,14 @@ type ParsedCodexNotification = exitCode: number | null; success: boolean | null; stderr: string | null; + threadId: string | null; } | { kind: "exec_command_output_delta"; callId: string | null; stream: string | null; chunk: string | null; + threadId: string | null; } | { kind: "terminal_interaction"; @@ -2127,11 +2291,13 @@ type ParsedCodexNotification = callId: string | null; processId: string | null; stdin: string | null; + threadId: string | null; } | { kind: "patch_apply_started"; callId: string | null; changes: unknown; + threadId: string | null; } | { kind: "patch_apply_completed"; @@ -2140,13 +2306,15 @@ type ParsedCodexNotification = stdout: string | null; stderr: string | null; success: boolean | null; + threadId: string | null; } | { kind: "file_change_output_delta"; itemId: string; delta: string | null; + threadId: string | null; } - | { kind: "thread_rolled_back"; numTurns: number } + | { kind: "thread_rolled_back"; numTurns: number; threadId: string | null } | { kind: "context_compacted"; threadId: string; turnId: string | null } | { kind: "invalid_payload"; method: string; params: unknown } | { kind: "unknown_method"; method: string; params: unknown }; @@ -2162,6 +2330,15 @@ type CodexDeltaNotification = Extract< } >; +type CodexThreadRoute = + | { kind: "root" } + | { kind: "sub_agent"; callId: string } + | { kind: "pending_sub_agent"; threadId: string }; + +function getCodexNotificationThreadId(parsed: ParsedCodexNotification): string | null { + return "threadId" in parsed ? parsed.threadId : null; +} + function isCodexDeltaNotification( parsed: ParsedCodexNotification, ): parsed is CodexDeltaNotification { @@ -2229,6 +2406,7 @@ const CodexNotificationSchema = z.union([ step: entry.step ?? null, status: entry.status ?? null, })), + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("turn/plan/updated"), params: z.unknown() }).transform( @@ -2241,7 +2419,11 @@ const CodexNotificationSchema = z.union([ z .object({ method: z.literal("turn/diff/updated"), params: TurnDiffUpdatedNotificationSchema }) .transform( - ({ params }): ParsedCodexNotification => ({ kind: "diff_updated", diff: params.diff }), + ({ params }): ParsedCodexNotification => ({ + kind: "diff_updated", + diff: params.diff, + threadId: params.threadId ?? null, + }), ), z.object({ method: z.literal("turn/diff/updated"), params: z.unknown() }).transform( ({ method, params }): ParsedCodexNotification => ({ @@ -2259,6 +2441,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "token_usage_updated", tokenUsage: params.tokenUsage, + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("thread/tokenUsage/updated"), params: z.unknown() }).transform( @@ -2367,7 +2550,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_started", source: "codex_event", - threadId: params.threadId ?? params.msg.threadId ?? params.msg.thread_id ?? null, + threadId: getCodexEventThreadId(params), item: params.msg.item, }), ), @@ -2387,7 +2570,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "item_completed", source: "codex_event", - threadId: params.threadId ?? params.msg.threadId ?? params.msg.thread_id ?? null, + threadId: getCodexEventThreadId(params), item: params.msg.item, }), ), @@ -2409,6 +2592,7 @@ const CodexNotificationSchema = z.union([ callId: params.msg.call_id ?? null, command: params.msg.command ?? null, cwd: params.msg.cwd ?? null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/exec_command_begin"), params: z.unknown() }).transform( @@ -2438,6 +2622,7 @@ const CodexNotificationSchema = z.union([ exitCode: params.msg.exit_code ?? params.msg.exitCode ?? null, success: params.msg.success ?? null, stderr: params.msg.stderr ?? null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/exec_command_end"), params: z.unknown() }).transform( @@ -2458,6 +2643,7 @@ const CodexNotificationSchema = z.union([ callId: params.msg.call_id ?? null, stream: params.msg.stream ?? null, chunk: params.msg.chunk ?? params.msg.delta ?? null, + threadId: getCodexEventThreadId(params), }), ), z @@ -2487,6 +2673,7 @@ const CodexNotificationSchema = z.union([ ? String(params.msg.process_id) : (params.msg.process_id ?? null), stdin: params.msg.stdin ?? null, + threadId: getCodexEventThreadId(params), }), ), z @@ -2513,6 +2700,7 @@ const CodexNotificationSchema = z.union([ ? String(params.processId) : (params.processId ?? null), stdin: params.stdin ?? null, + threadId: params.threadId ?? null, }), ), z @@ -2537,6 +2725,7 @@ const CodexNotificationSchema = z.union([ kind: "patch_apply_started", callId: params.msg.call_id ?? null, changes: params.msg.changes ?? null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/patch_apply_begin"), params: z.unknown() }).transform( @@ -2559,6 +2748,7 @@ const CodexNotificationSchema = z.union([ stdout: params.msg.stdout ?? null, stderr: params.msg.stderr ?? null, success: params.msg.success ?? null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/patch_apply_end"), params: z.unknown() }).transform( @@ -2578,6 +2768,7 @@ const CodexNotificationSchema = z.union([ kind: "file_change_output_delta", itemId: params.itemId, delta: params.delta ?? params.chunk ?? null, + threadId: params.threadId ?? null, }), ), z.object({ method: z.literal("item/fileChange/outputDelta"), params: z.unknown() }).transform( @@ -2596,6 +2787,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "diff_updated", diff: params.msg.unified_diff ?? params.msg.diff ?? "", + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/turn_diff"), params: z.unknown() }).transform( @@ -2611,11 +2803,11 @@ const CodexNotificationSchema = z.union([ params: CodexEventTurnAbortedNotificationSchema, }) .transform( - (): ParsedCodexNotification => ({ + ({ params }): ParsedCodexNotification => ({ kind: "turn_completed", status: "interrupted", errorMessage: null, - threadId: null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/turn_aborted"), params: z.unknown() }).transform( @@ -2631,11 +2823,11 @@ const CodexNotificationSchema = z.union([ params: CodexEventTaskCompleteNotificationSchema, }) .transform( - (): ParsedCodexNotification => ({ + ({ params }): ParsedCodexNotification => ({ kind: "turn_completed", status: "completed", errorMessage: null, - threadId: null, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/task_complete"), params: z.unknown() }).transform( @@ -2654,6 +2846,7 @@ const CodexNotificationSchema = z.union([ ({ params }): ParsedCodexNotification => ({ kind: "thread_rolled_back", numTurns: params.msg.num_turns ?? params.msg.numTurns ?? 0, + threadId: getCodexEventThreadId(params), }), ), z.object({ method: z.literal("codex/event/thread_rolled_back"), params: z.unknown() }).transform( @@ -2861,6 +3054,10 @@ function buildCodexCustomProviderConfig( interface CodexSubAgentCallState { callId: string; toolCall: ToolCallTimelineItem; + parentCallId: string | null; + activityItemIds: Set; + pendingCommandOutputDeltas: Map; + pendingFileChangeOutputDeltas: Map; childItemOrder: string[]; childItems: Map; } @@ -2908,6 +3105,7 @@ export class CodexAppServerAgentSession implements AgentSession { private emittedItemCompletedIds = new Set(); private subAgentCallsByCallId = new Map(); private subAgentCallIdByChildThreadId = new Map(); + private pendingSubAgentNotificationsByThreadId = new Map(); private warnedUnknownNotificationMethods = new Set(); private warnedInvalidNotificationPayloads = new Set(); private warnedIncompleteEditToolCallIds = new Set(); @@ -3240,23 +3438,32 @@ export class CodexAppServerAgentSession implements AgentSession { const client = this.client; const threadId = this.currentThreadId; - const timeline = await loadCodexThreadHistoryTimeline({ + const history = await loadCodexThreadHistoryTimeline({ threadId, cwd: this.config.cwd ?? null, requestThread: (threadIdToRead) => { return readCodexThread(client, threadIdToRead); }, }); + const { timeline, subAgentRoutes } = history; + this.subAgentCallsByCallId.clear(); + this.subAgentCallIdByChildThreadId.clear(); + this.pendingSubAgentNotificationsByThreadId.clear(); + for (const route of subAgentRoutes) { + this.registerSubAgentToolCall({ + timelineItem: route.toolCall, + rawItem: { agentThreadId: route.childThreadId }, + parentCallId: null, + }); + } this.resetCodexUserMessageTurns(); for (const entry of timeline) { if (entry.item.type === "user_message") { this.rememberCodexUserMessageTurn(entry.item.messageId); } } - if (timeline.length > 0) { - this.persistedHistory = timeline; - this.historyPending = true; - } + this.persistedHistory = timeline; + this.historyPending = timeline.length > 0; } private async ensureThreadLoaded(): Promise { @@ -3472,6 +3679,9 @@ export class CodexAppServerAgentSession implements AgentSession { } async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { + let currentAssistantMessageId: string | null = null; + let currentAssistantMessageHasBoundary = false; + let hasAssistantMessage = false; return runProviderTurn({ prompt, runOptions: options, @@ -3480,9 +3690,31 @@ export class CodexAppServerAgentSession implements AgentSession { getSessionId: async () => (await this.getRuntimeInfo()).sessionId ?? "", reduceFinalText: ({ current, item }) => { if (item.type === "assistant_message") { - return item.text; + const hasPreviousAssistantMessage = hasAssistantMessage; + hasAssistantMessage = true; + const isNewMessage = + item.messageId === undefined || item.messageId !== currentAssistantMessageId; + if (isNewMessage) { + currentAssistantMessageId = item.messageId ?? null; + currentAssistantMessageHasBoundary = + hasPreviousAssistantMessage && + item.text.startsWith(ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN); + } + const finalTextItem = currentAssistantMessageHasBoundary + ? { + ...item, + text: item.text.startsWith(ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN) + ? item.text.slice(ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN.length) + : item.text, + } + : item; + return isNewMessage + ? finalTextItem.text + : appendOrReplaceGrowingAssistantMessage({ current, item: finalTextItem }); } if (item.type === "tool_call" && item.detail.type === "plan") { + currentAssistantMessageId = null; + currentAssistantMessageHasBoundary = false; return item.detail.text; } return current; @@ -3914,6 +4146,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.pendingPermissionHandlers.clear(); this.pendingPermissions.clear(); this.resolvedPermissionRequests.clear(); + this.pendingSubAgentNotificationsByThreadId.clear(); this.subscribers.clear(); this.activeForegroundTurnId = null; if (this.client) { @@ -4239,6 +4472,56 @@ export class CodexAppServerAgentSession implements AgentSession { private handleNotification(method: string, params: unknown): void { const parsed = CodexNotificationSchema.parse({ method, params }); this.traceParsedNotification(method, params, parsed); + const route = this.resolveCodexThreadRoute(getCodexNotificationThreadId(parsed)); + if (route.kind === "pending_sub_agent") { + this.bufferPendingSubAgentNotification(route.threadId, parsed); + return; + } + if (route.kind === "sub_agent") { + this.dispatchSubAgentNotification(parsed, route.callId); + return; + } + this.dispatchParsedNotification(parsed); + } + + private dispatchSubAgentNotification(parsed: ParsedCodexNotification, callId: string): void { + switch (parsed.kind) { + case "thread_started": + this.emitSubAgentActivityUpdate(callId, "running"); + return; + case "turn_started": + case "turn_completed": + case "agent_message_delta": + case "reasoning_delta": + case "item_started": + case "item_completed": + this.dispatchParsedNotification(parsed); + return; + case "exec_command_output_delta": + case "file_change_output_delta": + this.handleCodexDeltaNotification(parsed, callId); + return; + case "exec_command_started": + this.handleExecCommandStartedNotification(parsed, callId); + return; + case "exec_command_completed": + this.handleExecCommandCompletedNotification(parsed, callId); + return; + case "patch_apply_started": + this.handlePatchApplyStartedNotification(parsed, callId); + return; + case "patch_apply_completed": + this.handlePatchApplyCompletedNotification(parsed, callId); + return; + default: + // Aggregate child telemetry is redundant and must not leak into the + // root timeline. Concrete legacy tools are projected above for Codex + // versions that do not also emit canonical item lifecycle events. + return; + } + } + + private dispatchParsedNotification(parsed: ParsedCodexNotification): void { if (isCodexDeltaNotification(parsed)) { this.handleCodexDeltaNotification(parsed); return; @@ -4299,6 +4582,52 @@ export class CodexAppServerAgentSession implements AgentSession { } } + private resolveCodexThreadRoute(threadId: string | null): CodexThreadRoute { + if (!threadId || !this.currentThreadId || threadId === this.currentThreadId) { + return { kind: "root" }; + } + const callId = this.subAgentCallIdByChildThreadId.get(threadId); + return callId ? { kind: "sub_agent", callId } : { kind: "pending_sub_agent", threadId }; + } + + private bufferPendingSubAgentNotification( + threadId: string, + parsed: ParsedCodexNotification, + ): void { + let pending = this.pendingSubAgentNotificationsByThreadId.get(threadId); + if (!pending) { + if (this.pendingSubAgentNotificationsByThreadId.size >= MAX_PENDING_SUB_AGENT_THREADS) { + const oldestThreadId = this.pendingSubAgentNotificationsByThreadId.keys().next().value; + if (typeof oldestThreadId === "string") { + this.pendingSubAgentNotificationsByThreadId.delete(oldestThreadId); + } + } + pending = []; + this.pendingSubAgentNotificationsByThreadId.set(threadId, pending); + } + if (pending.length >= MAX_PENDING_SUB_AGENT_NOTIFICATIONS_PER_THREAD) { + pending.shift(); + } + pending.push(parsed); + } + + private replayPendingSubAgentNotifications(threadIds: readonly string[]): void { + for (const threadId of threadIds) { + const pending = this.pendingSubAgentNotificationsByThreadId.get(threadId); + if (!pending) { + continue; + } + this.pendingSubAgentNotificationsByThreadId.delete(threadId); + const callId = this.subAgentCallIdByChildThreadId.get(threadId); + if (!callId) { + continue; + } + for (const parsed of pending) { + this.dispatchSubAgentNotification(parsed, callId); + } + } + } + private handleThreadStateNotification(parsed: ParsedCodexNotification): boolean { switch (parsed.kind) { case "context_compacted": @@ -4338,12 +4667,14 @@ export class CodexAppServerAgentSession implements AgentSession { return this.subAgentCallIdByChildThreadId.get(threadId) ?? null; } - private registerSubAgentToolCall( - timelineItem: ToolCallTimelineItem, - rawItem: { [key: string]: unknown }, - ): void { + private registerSubAgentToolCall(params: { + timelineItem: ToolCallTimelineItem; + rawItem: { [key: string]: unknown }; + parentCallId: string | null; + }): string[] { + const { timelineItem, rawItem, parentCallId } = params; if (timelineItem.detail.type !== "sub_agent") { - return; + return []; } const existing = this.subAgentCallsByCallId.get(timelineItem.callId); @@ -4352,6 +4683,10 @@ export class CodexAppServerAgentSession implements AgentSession { ({ callId: timelineItem.callId, toolCall: timelineItem, + parentCallId, + activityItemIds: new Set(), + pendingCommandOutputDeltas: new Map(), + pendingFileChangeOutputDeltas: new Map(), childItemOrder: [], childItems: new Map(), } satisfies CodexSubAgentCallState); @@ -4365,14 +4700,90 @@ export class CodexAppServerAgentSession implements AgentSession { (state.toolCall.detail.type === "sub_agent" ? state.toolCall.detail.log : ""), }, }; + state.parentCallId ??= parentCallId; + const activity = readCodexSubAgentActivity(rawItem); + if (activity?.id) { + state.activityItemIds.add(activity.id); + } 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) { + const agentThreadId = + typeof rawItem.agentThreadId === "string" && rawItem.agentThreadId.length > 0 + ? rawItem.agentThreadId + : null; + const childThreadIds = Array.from( + new Set(agentThreadId ? [...receiverThreadIds, agentThreadId] : receiverThreadIds), + ); + for (const receiverThreadId of childThreadIds) { this.subAgentCallIdByChildThreadId.set(receiverThreadId, timelineItem.callId); } + return childThreadIds; + } + + private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean { + const activity = readCodexSubAgentActivity(rawItem); + if (!activity) { + return false; + } + const callId = this.subAgentCallIdByChildThreadId.get(activity.agentThreadId); + if (!callId) { + return false; + } + const state = this.subAgentCallsByCallId.get(callId); + if (!state) { + return false; + } + if (activity.id && state.activityItemIds.has(activity.id)) { + return true; + } + if (activity.id) { + state.activityItemIds.add(activity.id); + } + this.emitSubAgentActivityUpdate( + callId, + activity.kind === "interrupted" ? "canceled" : "running", + ); + return true; + } + + private handleCompletedContextCompactionItem(item: { + id?: string; + type?: string; + [key: string]: unknown; + }): boolean { + if (!this.isContextCompactionItem(item)) { + return false; + } + if (this.unpairedCompactionNotificationCompletions > 0) { + this.unpairedCompactionNotificationCompletions -= 1; + return true; + } + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("completed", item.id), + }); + this.unpairedCompactionItemCompletions += 1; + return true; + } + + private handleCompletedSpecialItem( + parsed: Extract, + childSubAgentCallId: string | null, + ): boolean { + if ( + childSubAgentCallId && + this.handleSubAgentContextCompactionItem(childSubAgentCallId, parsed.item, "completed") + ) { + return true; + } + return ( + this.handleCompletedContextCompactionItem(parsed.item) || + this.handleRegisteredSubAgentActivity(parsed.item) + ); } private upsertSubAgentChildItem(callId: string, itemId: string, item: AgentTimelineItem): void { @@ -4386,6 +4797,21 @@ export class CodexAppServerAgentSession implements AgentSession { state.childItems.set(itemId, item); } + private emitCodexToolTimelineItem( + timelineItem: ToolCallTimelineItem, + subAgentCallId: string | null, + ): void { + if (!subAgentCallId) { + this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + return; + } + this.upsertSubAgentChildItem(subAgentCallId, timelineItem.callId, timelineItem); + this.emitSubAgentActivityUpdate( + subAgentCallId, + timelineItem.status === "running" ? "running" : undefined, + ); + } + private getSubAgentChildTimeline(state: CodexSubAgentCallState): AgentTimelineItem[] { return state.childItemOrder .map((itemId) => state.childItems.get(itemId)) @@ -4426,6 +4852,11 @@ export class CodexAppServerAgentSession implements AgentSession { error: null, }; state.toolCall = nextToolCall; + if (state.parentCallId && state.parentCallId !== callId) { + this.upsertSubAgentChildItem(state.parentCallId, state.callId, nextToolCall); + this.emitSubAgentActivityUpdate(state.parentCallId); + return; + } this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: nextToolCall }); } @@ -4445,6 +4876,21 @@ export class CodexAppServerAgentSession implements AgentSession { this.emitSubAgentActivityUpdate(callId, "running"); } + private handleSubAgentContextCompactionItem( + callId: string, + item: { id?: string; type?: string; [key: string]: unknown }, + status: "loading" | "completed", + ): boolean { + if (!this.isContextCompactionItem(item)) { + return false; + } + if (item.id) { + this.upsertSubAgentChildItem(callId, item.id, { type: "compaction", status }); + } + this.emitSubAgentActivityUpdate(callId); + return true; + } + private shouldSkipCompletedThreadItem( timelineItem: AgentTimelineItem, normalizedItemType: string | undefined, @@ -4458,7 +4904,10 @@ export class CodexAppServerAgentSession implements AgentSession { return Boolean(itemId && this.emittedItemCompletedIds.has(itemId)); } - private handleCodexDeltaNotification(parsed: CodexDeltaNotification): void { + private handleCodexDeltaNotification( + parsed: CodexDeltaNotification, + routedSubAgentCallId: string | null = null, + ): void { if (parsed.kind === "agent_message_delta") { const prev = this.pendingAgentMessages.get(parsed.itemId) ?? ""; const text = prev + parsed.delta; @@ -4512,12 +4961,23 @@ export class CodexAppServerAgentSession implements AgentSession { return; } if (parsed.kind === "exec_command_output_delta") { - this.appendOutputDeltaChunk(this.pendingCommandOutputDeltas, parsed.callId, parsed.chunk, { + const outputDeltas = routedSubAgentCallId + ? this.subAgentCallsByCallId.get(routedSubAgentCallId)?.pendingCommandOutputDeltas + : this.pendingCommandOutputDeltas; + if (!outputDeltas) { + return; + } + this.appendOutputDeltaChunk(outputDeltas, parsed.callId, parsed.chunk, { decodeBase64: true, }); return; } - this.appendOutputDeltaChunk(this.pendingFileChangeOutputDeltas, parsed.itemId, parsed.delta); + const outputDeltas = routedSubAgentCallId + ? this.subAgentCallsByCallId.get(routedSubAgentCallId)?.pendingFileChangeOutputDeltas + : this.pendingFileChangeOutputDeltas; + if (outputDeltas) { + this.appendOutputDeltaChunk(outputDeltas, parsed.itemId, parsed.delta); + } } private handleThreadStartedNotification( @@ -4577,6 +5037,7 @@ export class CodexAppServerAgentSession implements AgentSession { }); } this.activeForegroundTurnId = null; + this.pendingSubAgentNotificationsByThreadId.clear(); this.resetTurnTrackingState(); } @@ -4707,10 +5168,19 @@ export class CodexAppServerAgentSession implements AgentSession { private handleExecCommandStartedNotification( parsed: Extract, + subAgentCallId: string | null = null, ): void { - if (parsed.callId) { + const outputDeltas = subAgentCallId + ? this.subAgentCallsByCallId.get(subAgentCallId)?.pendingCommandOutputDeltas + : this.pendingCommandOutputDeltas; + if (!outputDeltas) { + return; + } + if (parsed.callId && !subAgentCallId) { this.emittedExecCommandStartedCallIds.add(parsed.callId); - this.pendingCommandOutputDeltas.delete(parsed.callId); + } + if (parsed.callId) { + outputDeltas.delete(parsed.callId); } const timelineItem = mapCodexExecNotificationToToolCall({ callId: parsed.callId, @@ -4719,16 +5189,25 @@ export class CodexAppServerAgentSession implements AgentSession { running: true, }); if (timelineItem) { - this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + this.emitCodexToolTimelineItem(timelineItem, subAgentCallId); } } private handleExecCommandCompletedNotification( parsed: Extract, + subAgentCallId: string | null = null, ): void { - const bufferedOutput = this.consumeOutputDelta(this.pendingCommandOutputDeltas, parsed.callId); + const outputDeltas = subAgentCallId + ? this.subAgentCallsByCallId.get(subAgentCallId)?.pendingCommandOutputDeltas + : this.pendingCommandOutputDeltas; + if (!outputDeltas) { + return; + } + const bufferedOutput = this.consumeOutputDelta(outputDeltas, parsed.callId); const resolvedOutput = parsed.output ?? bufferedOutput; - this.rememberTerminalProcessForCommand(parsed.command, resolvedOutput); + if (!subAgentCallId) { + this.rememberTerminalProcessForCommand(parsed.command, resolvedOutput); + } const timelineItem = mapCodexExecNotificationToToolCall({ callId: parsed.callId, command: parsed.command, @@ -4740,8 +5219,10 @@ export class CodexAppServerAgentSession implements AgentSession { running: false, }); if (timelineItem) { - this.emittedExecCommandCompletedCallIds.add(timelineItem.callId); - this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + if (!subAgentCallId) { + this.emittedExecCommandCompletedCallIds.add(timelineItem.callId); + } + this.emitCodexToolTimelineItem(timelineItem, subAgentCallId); } } @@ -4768,9 +5249,16 @@ export class CodexAppServerAgentSession implements AgentSession { private handlePatchApplyStartedNotification( parsed: Extract, + subAgentCallId: string | null = null, ): void { + const outputDeltas = subAgentCallId + ? this.subAgentCallsByCallId.get(subAgentCallId)?.pendingFileChangeOutputDeltas + : this.pendingFileChangeOutputDeltas; + if (!outputDeltas) { + return; + } if (parsed.callId) { - this.pendingFileChangeOutputDeltas.delete(parsed.callId); + outputDeltas.delete(parsed.callId); } const timelineItem = mapCodexPatchNotificationToToolCall({ callId: parsed.callId, @@ -4783,17 +5271,21 @@ export class CodexAppServerAgentSession implements AgentSession { callId: parsed.callId, changes: parsed.changes, }); - this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + this.emitCodexToolTimelineItem(timelineItem, subAgentCallId); } } private handlePatchApplyCompletedNotification( parsed: Extract, + subAgentCallId: string | null = null, ): void { - const bufferedOutput = this.consumeOutputDelta( - this.pendingFileChangeOutputDeltas, - parsed.callId, - ); + const outputDeltas = subAgentCallId + ? this.subAgentCallsByCallId.get(subAgentCallId)?.pendingFileChangeOutputDeltas + : this.pendingFileChangeOutputDeltas; + if (!outputDeltas) { + return; + } + const bufferedOutput = this.consumeOutputDelta(outputDeltas, parsed.callId); const timelineItem = mapCodexPatchNotificationToToolCall({ callId: parsed.callId, changes: parsed.changes, @@ -4809,7 +5301,7 @@ export class CodexAppServerAgentSession implements AgentSession { changes: parsed.changes, stdout: parsed.stdout, }); - this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + this.emitCodexToolTimelineItem(timelineItem, subAgentCallId); } } @@ -4826,17 +5318,8 @@ export class CodexAppServerAgentSession implements AgentSession { this.handleUserMessageItem(parsed); return; } - if (this.isContextCompactionItem(parsed.item)) { - if (this.unpairedCompactionNotificationCompletions > 0) { - this.unpairedCompactionNotificationCompletions -= 1; - return; - } - this.emitEvent({ - type: "timeline", - provider: CODEX_PROVIDER, - item: this.createContextCompactionTimelineItem("completed", parsed.item.id), - }); - this.unpairedCompactionItemCompletions += 1; + const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if (this.handleCompletedSpecialItem(parsed, childSubAgentCallId)) { return; } const timelineItem = threadItemToTimeline(parsed.item, { @@ -4846,9 +5329,17 @@ export class CodexAppServerAgentSession implements AgentSession { if (!timelineItem) { return; } - const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + const registeredChildThreadIds = + timelineItem.type === "tool_call" + ? this.registerSubAgentToolCall({ + timelineItem, + rawItem: parsed.item, + parentCallId: childSubAgentCallId, + }) + : []; if (childSubAgentCallId) { this.handleSubAgentChildItemCompleted(childSubAgentCallId, parsed.item.id, timelineItem); + this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } const normalizedItemType = normalizeCodexThreadItemType( @@ -4856,6 +5347,7 @@ export class CodexAppServerAgentSession implements AgentSession { ); const itemId = parsed.item.id; if (this.shouldSkipCompletedThreadItem(timelineItem, normalizedItemType, itemId)) { + this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } if (this.consumeStreamedTextCompletion(timelineItem, itemId)) { @@ -4866,11 +5358,11 @@ export class CodexAppServerAgentSession implements AgentSession { this.emittedItemCompletedIds.add(itemId); this.emittedItemStartedIds.delete(itemId); } + this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } 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 @@ -4896,6 +5388,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.pendingCommandOutputDeltas.delete(itemId); this.pendingFileChangeOutputDeltas.delete(itemId); } + this.replayPendingSubAgentNotifications(registeredChildThreadIds); } private consumeStreamedTextCompletion( @@ -4978,6 +5471,13 @@ export class CodexAppServerAgentSession implements AgentSession { this.handleUserMessageItem(parsed); return; } + const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + if ( + childSubAgentCallId && + this.handleSubAgentContextCompactionItem(childSubAgentCallId, parsed.item, "loading") + ) { + return; + } if (this.isContextCompactionItem(parsed.item)) { this.emitEvent({ type: "timeline", @@ -4986,6 +5486,9 @@ export class CodexAppServerAgentSession implements AgentSession { }); return; } + if (this.handleRegisteredSubAgentActivity(parsed.item)) { + return; + } const timelineItem = threadItemToTimeline(parsed.item, { includeUserMessage: false, cwd: this.config.cwd ?? null, @@ -4993,12 +5496,17 @@ export class CodexAppServerAgentSession implements AgentSession { if (!timelineItem || timelineItem.type !== "tool_call") { return; } - const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); + const registeredChildThreadIds = this.registerSubAgentToolCall({ + timelineItem, + rawItem: parsed.item, + parentCallId: childSubAgentCallId, + }); if (childSubAgentCallId) { if (parsed.item.id) { this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem); } this.emitSubAgentActivityUpdate(childSubAgentCallId, "running"); + this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } const normalizedItemType = normalizeCodexThreadItemType( @@ -5015,13 +5523,13 @@ export 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); this.pendingCommandOutputDeltas.delete(itemId); this.pendingFileChangeOutputDeltas.delete(itemId); } + this.replayPendingSubAgentNotifications(registeredChildThreadIds); } private handleUserMessageItem( diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index 4b85c714e..5a3da2409 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -6,6 +6,25 @@ import type { AgentSession, AgentStreamEvent } from "../../../agent-sdk-types.js type JsonObject = Record; type FakeCodexAppServerHandler = (params: unknown) => unknown; +interface FakeSubAgentActivity { + callId: string; + threadId: string; + agentPath: string; + kind: "started" | "interacted" | "interrupted"; + parentThreadId?: string; +} +interface FakeLegacyCommand { + threadId: string; + callId: string; + command: string; + output: string; +} +interface FakeLegacyPatch { + threadId: string; + callId: string; + path: string; + diff: string; +} type CodexAppServerChildProcess = ChildProcessWithoutNullStreams & { stdin: PassThrough; stdout: PassThrough; @@ -18,7 +37,21 @@ export interface FakeCodexAppServer { assertNoErrors(): void; waitForTurnStart(): Promise; nextResponse(): Promise; + startsTurn(params: { threadId: string; turnId?: string }): void; completeTurn(params?: { threadId?: string }): void; + startsSubAgent(params: { + callId: string; + threadId: string; + agentPath: string; + parentThreadId?: string; + }): void; + beginsSubAgentActivity(params: FakeSubAgentActivity): void; + completesSubAgentActivity(params: FakeSubAgentActivity): void; + completesCompaction(params: { threadId: string; itemId: string }): void; + runsLegacyCommand(params: FakeLegacyCommand): void; + appliesLegacyPatch(params: FakeLegacyPatch): void; + completesCommand(params: FakeLegacyCommand): void; + says(params: { threadId: string; itemId?: string; text: string; chunks?: string[] }): void; requestCommandApproval(params: { itemId: string; threadId: string; @@ -195,6 +228,34 @@ export function createFakeCodexAppServer( }); } + function writeSubAgentActivity( + method: "item/started" | "item/completed", + params: FakeSubAgentActivity, + ): void { + writeNotification(method, { + threadId: params.parentThreadId ?? "thread-1", + item: { + type: "subAgentActivity", + id: params.callId, + kind: params.kind, + agentThreadId: params.threadId, + agentPath: params.agentPath, + }, + }); + } + + function writeNotification(method: string, params: JsonObject): void { + child.stdout.write(`${JSON.stringify({ method, params })}\n`); + } + + function completeItem(threadId: string, item: JsonObject): void { + writeNotification("item/completed", { threadId, item }); + } + + function writeLegacyEvent(threadId: string, method: string, msg: JsonObject): void { + writeNotification(method, { threadId, msg }); + } + return { child, recordedRollbacks, @@ -215,6 +276,17 @@ export function createFakeCodexAppServer( child.stdin.once("data", (chunk) => resolve(chunk.toString())); }); }, + startsTurn(params) { + child.stdout.write( + `${JSON.stringify({ + method: "turn/started", + params: { + threadId: params.threadId, + turn: { id: params.turnId ?? `turn-${params.threadId}` }, + }, + })}\n`, + ); + }, completeTurn(params = {}) { child.stdout.write( `${JSON.stringify({ @@ -223,6 +295,88 @@ export function createFakeCodexAppServer( })}\n`, ); }, + startsSubAgent(params) { + writeSubAgentActivity("item/completed", { ...params, kind: "started" }); + }, + beginsSubAgentActivity(params) { + writeSubAgentActivity("item/started", params); + }, + completesSubAgentActivity(params) { + writeSubAgentActivity("item/completed", params); + }, + completesCompaction(params) { + completeItem(params.threadId, { type: "contextCompaction", id: params.itemId }); + }, + runsLegacyCommand(params) { + writeLegacyEvent(params.threadId, "codex/event/exec_command_begin", { + type: "exec_command_begin", + call_id: params.callId, + command: params.command, + }); + writeLegacyEvent(params.threadId, "codex/event/exec_command_output_delta", { + type: "exec_command_output_delta", + call_id: params.callId, + chunk: params.output, + }); + writeLegacyEvent(params.threadId, "codex/event/exec_command_end", { + type: "exec_command_end", + call_id: params.callId, + command: params.command, + exit_code: 0, + success: true, + }); + }, + appliesLegacyPatch(params) { + const changes = [ + { + path: params.path, + kind: "modify", + unified_diff: params.diff, + }, + ]; + for (const [method, type] of [ + ["codex/event/patch_apply_begin", "patch_apply_begin"], + ["codex/event/patch_apply_end", "patch_apply_end"], + ] as const) { + writeLegacyEvent(params.threadId, method, { + type, + call_id: params.callId, + changes, + ...(type === "patch_apply_end" ? { success: true } : {}), + }); + } + }, + completesCommand(params) { + completeItem(params.threadId, { + type: "commandExecution", + id: params.callId, + status: "completed", + command: params.command, + aggregatedOutput: params.output, + exitCode: 0, + }); + }, + says(params) { + if (params.itemId) { + for (const chunk of params.chunks ?? [params.text]) { + child.stdout.write( + `${JSON.stringify({ + method: "item/agentMessage/delta", + params: { + threadId: params.threadId, + itemId: params.itemId, + delta: chunk, + }, + })}\n`, + ); + } + } + completeItem(params.threadId, { + type: "agentMessage", + ...(params.itemId ? { id: params.itemId } : {}), + text: params.text, + }); + }, requestCommandApproval(params) { const requestId = nextServerRequestId; nextServerRequestId += 1; 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 af4fab521..32ca89563 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 @@ -221,6 +221,49 @@ describe("codex tool-call mapper", () => { }); }); + it.each([ + ["started", "running"], + ["interacted", "running"], + ["interrupted", "canceled"], + ] as const)("maps subAgentActivity %s into canonical sub-agent detail", (kind, status) => { + const item = mapCodexToolCallFromThreadItem({ + type: "subAgentActivity", + id: `activity-${kind}`, + kind, + agentThreadId: "child-thread-1", + agentPath: "/root/investigator", + }); + + expect(item).toEqual({ + type: "tool_call", + callId: `activity-${kind}`, + name: "Sub-agent", + status, + error: null, + detail: { + type: "sub_agent", + subAgentType: "Sub-agent", + description: "/root/investigator", + log: "", + actions: [], + }, + }); + }); + + it("preserves an empty subAgentActivity path as an empty description", () => { + const item = mapCodexToolCallFromThreadItem({ + type: "subAgentActivity", + id: "activity-empty-path", + kind: "started", + agentThreadId: "child-thread-empty-path", + agentPath: "", + }); + + expect(item).toMatchObject({ + detail: { type: "sub_agent", description: "" }, + }); + }); + it("does not fail a collabAgentToolCall from child error state alone", () => { const item = mapCodexToolCallFromThreadItem({ type: "collabAgentToolCall", 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 4979b7649..710529af0 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 @@ -193,6 +193,16 @@ const CodexCollabAgentToolCallItemSchema = z }) .passthrough(); +const CodexSubAgentActivityItemSchema = z + .object({ + type: z.literal("subAgentActivity"), + id: z.string().min(1), + kind: z.enum(["started", "interacted", "interrupted"]), + agentThreadId: z.string().min(1), + agentPath: z.string(), + }) + .passthrough(); + const CodexToolThreadItemSchema = z.discriminatedUnion("type", [ CodexCommandExecutionItemSchema, CodexFileChangeItemSchema, @@ -206,6 +216,7 @@ const CodexThreadItemSchema = z.discriminatedUnion("type", [ CodexMcpToolCallItemSchema, CodexWebSearchItemSchema, CodexCollabAgentToolCallItemSchema, + CodexSubAgentActivityItemSchema, ]); function maybeUnwrapShellWrapperCommand(command: string): string { @@ -977,6 +988,25 @@ function mapCollabAgentToolCallItem( }; } +function mapSubAgentActivityItem( + item: z.infer, +): ToolCallTimelineItem { + return { + type: "tool_call", + callId: item.id, + name: "Sub-agent", + status: item.kind === "interrupted" ? "canceled" : "running", + error: null, + detail: { + type: "sub_agent", + subAgentType: "Sub-agent", + description: item.agentPath, + log: "", + actions: [], + }, + }; +} + function mapThreadItemToNormalizedEnvelope( item: z.infer, options?: CodexMapperOptions, @@ -1012,6 +1042,9 @@ export function mapCodexToolCallFromThreadItem( if (parsed.data.type === "collabAgentToolCall") { return mapCollabAgentToolCallItem(parsed.data); } + if (parsed.data.type === "subAgentActivity") { + return mapSubAgentActivityItem(parsed.data); + } const envelope = mapThreadItemToNormalizedEnvelope(parsed.data, options); if (!envelope) { return null;