mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 13:13:25 +00:00
fix(codex): reconcile nullable collaboration results with spawned agents
This commit is contained in:
@@ -5004,6 +5004,120 @@ describe("Codex app-server provider", () => {
|
||||
});
|
||||
});
|
||||
|
||||
function collaboration(tool: string, states: Record<string, string>, id = tool) {
|
||||
return {
|
||||
type: "collabAgentToolCall",
|
||||
id,
|
||||
tool,
|
||||
status: "completed",
|
||||
prompt: tool === "spawnAgent" ? "Inspect the regression" : null,
|
||||
receiverThreadIds: Object.keys(states),
|
||||
agentsStates: Object.fromEntries(
|
||||
Object.entries(states).map(([threadId, status]) => [threadId, { status }]),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
test("replays nullable wait and close results onto the original spawn cards", async () => {
|
||||
const session = createSession();
|
||||
const items = [
|
||||
collaboration("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collaboration("spawnAgent", { second: "pendingInit" }, "spawn-second"),
|
||||
collaboration("wait", { first: "completed", second: "running" }, "wait-first"),
|
||||
collaboration("wait", { second: "completed" }, "wait-second"),
|
||||
collaboration("closeAgent", { first: "shutdown", second: "completed" }),
|
||||
];
|
||||
session.client = {
|
||||
request: vi.fn(async (_method: string, params: unknown) => ({
|
||||
thread: {
|
||||
turns: [
|
||||
{ items: (params as { threadId: string }).threadId === "test-thread" ? items : [] },
|
||||
],
|
||||
},
|
||||
})),
|
||||
};
|
||||
|
||||
const history = await Array.fromAsync(session.streamHistory());
|
||||
|
||||
expect(
|
||||
history.filter((event) => event.type === "timeline").map((event) => event.item),
|
||||
).toMatchObject([
|
||||
{ callId: "spawn-first", status: "completed" },
|
||||
{ callId: "spawn-second", status: "completed" },
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
),
|
||||
).toMatchObject([
|
||||
{ id: "first", toolCallId: "spawn-first", status: "completed" },
|
||||
{ id: "second", toolCallId: "spawn-second", status: "completed" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("updates live children independently and reopens only an explicit follow-up", async () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
async function notify(tool: string, states: Record<string, string>, id = tool) {
|
||||
await asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "test-thread",
|
||||
item: collaboration(tool, states, id),
|
||||
});
|
||||
}
|
||||
await notify("spawnAgent", { first: "pendingInit" }, "spawn-first");
|
||||
await notify("spawnAgent", { second: "pendingInit" }, "spawn-second");
|
||||
events.length = 0;
|
||||
await notify("wait", { first: "completed", second: "running" });
|
||||
|
||||
expect(
|
||||
events.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
),
|
||||
).toMatchObject([{ id: "first", toolCallId: "spawn-first", status: "completed" }]);
|
||||
expect(
|
||||
events.filter((event) => event.type === "timeline").map((event) => event.item),
|
||||
).toMatchObject([{ callId: "spawn-first", status: "completed" }]);
|
||||
|
||||
events.length = 0;
|
||||
await notify("closeAgent", { first: "shutdown" });
|
||||
await notify("wait", { first: "running" }, "stale-wait");
|
||||
await notify("sendInput", { first: "running" });
|
||||
expect(
|
||||
events.filter((event) => event.type === "timeline").map((event) => event.item),
|
||||
).toMatchObject([{ callId: "spawn-first", status: "running" }]);
|
||||
});
|
||||
|
||||
test.each([
|
||||
["shutdown", "canceled"],
|
||||
["notFound", "failed"],
|
||||
])("replays %s as %s without completing an unfinished child", async (childStatus, expected) => {
|
||||
const session = createSession();
|
||||
const items = [
|
||||
collaboration("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
collaboration("spawnAgent", { second: "pendingInit" }, "spawn-second"),
|
||||
collaboration("closeAgent", { first: childStatus }),
|
||||
collaboration("wait", {}),
|
||||
];
|
||||
session.client = {
|
||||
request: vi.fn(async (_method: string, params: unknown) => ({
|
||||
thread: {
|
||||
turns: [
|
||||
{ items: (params as { threadId: string }).threadId === "test-thread" ? items : [] },
|
||||
],
|
||||
},
|
||||
})),
|
||||
};
|
||||
const history = await Array.fromAsync(session.streamHistory());
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
),
|
||||
).toMatchObject([
|
||||
{ id: "first", status: expected },
|
||||
{ id: "second", status: "running" },
|
||||
]);
|
||||
});
|
||||
test("loads mixed legacy and MultiAgentV2 sub-agent history", async () => {
|
||||
const session = createSession();
|
||||
session.client = {
|
||||
|
||||
@@ -1857,6 +1857,59 @@ function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null
|
||||
};
|
||||
}
|
||||
|
||||
interface CodexCollabAgentUpdate {
|
||||
threadId: string;
|
||||
status: ToolCallTimelineItem["status"];
|
||||
shutdown: boolean;
|
||||
reopen: boolean;
|
||||
}
|
||||
|
||||
function readCodexCollabAgentUpdates(item: unknown): CodexCollabAgentUpdate[] {
|
||||
const record = toObjectRecord(item);
|
||||
if (
|
||||
!record ||
|
||||
normalizeCodexThreadItemType(typeof record.type === "string" ? record.type : undefined) !==
|
||||
"collabAgentToolCall" ||
|
||||
record.tool === "spawnAgent"
|
||||
) {
|
||||
return [];
|
||||
}
|
||||
return Object.entries(toObjectRecord(record.agentsStates) ?? {}).flatMap(([threadId, value]) => {
|
||||
const state = toObjectRecord(value);
|
||||
if (typeof state?.status !== "string") return [];
|
||||
const mapped = mapCodexToolCallFromThreadItem({
|
||||
type: "collabAgentToolCall",
|
||||
id: record.id,
|
||||
tool: record.tool,
|
||||
status: "completed",
|
||||
agentsStates: { [threadId]: state },
|
||||
});
|
||||
return mapped
|
||||
? [
|
||||
{
|
||||
threadId,
|
||||
status: mapped.status,
|
||||
shutdown: state.status === "shutdown",
|
||||
reopen: record.tool === "sendInput" && record.status === "completed",
|
||||
},
|
||||
]
|
||||
: [];
|
||||
});
|
||||
}
|
||||
|
||||
function resolveCodexCollabAgentUpdate(
|
||||
previous: ToolCallTimelineItem["status"],
|
||||
update: CodexCollabAgentUpdate,
|
||||
): ToolCallTimelineItem["status"] {
|
||||
if (
|
||||
isTerminalSubAgentStatus(previous) &&
|
||||
(update.shutdown || (update.status === "running" && !update.reopen))
|
||||
) {
|
||||
return previous;
|
||||
}
|
||||
return update.status;
|
||||
}
|
||||
|
||||
function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean {
|
||||
return source === "codex_event" && !readCodexSubAgentActivity(item);
|
||||
}
|
||||
@@ -2060,6 +2113,27 @@ async function threadItemToTimelineEntries(
|
||||
return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))];
|
||||
}
|
||||
|
||||
function updateHistoricalCollabAgentCalls(
|
||||
timeline: PersistedTimelineEntry[],
|
||||
timelineIndexByThreadId: Map<string, number>,
|
||||
item: unknown,
|
||||
): boolean {
|
||||
let updated = false;
|
||||
for (const update of readCodexCollabAgentUpdates(item)) {
|
||||
const existingIndex = timelineIndexByThreadId.get(update.threadId);
|
||||
const existing = existingIndex === undefined ? undefined : timeline[existingIndex];
|
||||
if (existingIndex === undefined || existing?.item.type !== "tool_call") continue;
|
||||
const status = resolveCodexCollabAgentUpdate(existing.item.status, update);
|
||||
const toolCall: ToolCallTimelineItem =
|
||||
status === "failed"
|
||||
? { ...existing.item, status, error: { message: "Sub-agent failed" } }
|
||||
: { ...existing.item, status, error: null };
|
||||
timeline[existingIndex] = { ...existing, item: toolCall };
|
||||
updated = true;
|
||||
}
|
||||
return updated;
|
||||
}
|
||||
|
||||
async function loadCodexThreadHistoryTimeline(params: {
|
||||
client: CodexAppServerClientLike;
|
||||
threadId: string;
|
||||
@@ -2095,6 +2169,9 @@ async function projectCodexThreadHistory(
|
||||
const { timeline, subAgentTimelineIndexByThreadId } = projection;
|
||||
for await (const row of readCodexThreadHistory(params.client, params.threadId)) {
|
||||
const { item } = row;
|
||||
if (updateHistoricalCollabAgentCalls(timeline, subAgentTimelineIndexByThreadId, item)) {
|
||||
continue;
|
||||
}
|
||||
const historicalSubAgentActivity = readCodexSubAgentActivity(item);
|
||||
if (historicalSubAgentActivity) {
|
||||
const existingIndex = subAgentTimelineIndexByThreadId.get(
|
||||
@@ -5737,6 +5814,19 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean {
|
||||
let updatedExistingSubagent = false;
|
||||
for (const update of readCodexCollabAgentUpdates(rawItem)) {
|
||||
const callId = this.subAgentCallIdByChildThreadId.get(update.threadId);
|
||||
if (!callId) continue;
|
||||
const state = this.subAgentCallsByCallId.get(callId);
|
||||
if (!state) continue;
|
||||
const status = resolveCodexCollabAgentUpdate(state.toolCall.status, update);
|
||||
if (status !== state.toolCall.status) {
|
||||
this.emitSubAgentActivityUpdate(callId, status, { reopen: update.reopen });
|
||||
}
|
||||
updatedExistingSubagent = true;
|
||||
}
|
||||
if (updatedExistingSubagent) return true;
|
||||
const activity = readCodexSubAgentActivity(rawItem);
|
||||
if (!activity) {
|
||||
return false;
|
||||
|
||||
@@ -213,6 +213,20 @@ describe("codex tool-call mapper", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("accepts a nullable prompt in completed collaboration control items", () => {
|
||||
expect(
|
||||
mapCodexToolCallFromThreadItem({
|
||||
type: "collabAgentToolCall",
|
||||
id: "wait-child",
|
||||
tool: "wait",
|
||||
prompt: null,
|
||||
status: "completed",
|
||||
receiverThreadIds: ["child"],
|
||||
agentsStates: { child: { status: "completed" } },
|
||||
}),
|
||||
).toMatchObject({ callId: "wait-child", status: "completed" });
|
||||
});
|
||||
|
||||
it("maps collabAgentToolCall into canonical sub-agent detail", () => {
|
||||
const item = mapCodexToolCallFromThreadItem({
|
||||
type: "collabAgentToolCall",
|
||||
|
||||
@@ -186,7 +186,7 @@ const CodexCollabAgentToolCallItemSchema = z
|
||||
id: z.string().min(1),
|
||||
status: z.string().optional(),
|
||||
error: z.unknown().optional(),
|
||||
prompt: z.string().optional(),
|
||||
prompt: z.string().nullish(),
|
||||
tool: z.string().optional(),
|
||||
receiverThreadIds: z.array(z.string()).optional(),
|
||||
agentsStates: z.record(z.string(), z.unknown()).optional(),
|
||||
|
||||
Reference in New Issue
Block a user