mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 13:13:25 +00:00
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.
This commit is contained in:
@@ -1864,9 +1864,13 @@ interface CodexCollabAgentUpdate {
|
||||
preservesTerminal: boolean;
|
||||
}
|
||||
|
||||
interface CodexCollabAgentControl {
|
||||
childThreadIds: string[];
|
||||
updates: CodexCollabAgentUpdate[];
|
||||
type CodexCollabAgentControl =
|
||||
| { kind: "submission"; childThreadIds: string[] }
|
||||
| { kind: "outcome"; childThreadIds: string[]; updates: CodexCollabAgentUpdate[] };
|
||||
|
||||
interface HistoricalCodexSubAgent {
|
||||
timelineIndex: number;
|
||||
pendingFollowUp: boolean;
|
||||
}
|
||||
|
||||
function readCodexCollabAgentControl(item: unknown): CodexCollabAgentControl | null {
|
||||
@@ -1875,32 +1879,37 @@ function readCodexCollabAgentControl(item: unknown): CodexCollabAgentControl | n
|
||||
!record ||
|
||||
normalizeCodexThreadItemType(typeof record.type === "string" ? record.type : undefined) !==
|
||||
"collabAgentToolCall" ||
|
||||
record.tool === "spawnAgent" ||
|
||||
record.tool === "sendInput"
|
||||
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;
|
||||
}
|
||||
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 =
|
||||
state.status === "errored" ? "failed" : normalizeCollabAgentChildStatus(state.status);
|
||||
const status = normalizeCollabAgentChildStatus(state.status, tool);
|
||||
return [
|
||||
{
|
||||
threadId,
|
||||
status: record.tool === "closeAgent" && status === "running" ? "canceled" : status,
|
||||
status,
|
||||
preservesTerminal: state.status === "shutdown" || record.tool === "closeAgent",
|
||||
},
|
||||
];
|
||||
},
|
||||
);
|
||||
const receivers = Array.isArray(record.receiverThreadIds)
|
||||
? record.receiverThreadIds.filter((id): id is string => typeof id === "string")
|
||||
: [];
|
||||
return {
|
||||
kind: "outcome",
|
||||
childThreadIds: [...new Set([...receivers, ...updates.map((update) => update.threadId)])],
|
||||
updates,
|
||||
};
|
||||
@@ -2124,25 +2133,35 @@ async function threadItemToTimelineEntries(
|
||||
|
||||
function updateHistoricalCollabAgentCalls(
|
||||
timeline: PersistedTimelineEntry[],
|
||||
timelineIndexByThreadId: Map<string, number>,
|
||||
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 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 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[existingIndex] = { ...existing, item: toolCall };
|
||||
timeline[child.timelineIndex] = { ...existing, item: toolCall };
|
||||
}
|
||||
return (
|
||||
control.childThreadIds.length > 0 &&
|
||||
control.childThreadIds.every((id) => timelineIndexByThreadId.has(id))
|
||||
control.childThreadIds.every((id) => subAgentsByThreadId.has(id))
|
||||
);
|
||||
}
|
||||
|
||||
@@ -2152,17 +2171,17 @@ async function loadCodexThreadHistoryTimeline(params: {
|
||||
cwd: string | null;
|
||||
}): Promise<CodexThreadHistoryProjection> {
|
||||
const timeline: PersistedTimelineEntry[] = [];
|
||||
const subAgentTimelineIndexByThreadId = new Map<string, number>();
|
||||
const subAgentsByThreadId = new Map<string, HistoricalCodexSubAgent>();
|
||||
try {
|
||||
await projectCodexThreadHistory(params, { timeline, subAgentTimelineIndexByThreadId });
|
||||
await projectCodexThreadHistory(params, { timeline, subAgentsByThreadId });
|
||||
} catch (error) {
|
||||
if (!(error instanceof CodexThreadHistoryTooLargeError)) throw error;
|
||||
// The session stays usable; the row says why the history stops here.
|
||||
timeline.push({ item: { type: "notification", level: "warning", message: error.message } });
|
||||
}
|
||||
const subAgentRoutes = Array.from(subAgentTimelineIndexByThreadId.entries()).flatMap(
|
||||
([childThreadId, timelineIndex]): PersistedSubAgentRoute[] => {
|
||||
const item = timeline[timelineIndex]?.item;
|
||||
const subAgentRoutes = Array.from(subAgentsByThreadId.entries()).flatMap(
|
||||
([childThreadId, child]): PersistedSubAgentRoute[] => {
|
||||
const item = timeline[child.timelineIndex]?.item;
|
||||
return item?.type === "tool_call" && item.detail.type === "sub_agent"
|
||||
? [{ childThreadId, toolCall: item }]
|
||||
: [];
|
||||
@@ -2175,20 +2194,20 @@ async function projectCodexThreadHistory(
|
||||
params: { client: CodexAppServerClientLike; threadId: string; cwd: string | null },
|
||||
projection: {
|
||||
timeline: PersistedTimelineEntry[];
|
||||
subAgentTimelineIndexByThreadId: Map<string, number>;
|
||||
subAgentsByThreadId: Map<string, HistoricalCodexSubAgent>;
|
||||
},
|
||||
): Promise<void> {
|
||||
const { timeline, subAgentTimelineIndexByThreadId } = projection;
|
||||
const { timeline, subAgentsByThreadId } = projection;
|
||||
for await (const row of readCodexThreadHistory(params.client, params.threadId)) {
|
||||
const { item } = row;
|
||||
if (updateHistoricalCollabAgentCalls(timeline, subAgentTimelineIndexByThreadId, item)) {
|
||||
if (updateHistoricalCollabAgentCalls(timeline, subAgentsByThreadId, item)) {
|
||||
continue;
|
||||
}
|
||||
const historicalSubAgentActivity = readCodexSubAgentActivity(item);
|
||||
if (historicalSubAgentActivity) {
|
||||
const existingIndex = subAgentTimelineIndexByThreadId.get(
|
||||
const existingIndex = subAgentsByThreadId.get(
|
||||
historicalSubAgentActivity.agentThreadId,
|
||||
);
|
||||
)?.timelineIndex;
|
||||
if (existingIndex !== undefined) {
|
||||
const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd });
|
||||
updateHistoricalSubAgentActivity(
|
||||
@@ -2219,7 +2238,10 @@ async function projectCodexThreadHistory(
|
||||
: {}),
|
||||
});
|
||||
for (const childThreadId of readCodexHistoricalSubAgentThreadIds(item)) {
|
||||
subAgentTimelineIndexByThreadId.set(childThreadId, timeline.length - 1);
|
||||
subAgentsByThreadId.set(childThreadId, {
|
||||
timelineIndex: timeline.length - 1,
|
||||
pendingFollowUp: false,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5829,6 +5851,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
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;
|
||||
|
||||
@@ -104,6 +104,76 @@ describe("Codex collaboration history", () => {
|
||||
},
|
||||
);
|
||||
|
||||
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("stops counting a child closed while running", async () => {
|
||||
const history = await replay([
|
||||
collab("spawnAgent", { first: "pendingInit" }, "spawn-first"),
|
||||
|
||||
@@ -545,12 +545,18 @@ function readStatus(value: unknown): string | undefined {
|
||||
return typeof value.status === "string" ? value.status : undefined;
|
||||
}
|
||||
|
||||
export function normalizeCollabAgentChildStatus(status: string): ToolCallTimelineItem["status"] {
|
||||
export function normalizeCollabAgentChildStatus(
|
||||
status: string,
|
||||
tool?: string,
|
||||
): ToolCallTimelineItem["status"] {
|
||||
const normalized = status.trim().toLowerCase();
|
||||
switch (normalized) {
|
||||
case "error":
|
||||
case "errored":
|
||||
return "running";
|
||||
return tool === "wait" || tool === "closeAgent" ? "failed" : "running";
|
||||
case "running":
|
||||
case "pendinginit":
|
||||
return tool === "closeAgent" ? "canceled" : "running";
|
||||
case "shutdown":
|
||||
return "canceled";
|
||||
case "notfound":
|
||||
@@ -580,7 +586,7 @@ function resolveCollabAgentStatus(
|
||||
const childStatuses = Object.values(item.agentsStates ?? {})
|
||||
.map(readStatus)
|
||||
.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")) {
|
||||
return "failed";
|
||||
|
||||
Reference in New Issue
Block a user