diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 4754942133..cb8cb2684b 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -5004,120 +5004,6 @@ describe("Codex app-server provider", () => { }); }); - function collaboration(tool: string, states: Record, 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, 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 = { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 2095f651de..9ff1a5b282 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -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", }, diff --git a/packages/server/src/server/agent/providers/codex/collab-replay.test.ts b/packages/server/src/server/agent/providers/codex/collab-replay.test.ts new file mode 100644 index 0000000000..4b9fd1a04d --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/collab-replay.test.ts @@ -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, 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(); + 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, 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(); + } + }); +});