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;