Compare commits

...
Author SHA1 Message Date
paseo-bot[bot] ec97c8982b fix(codex): keep control cards from claiming child routes
Only spawn announcements establish child ownership. Wait and close results update existing children, while failed resume attempts retain their own visible result without replacing a previous child outcome.
2026-10-30 06:45:18 +00:00
paseo-bot[bot] fca6e24fc0 fix(codex): restore outcomes after child follow-ups
Replay must distinguish an earlier terminal outcome from the next outcome after an accepted follow-up. Keep that distinction with each projected child, and normalize wait and close results consistently for original spawn cards and standalone control cards.
2026-10-30 06:27:32 +00:00
paseo-bot[bot] 9615afb837 fix(codex): preserve follow-up instructions in parent history
Follow-up submissions carry user-visible instructions and must keep their own cards. Their sampled child states describe submission time, so only the submission result settles the card while child lifecycle events continue to own child status.
2026-10-30 06:16:48 +00:00
paseo-bot[bot] a98fa27901 test(server): collect fixture garbage before Git filter measurement 2026-10-30 06:03:39 +00:00
paseo-bot[bot] 5e7c7e609e fix(codex): leave follow-up status to child lifecycle events 2026-10-30 06:03:39 +00:00
paseo-bot[bot] 8207f697c5 fix(codex): preserve final errors and settle rendered wait cards 2026-10-30 06:03:39 +00:00
paseo-bot[bot] a02fa39125 fix(codex): settle closed children independently of prior errors
Codex derives a close record's status from the child's previous state,
so closing an errored child produces a failed record even after shutdown.
Treat close as a terminal-preserving child event rather than a successful
control-card status, retaining earlier completion when a close is stale.
2026-10-30 06:03:39 +00:00
paseo-bot[bot] 6041e86dad fix(codex): fold live control starts into existing subagents
A wait starts without child states. Recognize it from its receivers so
live handling consumes both lifecycle notifications, as replay does,
instead of leaving an extra running control card behind. Normalize
child status directly with the mapper's existing child-status rule.
2026-10-30 06:03:39 +00:00
cbfb29c5fa fix(codex): settle children closed before they finish
A successful close records the child's previous running state. Replaying
that snapshot must stop counting the child as working. Exercise replay
and live control updates through the session interface instead of calling
private notification methods.

Co-authored-by: geeknonerd <14756651+geeknonerd@users.noreply.github.com>
Co-authored-by: altenli <liqian115@gmail.com>
2026-10-30 06:03:39 +00:00
popstk dc23fe6545 fix(codex): reconcile nullable collaboration results with spawned agents 2026-10-30 06:03:39 +00:00
5 changed files with 656 additions and 26 deletions
@@ -58,6 +58,7 @@ import {
mapCodexToolCallEnvelope, mapCodexToolCallEnvelope,
mapCodexToolCallFromThreadItem, mapCodexToolCallFromThreadItem,
normalizeCommandExecutionCommand, normalizeCommandExecutionCommand,
normalizeCollabAgentChildStatus,
splitCodexMcpToolResultImages, splitCodexMcpToolResultImages,
} from "./codex/tool-call-mapper.js"; } from "./codex/tool-call-mapper.js";
import { import {
@@ -1857,6 +1858,77 @@ function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null
}; };
} }
interface CodexCollabAgentUpdate {
threadId: string;
status: ToolCallTimelineItem["status"];
preservesTerminal: boolean;
}
type CodexCollabAgentControl =
| { kind: "submission"; childThreadIds: string[] }
| { kind: "outcome"; childThreadIds: string[]; updates: CodexCollabAgentUpdate[] };
interface HistoricalCodexSubAgent {
timelineIndex: number;
pendingFollowUp: boolean;
}
function readCodexCollabAgentControl(item: unknown): CodexCollabAgentControl | null {
const record = toObjectRecord(item);
if (
!record ||
normalizeCodexThreadItemType(typeof record.type === "string" ? record.type : undefined) !==
"collabAgentToolCall" ||
record.tool === "spawnAgent"
) {
return null;
}
const tool = typeof record.tool === "string" ? record.tool : undefined;
const receivers = Array.isArray(record.receiverThreadIds)
? record.receiverThreadIds.filter((id): id is string => typeof id === "string")
: [];
if (tool === "sendInput") {
return record.status === "completed" && record.error == null
? { kind: "submission", childThreadIds: receivers }
: null;
}
if (tool !== "wait" && tool !== "closeAgent") return null;
const outcomeStates = toObjectRecord(record.agentsStates);
const updates = Object.entries(outcomeStates ?? {}).flatMap(
([threadId, value]): CodexCollabAgentUpdate[] => {
const state = toObjectRecord(value);
if (typeof state?.status !== "string") return [];
// Unlike a spawn snapshot, a control result can report the child's final error.
const status = normalizeCollabAgentChildStatus(state.status, tool);
return [
{
threadId,
status,
preservesTerminal: state.status === "shutdown" || record.tool === "closeAgent",
},
];
},
);
return {
kind: "outcome",
childThreadIds: [...new Set([...receivers, ...updates.map((update) => update.threadId)])],
updates,
};
}
function resolveCodexCollabAgentUpdate(
previous: ToolCallTimelineItem["status"],
update: CodexCollabAgentUpdate,
): ToolCallTimelineItem["status"] {
if (
isTerminalSubAgentStatus(previous) &&
(update.preservesTerminal || update.status === "running")
) {
return previous;
}
return update.status;
}
function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean { function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean {
return source === "codex_event" && !readCodexSubAgentActivity(item); return source === "codex_event" && !readCodexSubAgentActivity(item);
} }
@@ -1897,7 +1969,7 @@ function updateHistoricalSubAgentActivity(
}; };
} }
function readCodexHistoricalSubAgentThreadIds(item: unknown): string[] { function readCodexSpawnedSubAgentThreadIds(item: unknown): string[] {
const activity = readCodexSubAgentActivity(item); const activity = readCodexSubAgentActivity(item);
if (activity) { if (activity) {
return activity.kind === "started" ? [activity.agentThreadId] : []; return activity.kind === "started" ? [activity.agentThreadId] : [];
@@ -2060,23 +2132,57 @@ async function threadItemToTimelineEntries(
return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))]; return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))];
} }
function updateHistoricalCollabAgentCalls(
timeline: PersistedTimelineEntry[],
subAgentsByThreadId: Map<string, HistoricalCodexSubAgent>,
item: unknown,
): boolean {
const control = readCodexCollabAgentControl(item);
if (!control) return false;
if (control.kind === "submission") {
for (const threadId of control.childThreadIds) {
const child = subAgentsByThreadId.get(threadId);
if (child) child.pendingFollowUp = true;
}
return false;
}
for (const update of control.updates) {
const child = subAgentsByThreadId.get(update.threadId);
const existing = child === undefined ? undefined : timeline[child.timelineIndex];
if (!child || existing?.item.type !== "tool_call") continue;
const status = child.pendingFollowUp
? update.status
: resolveCodexCollabAgentUpdate(existing.item.status, update);
child.pendingFollowUp = false;
const toolCall: ToolCallTimelineItem =
status === "failed"
? { ...existing.item, status, error: { message: "Sub-agent failed" } }
: { ...existing.item, status, error: null };
timeline[child.timelineIndex] = { ...existing, item: toolCall };
}
return (
control.childThreadIds.length > 0 &&
control.childThreadIds.every((id) => subAgentsByThreadId.has(id))
);
}
async function loadCodexThreadHistoryTimeline(params: { async function loadCodexThreadHistoryTimeline(params: {
client: CodexAppServerClientLike; client: CodexAppServerClientLike;
threadId: string; threadId: string;
cwd: string | null; cwd: string | null;
}): Promise<CodexThreadHistoryProjection> { }): Promise<CodexThreadHistoryProjection> {
const timeline: PersistedTimelineEntry[] = []; const timeline: PersistedTimelineEntry[] = [];
const subAgentTimelineIndexByThreadId = new Map<string, number>(); const subAgentsByThreadId = new Map<string, HistoricalCodexSubAgent>();
try { try {
await projectCodexThreadHistory(params, { timeline, subAgentTimelineIndexByThreadId }); await projectCodexThreadHistory(params, { timeline, subAgentsByThreadId });
} catch (error) { } catch (error) {
if (!(error instanceof CodexThreadHistoryTooLargeError)) throw error; if (!(error instanceof CodexThreadHistoryTooLargeError)) throw error;
// The session stays usable; the row says why the history stops here. // The session stays usable; the row says why the history stops here.
timeline.push({ item: { type: "notification", level: "warning", message: error.message } }); timeline.push({ item: { type: "notification", level: "warning", message: error.message } });
} }
const subAgentRoutes = Array.from(subAgentTimelineIndexByThreadId.entries()).flatMap( const subAgentRoutes = Array.from(subAgentsByThreadId.entries()).flatMap(
([childThreadId, timelineIndex]): PersistedSubAgentRoute[] => { ([childThreadId, child]): PersistedSubAgentRoute[] => {
const item = timeline[timelineIndex]?.item; const item = timeline[child.timelineIndex]?.item;
return item?.type === "tool_call" && item.detail.type === "sub_agent" return item?.type === "tool_call" && item.detail.type === "sub_agent"
? [{ childThreadId, toolCall: item }] ? [{ childThreadId, toolCall: item }]
: []; : [];
@@ -2089,17 +2195,20 @@ async function projectCodexThreadHistory(
params: { client: CodexAppServerClientLike; threadId: string; cwd: string | null }, params: { client: CodexAppServerClientLike; threadId: string; cwd: string | null },
projection: { projection: {
timeline: PersistedTimelineEntry[]; timeline: PersistedTimelineEntry[];
subAgentTimelineIndexByThreadId: Map<string, number>; subAgentsByThreadId: Map<string, HistoricalCodexSubAgent>;
}, },
): Promise<void> { ): Promise<void> {
const { timeline, subAgentTimelineIndexByThreadId } = projection; const { timeline, subAgentsByThreadId } = projection;
for await (const row of readCodexThreadHistory(params.client, params.threadId)) { for await (const row of readCodexThreadHistory(params.client, params.threadId)) {
const { item } = row; const { item } = row;
if (updateHistoricalCollabAgentCalls(timeline, subAgentsByThreadId, item)) {
continue;
}
const historicalSubAgentActivity = readCodexSubAgentActivity(item); const historicalSubAgentActivity = readCodexSubAgentActivity(item);
if (historicalSubAgentActivity) { if (historicalSubAgentActivity) {
const existingIndex = subAgentTimelineIndexByThreadId.get( const existingIndex = subAgentsByThreadId.get(
historicalSubAgentActivity.agentThreadId, historicalSubAgentActivity.agentThreadId,
); )?.timelineIndex;
if (existingIndex !== undefined) { if (existingIndex !== undefined) {
const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd }); const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd });
updateHistoricalSubAgentActivity( updateHistoricalSubAgentActivity(
@@ -2129,8 +2238,11 @@ async function projectCodexThreadHistory(
? { providerTurnId: row.turnId } ? { providerTurnId: row.turnId }
: {}), : {}),
}); });
for (const childThreadId of readCodexHistoricalSubAgentThreadIds(item)) { for (const childThreadId of readCodexSpawnedSubAgentThreadIds(item)) {
subAgentTimelineIndexByThreadId.set(childThreadId, timeline.length - 1); subAgentsByThreadId.set(childThreadId, {
timelineIndex: timeline.length - 1,
pendingFollowUp: false,
});
} }
} }
} }
@@ -4026,6 +4138,7 @@ export class CodexAppServerAgentSession implements AgentSession {
this.registerSubAgentToolCall({ this.registerSubAgentToolCall({
timelineItem: next.route.toolCall, timelineItem: next.route.toolCall,
rawItem: { agentThreadId: next.route.childThreadId }, rawItem: { agentThreadId: next.route.childThreadId },
childThreadIds: [next.route.childThreadId],
parentCallId: next.parentCallId, parentCallId: next.parentCallId,
parentSubagentId: next.parentSubagentId, parentSubagentId: next.parentSubagentId,
}); });
@@ -5673,6 +5786,7 @@ export class CodexAppServerAgentSession implements AgentSession {
private registerSubAgentToolCall(params: { private registerSubAgentToolCall(params: {
timelineItem: ToolCallTimelineItem; timelineItem: ToolCallTimelineItem;
rawItem: { [key: string]: unknown }; rawItem: { [key: string]: unknown };
childThreadIds: readonly string[];
parentCallId: string | null; parentCallId: string | null;
parentSubagentId: string | null; parentSubagentId: string | null;
}): string[] { }): string[] {
@@ -5714,16 +5828,9 @@ export class CodexAppServerAgentSession implements AgentSession {
} }
this.subAgentCallsByCallId.set(timelineItem.callId, state); this.subAgentCallsByCallId.set(timelineItem.callId, state);
const receiverThreadIds = Array.isArray(rawItem.receiverThreadIds) const childThreadIds = Array.from(new Set(params.childThreadIds)).filter(
? rawItem.receiverThreadIds.filter((value): value is string => typeof value === "string") (threadId) => threadId !== this.currentThreadId,
: []; );
const agentThreadId =
typeof rawItem.agentThreadId === "string" && rawItem.agentThreadId.length > 0
? rawItem.agentThreadId
: null;
const childThreadIds = Array.from(
new Set(agentThreadId ? [...receiverThreadIds, agentThreadId] : receiverThreadIds),
).filter((threadId) => threadId !== this.currentThreadId);
for (const receiverThreadId of childThreadIds) { for (const receiverThreadId of childThreadIds) {
const owningCallId = this.subAgentCallIdByChildThreadId.get(receiverThreadId); const owningCallId = this.subAgentCallIdByChildThreadId.get(receiverThreadId);
if (owningCallId && owningCallId !== timelineItem.callId) { if (owningCallId && owningCallId !== timelineItem.callId) {
@@ -5736,7 +5843,36 @@ export class CodexAppServerAgentSession implements AgentSession {
return childThreadIds; return childThreadIds;
} }
private handleRegisteredCollabAgentControl(
rawItem: { [key: string]: unknown },
control: CodexCollabAgentControl,
): boolean {
if (control.kind === "submission") return false;
for (const update of control.updates) {
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);
}
}
// A control rendered at start must also render its completion, even if
// Codex's final result lists fewer receivers than its started record.
const hasOwnCard = typeof rawItem.id === "string" && this.subAgentCallsByCallId.has(rawItem.id);
return (
!hasOwnCard &&
control.childThreadIds.length > 0 &&
control.childThreadIds.every((id) => this.subAgentCallIdByChildThreadId.has(id))
);
}
private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean { private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean {
const control = readCodexCollabAgentControl(rawItem);
if (control) {
return this.handleRegisteredCollabAgentControl(rawItem, control);
}
const activity = readCodexSubAgentActivity(rawItem); const activity = readCodexSubAgentActivity(rawItem);
if (!activity) { if (!activity) {
return false; return false;
@@ -6630,6 +6766,7 @@ export class CodexAppServerAgentSession implements AgentSession {
? this.registerSubAgentToolCall({ ? this.registerSubAgentToolCall({
timelineItem, timelineItem,
rawItem: parsed.item, rawItem: parsed.item,
childThreadIds: readCodexSpawnedSubAgentThreadIds(parsed.item),
parentCallId: childSubAgentCallId, parentCallId: childSubAgentCallId,
parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId), parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId),
}) })
@@ -6797,6 +6934,7 @@ export class CodexAppServerAgentSession implements AgentSession {
const registeredChildThreadIds = this.registerSubAgentToolCall({ const registeredChildThreadIds = this.registerSubAgentToolCall({
timelineItem, timelineItem,
rawItem: parsed.item, rawItem: parsed.item,
childThreadIds: readCodexSpawnedSubAgentThreadIds(parsed.item),
parentCallId: childSubAgentCallId, parentCallId: childSubAgentCallId,
parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId), parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId),
}); });
@@ -0,0 +1,463 @@
import { describe, expect, test, vi } from "vitest";
import { CodexAppServerAgentSession } from "../codex-app-server-agent.js";
import { createTestLogger } from "../../../../test-utils/test-logger.js";
import { createFakeCodexAppServer } from "./test-utils/fake-app-server.js";
function collab(tool: string, states: Record<string, string>, id = tool) {
return {
type: "collabAgentToolCall",
id,
tool,
status: "completed",
prompt: tool === "spawnAgent" ? "Reply with a short result." : null,
receiverThreadIds: Object.keys(states),
agentsStates: Object.fromEntries(
Object.entries(states).map(([threadId, status]) => [threadId, { status }]),
),
};
}
async function replay(items: unknown[]) {
const appServer = createFakeCodexAppServer({
"thread/read": (params) => ({
thread: {
turns: [
{
items: (params as { threadId: string }).threadId === "parent" ? items : [],
},
],
},
}),
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
{ sessionId: "parent" },
createTestLogger(),
async () => appServer.child,
{},
false,
false,
false,
undefined,
"history",
);
try {
await session.connect();
return await Array.fromAsync(session.streamHistory());
} finally {
await session.close();
appServer.assertNoErrors();
}
}
describe("Codex collaboration history", () => {
test("restores completed children from nullable wait and close results", async () => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("spawnAgent", { second: "pendingInit" }, "spawn-second"),
collab("wait", { first: "completed" }, "wait-first"),
collab("wait", { second: "completed" }, "wait-second"),
collab("closeAgent", { first: "completed" }, "close-first"),
collab("closeAgent", { second: "completed" }, "close-second"),
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([
{ id: "first", status: "completed" },
{ id: "second", status: "completed" },
]);
});
test.each(["completed", "interrupted", "running"])(
"keeps follow-up instructions in replay without applying the %s submission snapshot",
async (submissionStatus) => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("wait", { first: "completed" }),
{ ...collab("sendInput", { first: submissionStatus }), prompt: "Now add tests" },
]);
expect(
history.flatMap((event) =>
event.type === "timeline" && event.item.type === "tool_call"
? [{ id: event.item.callId, status: event.item.status, detail: event.item.detail }]
: [],
),
).toEqual([
{ id: "spawn-first", status: "completed", detail: expect.anything() },
{
id: "sendInput",
status: "completed",
detail: expect.objectContaining({ description: "Now add tests" }),
},
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([{ id: "first", status: "completed" }]);
},
);
test.each([
["completed", "closeAgent", "errored", "failed"],
["errored", "closeAgent", "completed", "completed"],
["completed", "wait", "running", "running"],
["completed", "closeAgent", "running", "canceled"],
])(
"replaces the earlier %s outcome after a follow-up and %s result %s",
async (initialStatus, tool, resultStatus, expected) => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("spawnAgent", { second: "pendingInit" }, "spawn-second"),
{
...collab("wait", { first: initialStatus, second: "completed" }),
status: initialStatus === "errored" ? "failed" : "completed",
},
{ ...collab("sendInput", { first: "running" }), prompt: "Do more work" },
{
...collab(tool, { first: resultStatus }),
status: resultStatus === "errored" ? "failed" : "completed",
},
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([
{ id: "first", status: expected },
{ id: "second", status: "completed" },
]);
expect(
history.flatMap((event) =>
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.callId === "spawn-first"
? [event.item.status]
: [],
),
).toEqual([expected]);
},
);
test("keeps the prior outcome when a follow-up submission fails", async () => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("wait", { first: "completed" }),
{ ...collab("sendInput", { first: "running" }), status: "failed", prompt: "Do more work" },
collab("wait", { first: "running" }, "stale-wait"),
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [event.event.status]
: [],
),
).toEqual(["completed"]);
});
test("settles a close card without a registered spawn", async () => {
const history = await replay([collab("closeAgent", { external: "running" })]);
expect(
history.flatMap((event) =>
event.type === "timeline" && event.item.type === "tool_call"
? [{ id: event.item.callId, status: event.item.status }]
: [],
),
).toEqual([{ id: "closeAgent", status: "canceled" }]);
});
test.each(["notFound", "errored"])(
"preserves completion and shows a failed resume reporting %s",
async (status) => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("wait", { first: "completed" }),
collab("closeAgent", { first: "completed" }),
{ ...collab("resumeAgent", { first: status }), status: "failed" },
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [event.event.status]
: [],
),
).toEqual(["completed"]);
expect(
history.flatMap((event) =>
event.type === "timeline" && event.item.type === "tool_call"
? [{ id: event.item.callId, status: event.item.status }]
: [],
),
).toEqual([
{ id: "spawn-first", status: "completed" },
{ id: "resumeAgent", status: "failed" },
]);
},
);
test("stops counting a child closed while running", async () => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("closeAgent", { first: "running" }),
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([{ id: "first", status: "canceled" }]);
});
test.each(["wait", "closeAgent"])(
"restores an errored child as failed after %s",
async (tool) => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
{ ...collab(tool, { first: "errored" }), status: "failed" },
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([{ id: "first", status: "failed" }]);
expect(
history.flatMap((event) =>
event.type === "timeline" && event.item.type === "tool_call"
? [{ id: event.item.callId, status: event.item.status, error: event.item.error }]
: [],
),
).toEqual([{ id: "spawn-first", status: "failed", error: { message: "Sub-agent failed" } }]);
},
);
test.each(["wait", "closeAgent"])(
"settles a live errored child as failed after %s",
async (tool) => {
const appServer = createFakeCodexAppServer();
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
null,
createTestLogger(),
async () => appServer.child,
);
const statuses = new Map<string, string>();
const cards = new Map<string, { status: string; error: unknown }>();
session.subscribe((event) => {
if (event.type === "provider_subagent" && event.event.type === "upsert") {
statuses.set(event.event.id, event.event.status);
}
if (event.type === "timeline" && event.item.type === "tool_call") {
cards.set(event.item.callId, { status: event.item.status, error: event.item.error });
}
});
function notify(method: string, item: ReturnType<typeof collab>) {
appServer.child.stdout.write(
`${JSON.stringify({ method, params: { threadId: "thread-1", item } })}\n`,
);
}
try {
const result = session.run("Delegate the work.");
await appServer.waitForTurnStart();
notify("item/completed", collab("spawnAgent", { first: "pendingInit" }, "spawn-first"));
notify("item/started", {
...collab(tool, { first: "errored" }),
status: "inProgress",
agentsStates: {},
});
notify("item/completed", { ...collab(tool, { first: "errored" }), status: "failed" });
appServer.completeTurn();
await result;
expect([...statuses]).toEqual([["first", "failed"]]);
expect([...cards]).toEqual([
["spawn-first", { status: "failed", error: { message: "Sub-agent failed" } }],
]);
} finally {
await session.close();
appServer.assertNoErrors();
}
},
);
test("preserves completion when a close carries an older running snapshot", async () => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("wait", { first: "completed" }),
collab("closeAgent", { first: "running" }),
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([{ id: "first", status: "completed" }]);
});
test.each([
["shutdown", "canceled"],
["notFound", "failed"],
])("restores a %s child as %s and preserves its running sibling", async (status, expected) => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
collab("spawnAgent", { second: "pendingInit" }, "spawn-second"),
collab("closeAgent", { first: status }),
]);
expect(
history.flatMap((event) =>
event.type === "provider_subagent" && event.event.type === "upsert"
? [{ id: event.event.id, status: event.event.status }]
: [],
),
).toEqual([
{ id: "first", status: expected },
{ id: "second", status: "running" },
]);
});
test("settles a rendered live wait when its completion lists only the finished receiver", async () => {
const appServer = createFakeCodexAppServer();
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
null,
createTestLogger(),
async () => appServer.child,
);
const cards = new Map<string, string>();
const statuses = new Map<string, string>();
session.subscribe((event) => {
if (event.type === "timeline" && event.item.type === "tool_call") {
cards.set(event.item.callId, event.item.status);
}
if (event.type === "provider_subagent" && event.event.type === "upsert") {
statuses.set(event.event.id, event.event.status);
}
});
function notify(method: string, item: ReturnType<typeof collab>) {
appServer.child.stdout.write(
`${JSON.stringify({ method, params: { threadId: "thread-1", item } })}\n`,
);
}
try {
const result = session.run("Wait for the children.");
await appServer.waitForTurnStart();
notify("item/completed", collab("spawnAgent", { first: "pendingInit" }, "spawn-first"));
notify("item/started", {
...collab("wait", { first: "running", second: "running" }, "mixed-wait"),
status: "inProgress",
agentsStates: {},
});
await vi.waitFor(() => expect(cards.get("mixed-wait")).toBe("running"));
notify("item/completed", collab("wait", { first: "completed" }, "mixed-wait"));
notify("item/completed", collab("wait", { second: "completed" }, "wait-second"));
appServer.completeTurn();
await result;
expect([...statuses]).toEqual([["first", "completed"]]);
expect([...cards]).toEqual([
["spawn-first", "completed"],
["mixed-wait", "completed"],
["wait-second", "completed"],
]);
} finally {
await session.close();
appServer.assertNoErrors();
}
});
test.each(["completed", "interrupted", "running"])(
"keeps a live follow-up running despite a late %s sendInput snapshot",
async (submissionStatus) => {
const appServer = createFakeCodexAppServer();
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: process.cwd(), modeId: "full-access" },
null,
createTestLogger(),
async () => appServer.child,
);
const statuses = new Map<string, string>();
const timelineCards = new Map<string, { status: string; detail: unknown }>();
let parentMessage = "";
session.subscribe((event) => {
if (event.type === "provider_subagent" && event.event.type === "upsert") {
statuses.set(event.event.id, event.event.status);
}
if (event.type === "timeline" && event.item.type === "tool_call") {
timelineCards.set(event.item.callId, {
status: event.item.status,
detail: event.item.detail,
});
}
if (event.type === "timeline" && event.item.type === "assistant_message") {
parentMessage = event.item.text;
}
});
function complete(tool: string, states: Record<string, string>, id = tool) {
const item = {
...collab(tool, states, id),
...(tool === "sendInput" ? { prompt: "Now add tests" } : {}),
};
if (tool !== "spawnAgent") {
appServer.child.stdout.write(
`${JSON.stringify({
method: "item/started",
params: {
threadId: "thread-1",
item: { ...item, status: "inProgress", agentsStates: {} },
},
})}\n`,
);
}
appServer.child.stdout.write(
`${JSON.stringify({
method: "item/completed",
params: { threadId: "thread-1", item },
})}\n`,
);
}
try {
const result = session.run("Delegate the work.");
await appServer.waitForTurnStart();
complete("spawnAgent", { first: "pendingInit" }, "spawn-first");
complete("spawnAgent", { second: "pendingInit" }, "spawn-second");
complete("wait", { first: "completed", second: "running" });
await vi.waitFor(() => {
expect(statuses.get("first")).toBe("completed");
expect(statuses.get("second")).toBe("running");
});
complete("closeAgent", { first: "shutdown" });
complete("wait", { first: "running" }, "stale-wait");
appServer.says({ threadId: "thread-1", text: "Still waiting." });
appServer.completeTurn();
await result;
expect(statuses.get("first")).toBe("completed");
appServer.startsTurn({ threadId: "first", turnId: "follow-up" });
await vi.waitFor(() => expect(statuses.get("first")).toBe("running"));
complete("sendInput", { first: submissionStatus });
appServer.says({ threadId: "thread-1", text: "Follow-up submitted." });
await vi.waitFor(() => expect(parentMessage).toBe("Follow-up submitted."));
expect(statuses.get("first")).toBe("running");
expect([...timelineCards.keys()]).toEqual(["spawn-first", "spawn-second", "sendInput"]);
expect(timelineCards.get("sendInput")).toEqual({
status: "completed",
detail: expect.objectContaining({ description: "Now add tests" }),
});
// Only the child's completion settles the follow-up.
appServer.completeTurn({ threadId: "first" });
await vi.waitFor(() => expect(statuses.get("first")).toBe("completed"));
} finally {
await session.close();
appServer.assertNoErrors();
}
},
);
});
@@ -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", () => { it("maps collabAgentToolCall into canonical sub-agent detail", () => {
const item = mapCodexToolCallFromThreadItem({ const item = mapCodexToolCallFromThreadItem({
type: "collabAgentToolCall", type: "collabAgentToolCall",
@@ -186,7 +186,7 @@ const CodexCollabAgentToolCallItemSchema = z
id: z.string().min(1), id: z.string().min(1),
status: z.string().optional(), status: z.string().optional(),
error: z.unknown().optional(), error: z.unknown().optional(),
prompt: z.string().optional(), prompt: z.string().nullish(),
tool: z.string().optional(), tool: z.string().optional(),
receiverThreadIds: z.array(z.string()).optional(), receiverThreadIds: z.array(z.string()).optional(),
agentsStates: z.record(z.string(), z.unknown()).optional(), agentsStates: z.record(z.string(), z.unknown()).optional(),
@@ -545,12 +545,18 @@ function readStatus(value: unknown): string | undefined {
return typeof value.status === "string" ? value.status : undefined; return typeof value.status === "string" ? value.status : undefined;
} }
function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] { export function normalizeCollabAgentChildStatus(
status: string,
tool?: string,
): ToolCallTimelineItem["status"] {
const normalized = status.trim().toLowerCase(); const normalized = status.trim().toLowerCase();
switch (normalized) { switch (normalized) {
case "error": case "error":
case "errored": case "errored":
return "running"; return tool === "wait" || tool === "closeAgent" ? "failed" : "running";
case "running":
case "pendinginit":
return tool === "closeAgent" ? "canceled" : "running";
case "shutdown": case "shutdown":
return "canceled"; return "canceled";
case "notfound": case "notfound":
@@ -572,10 +578,15 @@ function resolveCollabAgentStatus(
return "failed"; return "failed";
} }
// Sending input completes the submission, not the child's new turn.
if (item.tool === "sendInput") {
return parentStatus;
}
const childStatuses = Object.values(item.agentsStates ?? {}) const childStatuses = Object.values(item.agentsStates ?? {})
.map(readStatus) .map(readStatus)
.filter((status): status is string => typeof status === "string" && status.trim().length > 0) .filter((status): status is string => typeof status === "string" && status.trim().length > 0)
.map(normalizeCollabAgentChildStatus); .map((status) => normalizeCollabAgentChildStatus(status, item.tool));
if (childStatuses.some((status) => status === "failed")) { if (childStatuses.some((status) => status === "failed")) {
return "failed"; return "failed";
@@ -581,6 +581,9 @@ test("Git refresh, command tracing and 10,000 ignored alias events do no synchro
let fsCalls = 0; let fsCalls = 0;
const hook = createHook({ init(_id, type) { if (type.startsWith('FSREQ')) fsCalls++; } }); const hook = createHook({ init(_id, type) { if (type.startsWith('FSREQ')) fsCalls++; } });
await tick(); await tick();
// Do not charge garbage left by repository setup to the event filter.
// Allocations and GC caused by the measured callback remain in its budget.
global.gc();
const { blockingMs } = await measureMainThreadWork('Git filter main-thread work', async () => { const { blockingMs } = await measureMainThreadWork('Git filter main-thread work', async () => {
await marker('FILTER_START'); await marker('FILTER_START');
hook.enable(); hook.enable();
@@ -614,6 +617,7 @@ test("Git refresh, command tracing and 10,000 ignored alias events do no synchro
const { stdout, stderr } = await promisify(execFile)( const { stdout, stderr } = await promisify(execFile)(
process.execPath, process.execPath,
[ [
"--expose-gc",
"--import", "--import",
"tsx", "tsx",
...(tracing ? ["--trace-sync-io"] : []), ...(tracing ? ["--trace-sync-io"] : []),