fix(codex): reconcile nullable collaboration results with spawned agents

This commit is contained in:
popstk
2026-10-30 04:31:05 +00:00
committed by paseo-bot[bot]
parent deb1bdc266
commit 198d8a5c68
4 changed files with 219 additions and 1 deletions
@@ -5001,6 +5001,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 = {
@@ -1860,6 +1860,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);
}
@@ -2093,6 +2146,27 @@ async function requestCodexThreadHistory(
return CodexThreadReadResponseSchema.parse(response);
}
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: {
threadId: string;
cwd: string | null;
@@ -2103,6 +2177,9 @@ async function loadCodexThreadHistoryTimeline(params: {
const subAgentTimelineIndexByThreadId = new Map<string, number>();
for (const turn of response.thread.turns) {
for (const item of turn.items) {
if (updateHistoricalCollabAgentCalls(timeline, subAgentTimelineIndexByThreadId, item)) {
continue;
}
const historicalSubAgentActivity = readCodexSubAgentActivity(item);
if (historicalSubAgentActivity) {
const existingIndex = subAgentTimelineIndexByThreadId.get(
@@ -5761,6 +5838,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(),