From cbfb29c5fa4781c4b8ff9749417f5683fc6250f7 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:30:21 +0000 Subject: [PATCH] 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 --- .../providers/codex-app-server-agent.test.ts | 114 ------------- .../agent/providers/codex-app-server-agent.ts | 7 +- .../providers/codex/collab-replay.test.ts | 155 ++++++++++++++++++ 3 files changed, 161 insertions(+), 115 deletions(-) create mode 100644 packages/server/src/server/agent/providers/codex/collab-replay.test.ts 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(); + } + }); +});