From 4285fd91f3e68c1cfc945023ff12373a6786387c Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 26 Oct 2026 15:51:38 +0100 Subject: [PATCH] feat: replay historical child outcomes --- readme-dev.md | 10 +- src/CodexAcpClient.ts | 21 +- src/CodexAcpServer.ts | 14 +- .../data/historical-outcome-v1-completed.json | 13 ++ .../data/historical-outcome-v1-failed.json | 17 ++ .../historical-outcome-v1-interrupted.json | 13 ++ .../data/historical-outcome-v2-completed.json | 21 ++ .../data/historical-outcome-v2-failed.json | 29 +++ .../historical-outcome-v2-interrupted.json | 21 ++ .../subagent-replay-v2-completed-tool.json | 8 + .../data/subagent-replay-v2-transcript.json | 8 + .../standard-subagent-replay.test.ts | 6 +- .../subagent-historical-outcomes.test.ts | 221 ++++++++++++++++++ .../standard-subagent-live.test.ts | 4 + src/subagents/StandardSubagentEventRouter.ts | 38 ++- 15 files changed, 422 insertions(+), 22 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v1-completed.json create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v1-failed.json create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v1-interrupted.json create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v2-completed.json create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v2-failed.json create mode 100644 src/__tests__/CodexACPAgent/data/historical-outcome-v2-interrupted.json create mode 100644 src/__tests__/CodexACPAgent/subagent-historical-outcomes.test.ts diff --git a/readme-dev.md b/readme-dev.md index 5b7ae54e5..da61a1951 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -276,9 +276,13 @@ partial representation are omitted rather than replayed over newer content. Agen rendered as human input. Missing, omitted, or incomplete child history produces a parent advisory; an empty transcript alone does not invent a gap. -Historical work-state outcomes remain deferred. Replayed content alone confirms neither current work -nor controls. Rejoining a loaded child still requires current provider parentage; copied fork history -can be displayed but cannot acquire the original runtime's subscriptions. +Replay reports the latest recorded child turn's known success, cancellation, or failure as historical +`idle` before the response. A missing or still-running turn invents no outcome. Legacy history reuses +its turn data; paginated history reads the latest turn summary separately, retaining available text +and reporting a gap if that summary is unavailable. Recorded outcomes do not enter the live foreground +cache, withdraw current interactions, confirm runtime ownership, or grant controls. +Rejoining a loaded child still requires current provider parentage; copied fork history can be displayed +but cannot acquire the original runtime's subscriptions. After replay, current child work/configuration snapshots follow the actual response write, including whole JSON-RPC batches. Parent setup operations serialize without waiting on each other's batch delivery. diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 6f88c3286..1197d1c80 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -54,6 +54,7 @@ import type { ThreadSourceKind, ThreadItem, ThreadItemEntry, + Turn, TurnCompletedNotification, TurnSteerResponse, UserInput, @@ -85,7 +86,8 @@ import { import {findSupportedEffort} from "./ModelConfigOption"; import {disableUnsupportedSkills, withoutUnsupportedSkills} from "./UnsupportedSkills"; export type {SessionMetadata, SessionMetadataWithThread} from "./SessionMetadata"; -export type RecordedSubagentItem = {item: ThreadItem; completed: boolean; turnId?: string}; +/** A null turn marks unavailable outcome metadata, not a failed turn. */ +export type RecordedSubagentEntry = {item: ThreadItem; completed: boolean; turnId?: string} | {turn: Turn | null}; /** * The slice of `thread/resume` the session layer consumes, plus whether Codex @@ -828,8 +830,8 @@ export class CodexAcpClient { return null; } - /** Lazy item reads let the caller restore verified live delivery before reading snapshots. */ - async readSubagentHistory(threadId: string): Promise<{thread: Thread; items: AsyncIterable}> { + /** Lazy history reads let the caller restore verified live delivery before reading snapshots. */ + async readSubagentHistory(threadId: string): Promise<{thread: Thread; items: AsyncIterable}> { const {thread} = await this.codexClient.threadRead({threadId}); if (thread.id !== threadId) throw new Error("Subagent history metadata names a different thread"); return { @@ -1808,20 +1810,29 @@ async function* oneItemPage(items: T[]): AsyncGenerator { if (items.length > 0) yield items; } -async function* legacySubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator { +async function* legacySubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator { const {thread} = await client.threadRead({threadId, includeTurns: true}); if (thread.id !== threadId) throw new Error("Subagent history names a different thread"); yield thread.turns.flatMap(turn => turn.items.map(item => ({ item, turnId: turn.id, completed: recordedItemCompleted(item, turn.status !== "inProgress"), }))); + const turn = thread.turns.at(-1); + if (turn) yield [{turn}]; } -async function* paginatedSubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator { +async function* paginatedSubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator { for await (const page of client.threadItemEntryPages(threadId)) { yield page.map(({item, turnId, completedAtMs}) => ({ item, turnId, completed: recordedItemCompleted(item, completedAtMs !== null), })); } + try { + const {data} = await client.threadTurnsList({threadId, limit: 1, sortDirection: "desc", itemsView: "notLoaded"}); + if (data[0]) yield [{turn: data[0]}]; + } catch (error) { + logger.error(`Failed to read recorded outcome for subagent ${threadId}`, error); + yield [{turn: null}]; + } } function recordedItemCompleted(item: ThreadItem, completed: boolean): boolean { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index df9dee8f0..2ae7d5fe5 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -15,7 +15,7 @@ import { OPENAI_PROVIDER_ID, type SessionMetadata, type SessionMetadataWithThread, - type RecordedSubagentItem, + type RecordedSubagentEntry, type UrlElicitationRequester } from "./CodexAcpClient"; import { @@ -3702,7 +3702,7 @@ export class CodexAcpServer { /** Recorded conversation data is independent of verified runtime ownership and current work. */ private async streamStandardHistory( - sessionId: string, pages: AsyncIterable, state: SessionState, isOpen: () => boolean, + sessionId: string, pages: AsyncIterable, state: SessionState, isOpen: () => boolean, threadId = sessionId, ancestry = new Set([threadId]), replayed = new Set(), gaps = new Set(), ): Promise { const router = state.subagents; @@ -3744,8 +3744,14 @@ export class CodexAcpServer { }; let unresolved: Array = []; for await (const items of pages) { - for (const {item, completed, turnId} of items) { + for (const entry of items) { if (!isOpen()) throw new SessionClosedDuringLoadError(); + if ("turn" in entry) { + if (entry.turn === null) gaps.add(threadId); + else if (threadId !== state.sessionId) await router.replayTurnOutcome(threadId, entry.turn); + continue; + } + const {item, completed, turnId} = entry; if (item.type === "subAgentActivity") { if (isRootAgentPath(item.agentPath)) continue; if (ancestry.has(item.agentThreadId)) { gaps.add(item.agentThreadId); continue; } @@ -5586,7 +5592,7 @@ async function* itemsOfEntries(pages: AsyncIterable): AsyncGe } } -async function* standardHistoryItems(pages: AsyncIterable): AsyncGenerator { +async function* standardHistoryItems(pages: AsyncIterable): AsyncGenerator { for await (const page of pages) { yield page.map(item => ({item, completed: !("status" in item) || item.status !== "inProgress"})); } diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v1-completed.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-completed.json new file mode 100644 index 000000000..23709c80a --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-completed.json @@ -0,0 +1,13 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "stopReason": "end_turn", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v1-failed.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-failed.json new file mode 100644 index 000000000..2b74510d2 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-failed.json @@ -0,0 +1,17 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "error": { + "code": -32603, + "message": "Recorded failure" + }, + "stopReason": "error", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v1-interrupted.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-interrupted.json new file mode 100644 index 000000000..8252fe64e --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v1-interrupted.json @@ -0,0 +1,13 @@ +[ + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "stopReason": "cancelled", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v2-completed.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-completed.json new file mode 100644 index 000000000..dde0905a5 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-completed.json @@ -0,0 +1,21 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "stopReason": "end_turn", + "state": "idle", + "sessionUpdate": "state_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "stopReason": "end_turn", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v2-failed.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-failed.json new file mode 100644 index 000000000..8d8ad7a57 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-failed.json @@ -0,0 +1,29 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "error": { + "code": -32603, + "message": "Recorded failure" + }, + "stopReason": "error", + "state": "idle", + "sessionUpdate": "state_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "error": { + "code": -32603, + "message": "Recorded failure" + }, + "stopReason": "error", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/historical-outcome-v2-interrupted.json b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-interrupted.json new file mode 100644 index 000000000..bd34f3433 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/historical-outcome-v2-interrupted.json @@ -0,0 +1,21 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "stopReason": "cancelled", + "state": "idle", + "sessionUpdate": "state_update" + } + }, + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "state": { + "stopReason": "cancelled", + "state": "idle" + }, + "sessionUpdate": "subagent_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json index c241adb64..b1cec2bb0 100644 --- a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json @@ -61,6 +61,14 @@ "sessionUpdate": "agent_message_chunk" } }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "stopReason": "end_turn", + "state": "idle", + "sessionUpdate": "state_update" + } + }, { "sessionId": "thread-1:subagent:worker", "update": { diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json index 2f892592a..214b0cddd 100644 --- a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json @@ -17,5 +17,13 @@ }, "sessionUpdate": "agent_message_chunk" } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "stopReason": "end_turn", + "state": "idle", + "sessionUpdate": "state_update" + } } ] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts index 3d31132d0..8c52c57de 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts @@ -145,7 +145,7 @@ describe.each([1, 2] as const)("ACP v%s recorded child replay", version => { expect(gap()).toEqual([]); }); - it("reads paginated child items in order without requesting legacy turns or work outcomes", async () => { + it("retains ordered paginated items without legacy turns when the outcome is unavailable", async () => { await connect(); const {reads} = history(h, {[sessionId]: {items: [spawn()]}, [worker]: {items: [], paginated: true}}); const entries = [message("first-page", "First page"), message("last-page", "Last page")].map(item => @@ -155,12 +155,12 @@ describe.each([1, 2] as const)("ACP v%s recorded child replay", version => { if (sortDirection === "desc") return {data: [entries[1]], nextCursor: null, backwardsCursor: null}; return {data: cursor ? [entries[1]] : [entries[0]], nextCursor: cursor ? null : "second", backwardsCursor: null}; }); - const outcomeRead = vi.fn(async () => { throw new Error("Outcomes are not part of this slice"); }); + const outcomeRead = vi.fn(async () => { throw new Error("Outcome metadata unavailable"); }); h.setCodexResponse("thread/turns/list", outcomeRead); await load(); await flush(h); expect(reads.some(read => read.id === worker && read.includeTurns)).toBe(false); - expect(outcomeRead).not.toHaveBeenCalled(); + expect(outcomeRead).toHaveBeenCalledExactlyOnceWith({threadId: worker, limit: 1, sortDirection: "desc", itemsView: "notLoaded"}); const ids = childUpdates().flatMap(entry => entry.update.sessionUpdate === "agent_message_chunk" ? [entry.update.messageId] : []); expect(ids).toEqual(["first-page", "last-page"]); }); diff --git a/src/__tests__/CodexACPAgent/subagent-historical-outcomes.test.ts b/src/__tests__/CodexACPAgent/subagent-historical-outcomes.test.ts new file mode 100644 index 000000000..5afc8afc3 --- /dev/null +++ b/src/__tests__/CodexACPAgent/subagent-historical-outcomes.test.ts @@ -0,0 +1,221 @@ +import {afterEach, describe, expect, it, vi} from "vitest"; +import type {StateUpdate} from "@agentclientprotocol/sdk/experimental/v2/unstable"; +import type {ServerNotification} from "../../app-server"; +import type {ThreadItem, Turn, TurnStatus} from "../../app-server/v2"; +import { + connectSession, createThread, createTurn, cwd, itemStarted, sessionId, turnCompleted, turnStarted, type PromptSession, +} from "./v2-prompt-harness"; +import {expectConformingV2SessionUpdates} from "./v2-session-update-guard"; + +const worker = "worker"; +const childId = `${sessionId}:subagent:${worker}`; +const spawn: ThreadItem = {type: "subAgentActivity", id: "spawn", kind: "started", + agentThreadId: worker, agentPath: "/root/worker", model: null, reasoningEffort: null}; +const answer: ThreadItem = {type: "agentMessage", id: "retained", text: "Retained answer", + phase: null, memoryCitation: null, delivery: null, questions: null}; +const onWorker = (event: ServerNotification): ServerNotification => + ({...event, params: {...event.params, threadId: worker}} as ServerNotification); + +async function flush(h: PromptSession) { await h.waitForNotifications(); await h.request("session/list", {}); } +function recordedTurn(status: TurnStatus, id = "recorded"): Turn { + return {...createTurn(status, id), items: [answer], itemsView: "full", + error: status === "failed" ? {message: "Recorded failure", codexErrorInfo: null, + additionalDetails: "Private backend diagnostic", misalignment: null} : null}; +} + +describe.each([1, 2] as const)("ACP v%s historical child outcomes", version => { + let h: PromptSession; + afterEach(async () => { + if (h) { await flush(h); h.connection.close(); } + expectConformingV2SessionUpdates(); + }); + async function connect(options: Parameters[1] = {}) { + h = await connectSession(version, {v1ClientCapabilities: {subagents: {}}, ...options}); + } + const load = () => h.request(version === 1 ? "session/load" : "session/resume", { + sessionId, cwd, mcpServers: [], ...(version === 2 ? {replayFrom: {type: "start"}} : {}), + }); + const states = () => h.scopedUpdates.flatMap(entry => entry.update.sessionUpdate === "subagent_update" + && entry.update.sessionId === childId && entry.update.state != null ? [entry.update.state as StateUpdate] : []); + const stateUpdates = () => h.scopedUpdates.filter(entry => entry.update.sessionUpdate === "subagent_update" + && entry.update.sessionId === childId && entry.update.state != null + || entry.sessionId === childId && entry.update.sessionUpdate === "state_update"); + const gap = () => h.scopedUpdates.some(entry => entry.update.sessionUpdate === "agent_thought_chunk" + && entry.update.messageId === `${sessionId}:subagent-history-gap`); + + function history(turns: Turn[], paginated = false) { + const reads: Array<{threadId: string; includeTurns?: boolean}> = []; + h.setCodexResponse("thread/read", async params => { + const request = params as {threadId: string; includeTurns?: boolean}; + reads.push(request); + return {thread: {...createThread(), id: request.threadId, status: {type: "notLoaded"}, + parentThreadId: request.threadId === worker ? "foreign-root" : null, + historyMode: request.threadId === worker && paginated ? "paginated" : "legacy", + turns: request.threadId === worker ? request.includeTurns ? turns : [] + : [{...createTurn("completed"), itemsView: "full", items: [spawn]}]}}; + }); + const summaries = vi.fn(async () => ({ + data: turns.length ? [{...turns.at(-1)!, items: [], itemsView: "notLoaded"}] : [], + nextCursor: null, backwardsCursor: null, + })); + h.setCodexResponse("thread/turns/list", summaries); + h.setCodexResponse("thread/items/list", async () => ({ + data: turns.flatMap(turn => turn.items.map(item => ({item, turnId: turn.id, startedAtMs: 0, + completedAtMs: turn.status === "inProgress" ? null : 1}))), nextCursor: null, backwardsCursor: null, + })); + return {reads, summaries}; + } + + it.each(([false, true]).flatMap(paginated => + (["completed", "interrupted", "failed", "inProgress", "missing"] as const).map(status => ({paginated, status})), + ))("replays only a known final outcome before the response ($status, paginated: $paginated)", async ({status, paginated}) => { + await connect(); + const {reads, summaries} = history(status === "missing" ? [] : [recordedTurn(status)], paginated); + const start = h.wireMessages.length; + await load(); + const boundary = h.scopedUpdates.length; + await flush(h); + expect(h.scopedUpdates.slice(boundary).filter(entry => entry.sessionId === childId + || entry.update.sessionUpdate === "subagent_update")).toEqual([]); + expect(h.turnInterruptCalls()).toEqual([]); + await expect(h.request("session/set_config_option", {sessionId: childId, configId: "model", value: "anything", + ...(version === 2 ? {type: "id"} : {})})).rejects.toMatchObject({code: -32600}); + const final = status !== "missing" && status !== "inProgress"; + expect(states()).toHaveLength(final ? 1 : 0); + if (final) { + await expect(JSON.stringify(stateUpdates(), null, 2)) + .toMatchFileSnapshot(`data/historical-outcome-v${version}-${status}.json`); + const frames = h.wireMessages.slice(start).flatMap(frame => Array.isArray(frame) ? frame : [frame]); + const response = frames.findIndex(frame => "result" in frame && frame.result !== null + && typeof frame.result === "object" && "configOptions" in frame.result); + const outcome = frames.findIndex(frame => "method" in frame && frame.method === "session/update" + && (frame.params as {update: {state?: {state: string}}}).update.state?.state === "idle"); + expect(outcome).toBeGreaterThanOrEqual(0); + expect(response).toBeGreaterThan(outcome); + } + expect(JSON.stringify(h.scopedUpdates)).not.toContain("Private backend diagnostic"); + if (paginated) { + expect(summaries).toHaveBeenCalledExactlyOnceWith({threadId: worker, limit: 1, sortDirection: "desc", itemsView: "notLoaded"}); + expect(reads.some(read => read.threadId === worker && read.includeTurns)).toBe(false); + } else { + expect(summaries).not.toHaveBeenCalled(); + expect(reads.filter(read => read.threadId === worker && read.includeTurns)).toHaveLength(1); + } + }); + + it.each(["completed", "inProgress"] as const)("does not substitute an older outcome for the latest %s turn", async status => { + await connect(); + history([recordedTurn("failed", "older"), recordedTurn(status, "latest")]); + await load(); + await flush(h); + expect(states()).toEqual(status === "completed" ? [{state: "idle", stopReason: "end_turn"}] : []); + }); + + it("retains paginated transcripts and reports a gap when outcome summaries are unavailable", async () => { + await connect(); + history([recordedTurn("completed")], true); + h.setCodexResponse("thread/turns/list", async () => { throw new Error("Private outcome lookup diagnostic"); }); + await load(); + await flush(h); + expect(states()).toEqual([]); + expect(h.scopedUpdates.some(entry => entry.sessionId === childId && entry.update.sessionUpdate === "agent_message_chunk" + && entry.update.messageId === "retained")).toBe(true); + expect(gap()).toBe(true); + expect(JSON.stringify(h.scopedUpdates)).not.toContain("Private outcome lookup diagnostic"); + }); + + it("does not turn a replayed failure into current foreground state on the next resume", async () => { + await connect(); + history([recordedTurn("failed")]); + await load(); + await flush(h); + expect(states()).toHaveLength(1); + const before = states().length; + await h.request("session/resume", {sessionId, cwd, mcpServers: []}); + await flush(h); + expect(states().slice(before)).toEqual([]); + await h.cancel(); + await flush(h); + expect(h.turnInterruptCalls()).toEqual([]); + }); + + it("keeps a newer live permission and foreground state independent of a recorded ending", async () => { + let answerPermission!: () => void; + await connect({onRequestPermission: request => new Promise(resolve => { + answerPermission = () => resolve({outcome: {outcome: "selected", optionId: request.options[0]!.optionId}}); + })}); + h.emit(itemStarted(spawn)); + h.emit(onWorker(turnStarted("live-turn"))); + await flush(h); + const pending = h.triggerApproval("item/commandExecution/requestApproval", { + threadId: worker, turnId: "live-turn", itemId: "waiting", command: "echo next", cwd, + commandActions: [], reason: null, availableDecisions: ["accept", "decline"], + }); + try { + await vi.waitFor(() => expect(answerPermission).toBeTypeOf("function")); + history([recordedTurn("failed")]); + const before = states().length; + await load(); + await vi.waitFor(() => expect(states().at(-1)).toMatchObject({state: "requires_action"})); + expect(states().slice(before).some(state => state.state === "idle" && state.stopReason === "error")).toBe(true); + answerPermission(); + await expect(pending).resolves.toEqual({decision: "accept"}); + h.emit(onWorker(turnCompleted("live-turn"))); + await vi.waitFor(() => expect(states().at(-1)).toEqual({state: "idle", stopReason: "end_turn"})); + } finally { answerPermission?.(); } + }); + + it("preserves nested parentage and emits each child's outcome once across repeated spawn records", async () => { + await connect(); + history([]); + const coordinatorId = `${sessionId}:subagent:coordinator`; + const collaboration: ThreadItem = {type: "collabAgentToolCall", id: "legacy-spawn", tool: "spawnAgent", status: "completed", + senderThreadId: "coordinator", receiverThreadIds: [worker], prompt: null, model: null, reasoningEffort: null, agentsStates: {}}; + h.setCodexResponse("thread/read", async params => { + const id = (params as {threadId: string}).threadId; + return {thread: {...createThread(), id, status: {type: "notLoaded"}, parentThreadId: "foreign-root", + turns: id === sessionId ? [{...createTurn("completed"), itemsView: "full", items: [ + {...spawn, agentThreadId: "coordinator", agentPath: "/root/coordinator"}, + {...collaboration, senderThreadId: sessionId}, + ]}] : id === "coordinator" ? [{...createTurn("completed"), itemsView: "full", items: [collaboration]}] + : [recordedTurn("interrupted")]}}; + }); + await load(); + await flush(h); + const outcomes = h.scopedUpdates.flatMap(entry => entry.update.sessionUpdate === "subagent_update" + && entry.update.state != null ? [{parent: entry.sessionId, child: entry.update.sessionId, state: entry.update.state}] : []); + expect(outcomes).toEqual([ + {parent: coordinatorId, child: childId, state: {state: "idle", stopReason: "cancelled"}}, + {parent: sessionId, child: coordinatorId, state: {state: "idle", stopReason: "end_turn"}}, + ]); + }); + + it("does not publish a late summary after close", async () => { + await connect(); + history([recordedTurn("completed")], true); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + let reading = false; + h.setCodexResponse("thread/turns/list", async () => { + reading = true; + await held; + return {data: [recordedTurn("completed")], nextCursor: null, backwardsCursor: null}; + }); + const replay = load(); + void replay.catch(() => {}); + try { + await vi.waitFor(() => expect(reading).toBe(true)); + await h.request("session/close", {sessionId}); + release(); + await expect(replay).rejects.toMatchObject({code: -32600}); + expect(states().some(state => state.state === "idle")).toBe(false); + } finally { release(); } + }); + + it("uses the existing readable failure fallback when the recorded failure has no details", async () => { + await connect(); + history([{...recordedTurn("failed"), error: null}]); + await load(); + expect(states()).toEqual([{state: "idle", stopReason: "error", error: {code: -32603, message: "Subagent work failed"}}]); + }); +}); diff --git a/src/__tests__/integration/standard-subagent-live.test.ts b/src/__tests__/integration/standard-subagent-live.test.ts index 660172157..d696dbfc7 100644 --- a/src/__tests__/integration/standard-subagent-live.test.ts +++ b/src/__tests__/integration/standard-subagent-live.test.ts @@ -160,11 +160,15 @@ describeIntegration("integration: live standard subagents", () => { expect(updates.some(entry => state(entry)?.state === "idle")).toBe(true); }, {timeout: 15_000}); if (ending === "complete" || ending === "terminal") { + // A cold attachment has no cached live ending to mistake for a replayed outcome. + if (ending === "complete") await request("session/close", {sessionId}); const replayStart = updates.length; await load(sessionId); await vi.waitFor(() => expect(updates.slice(replayStart).some(entry => entry.sessionId === childId && entry.update.sessionUpdate === "agent_message_chunk" && entry.update.content.text.includes("CHILD_RESULT")), stderr).toBe(true)); + await vi.waitFor(() => expect(updates.slice(replayStart).some(entry => + state(entry)?.state === "idle" && state(entry)?.stopReason === "end_turn"), stderr).toBe(true)); } expect(updates.some(entry => ["subagent_spawned", "subagent_state_update", "_subagent_update"].includes(entry.update.sessionUpdate))).toBe(false); expect(mock.unexpectedRequests).toBe(0); diff --git a/src/subagents/StandardSubagentEventRouter.ts b/src/subagents/StandardSubagentEventRouter.ts index 149daa8f5..7b01a9524 100644 --- a/src/subagents/StandardSubagentEventRouter.ts +++ b/src/subagents/StandardSubagentEventRouter.ts @@ -1,6 +1,6 @@ import type {ReasoningEffort, ServerNotification} from "../app-server"; import type {StateUpdate} from "@agentclientprotocol/sdk/experimental/v2/unstable"; -import type {Thread, ThreadItem} from "../app-server/v2"; +import type {Thread, ThreadItem, Turn} from "../app-server/v2"; import {ACPSessionConnection} from "../ACPSessionConnection"; import {AcpV2ForegroundState} from "../AcpV2ForegroundState"; import {CodexPlanStream} from "../CodexPlanStream"; @@ -218,17 +218,14 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { return true; case "turn/completed": { const turn = notification.params.turn; - if (turn.status === "inProgress" || child.endedTurns.has(turn.id)) return true; + const state = turnEndState(turn); + if (state === undefined || child.endedTurns.has(turn.id)) return true; this.endWorkRequests(child, turn.id); child.endedTurns.add(turn.id); if (child.turnId !== null && child.turnId !== turn.id) return true; child.turnId = null; if (child.foregroundTurnId !== null && child.foregroundTurnId !== turn.id) return true; - await child.foreground.update(toV2IdleState( - {stopReason: turn.status === "interrupted" ? "cancelled" : "end_turn"}, - turn.status === "failed" ? {code: -32603, - message: turn.error ? readableServiceErrorMessage(turn.error.message) : "Subagent work failed"} : undefined, - )); + await child.foreground.update(state); child.foregroundTurnId = null; this.onChildWorkEnded(child.sessionId); return true; @@ -586,6 +583,17 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { await child.foreground.start(); } } + + /** Historical state bypasses the current foreground cache and grants no runtime ownership. */ + async replayTurnOutcome(threadId: string, turn: Turn): Promise { + const child = this.children.get(threadId); + const state = turnEndState(turn); + if (!child || state === undefined || this.retired) return; + await this.ensureAnnounced(child); + if (this.retired) return; + await this.session.updateSubagentState(child.sessionId, child.parentSessionId, state); + } + childSessionIds(): string[] { return [...this.children.values()].map(child => child.sessionId); } isNativeSubagentThread(threadId: string): boolean { return this.children.has(threadId); } shouldIgnore(): boolean { return false; } @@ -622,3 +630,19 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { legacyCollaborationStarted(item: ThreadItem & {type: "collabAgentToolCall"}) { return CollabAgentReporter.started(item); } legacyCollaborationCompleted(item: ThreadItem & {type: "collabAgentToolCall"}) { return CollabAgentReporter.completed(item); } } + +function turnEndState(turn: Turn): StateUpdate | undefined { + switch (turn.status) { + case "inProgress": + return undefined; + case "completed": + return toV2IdleState({stopReason: "end_turn"}); + case "interrupted": + return toV2IdleState({stopReason: "cancelled"}); + case "failed": + return toV2IdleState({stopReason: "end_turn"}, { + code: -32603, + message: turn.error ? readableServiceErrorMessage(turn.error.message) : "Subagent work failed", + }); + } +}