mirror of
https://github.com/getpaseo/paseo.git
synced 2026-07-29 12:01:31 +00:00
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
This commit is contained in:
Mohamed Boudra
@@ -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[] = [
|
||||
{
|
||||
|
||||
@@ -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":
|
||||
|
||||
@@ -45,9 +45,17 @@ function assistant(
|
||||
options?: {
|
||||
provider?: AgentProvider;
|
||||
turnId?: string;
|
||||
messageId?: string;
|
||||
},
|
||||
): Extract<AgentStreamEvent, { type: "timeline" }> {
|
||||
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();
|
||||
|
||||
|
||||
@@ -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<string, PendingAgentStreamBuffer>();
|
||||
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
|
||||
) {
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -6,6 +6,25 @@ import type { AgentSession, AgentStreamEvent } from "../../../agent-sdk-types.js
|
||||
|
||||
type JsonObject = Record<string, unknown>;
|
||||
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<JsonObject>;
|
||||
nextResponse(): Promise<string>;
|
||||
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;
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<typeof CodexSubAgentActivityItemSchema>,
|
||||
): 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<typeof CodexToolThreadItemSchema>,
|
||||
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;
|
||||
|
||||
Reference in New Issue
Block a user