Compare commits

...
Author SHA1 Message Date
paseo-bot[bot] c299598aea 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 04:54:17 +00:00
paseo-bot[bot] f110243b4c 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 04:45:17 +00:00
1bb0798857 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 04:31:05 +00:00
popstk 198d8a5c68 fix(codex): reconcile nullable collaboration results with spawned agents 2026-10-30 04:31:05 +00:00
4 changed files with 320 additions and 2 deletions
@@ -58,6 +58,7 @@ import {
mapCodexToolCallEnvelope,
mapCodexToolCallFromThreadItem,
normalizeCommandExecutionCommand,
normalizeCollabAgentChildStatus,
splitCodexMcpToolResultImages,
} from "./codex/tool-call-mapper.js";
import {
@@ -1860,6 +1861,65 @@ function readCodexSubAgentActivity(item: unknown): CodexSubAgentActivity | null
};
}
interface CodexCollabAgentUpdate {
threadId: string;
status: ToolCallTimelineItem["status"];
preservesTerminal: boolean;
reopen: boolean;
}
interface CodexCollabAgentControl {
childThreadIds: string[];
updates: CodexCollabAgentUpdate[];
}
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 updates = Object.entries(toObjectRecord(record.agentsStates) ?? {}).flatMap(
([threadId, value]): CodexCollabAgentUpdate[] => {
const state = toObjectRecord(value);
if (typeof state?.status !== "string") return [];
const status = normalizeCollabAgentChildStatus(state.status);
return [
{
threadId,
status: record.tool === "closeAgent" && status === "running" ? "canceled" : status,
preservesTerminal: state.status === "shutdown" || record.tool === "closeAgent",
reopen: record.tool === "sendInput" && record.status === "completed",
},
];
},
);
const receivers = Array.isArray(record.receiverThreadIds)
? record.receiverThreadIds.filter((id): id is string => typeof id === "string")
: [];
return {
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" && !update.reopen))
) {
return previous;
}
return update.status;
}
function shouldIgnoreMirroredLifecycleItem(source: "item" | "codex_event", item: unknown): boolean {
return source === "codex_event" && !readCodexSubAgentActivity(item);
}
@@ -2093,6 +2153,30 @@ async function requestCodexThreadHistory(
return CodexThreadReadResponseSchema.parse(response);
}
function updateHistoricalCollabAgentCalls(
timeline: PersistedTimelineEntry[],
timelineIndexByThreadId: Map<string, number>,
item: unknown,
): boolean {
const control = readCodexCollabAgentControl(item);
if (!control) return false;
for (const update of control.updates) {
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 };
}
return (
control.childThreadIds.length > 0 &&
control.childThreadIds.every((id) => timelineIndexByThreadId.has(id))
);
}
async function loadCodexThreadHistoryTimeline(params: {
threadId: string;
cwd: string | null;
@@ -2103,6 +2187,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 +5848,23 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private handleRegisteredSubAgentActivity(rawItem: { [key: string]: unknown }): boolean {
const control = readCodexCollabAgentControl(rawItem);
if (control) {
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, { reopen: update.reopen });
}
}
return (
control.childThreadIds.length > 0 &&
control.childThreadIds.every((id) => this.subAgentCallIdByChildThreadId.has(id))
);
}
const activity = readCodexSubAgentActivity(rawItem);
if (!activity) {
return false;
@@ -0,0 +1,200 @@
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("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("stops counting an errored child whose close record reports failed", async () => {
const history = await replay([
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
{ ...collab("closeAgent", { 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: "canceled" }]);
});
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("keeps completed live children settled until an explicit follow-up", async () => {
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 timelineCallIds = new Set<string>();
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") {
timelineCallIds.add(event.item.callId);
}
});
function complete(tool: string, states: Record<string, string>, id = tool) {
if (tool !== "spawnAgent") {
appServer.child.stdout.write(
`${JSON.stringify({
method: "item/started",
params: {
threadId: "thread-1",
item: { ...collab(tool, states, id), status: "inProgress", agentsStates: {} },
},
})}\n`,
);
}
appServer.child.stdout.write(
`${JSON.stringify({
method: "item/completed",
params: { threadId: "thread-1", item: collab(tool, states, id) },
})}\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");
complete("sendInput", { first: "running" });
await vi.waitFor(() => expect(statuses.get("first")).toBe("running"));
expect([...timelineCallIds]).toEqual(["spawn-first", "spawn-second"]);
} 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", () => {
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(),
@@ -545,7 +545,7 @@ function readStatus(value: unknown): string | undefined {
return typeof value.status === "string" ? value.status : undefined;
}
function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] {
export function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] {
const normalized = status.trim().toLowerCase();
switch (normalized) {
case "error":