mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 13:13:25 +00:00
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>
This commit is contained in:
co-authored by
geeknonerd
altenli
parent
dc23fe6545
commit
cbfb29c5fa
@@ -5004,120 +5004,6 @@ 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 = {
|
||||
|
||||
@@ -1888,7 +1888,12 @@ function readCodexCollabAgentUpdates(item: unknown): CodexCollabAgentUpdate[] {
|
||||
? [
|
||||
{
|
||||
threadId,
|
||||
status: mapped.status,
|
||||
status:
|
||||
record.tool === "closeAgent" &&
|
||||
record.status === "completed" &&
|
||||
mapped.status === "running"
|
||||
? "canceled"
|
||||
: mapped.status,
|
||||
shutdown: state.status === "shutdown",
|
||||
reopen: record.tool === "sendInput" && record.status === "completed",
|
||||
},
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
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.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>();
|
||||
session.subscribe((event) => {
|
||||
if (event.type === "provider_subagent" && event.event.type === "upsert") {
|
||||
statuses.set(event.event.id, event.event.status);
|
||||
}
|
||||
});
|
||||
function complete(tool: string, states: Record<string, string>, id = tool) {
|
||||
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"));
|
||||
} finally {
|
||||
await session.close();
|
||||
appServer.assertNoErrors();
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user