From b481af72ec2bd30838c0b268f0455521770d2aaa Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 26 Oct 2026 14:26:16 +0100 Subject: [PATCH 1/3] feat: replay recorded subagent conversations --- readme-dev.md | 14 +- src/CodexAcpClient.ts | 29 ++ src/CodexAcpServer.ts | 88 +++-- src/CodexEventHandler.ts | 17 +- src/CodexSessionToolCalls.ts | 13 + .../subagent-replay-v1-completed-tool.json | 68 ++++ .../data/subagent-replay-v1-running-tool.json | 56 +++ .../data/subagent-replay-v1-transcript.json | 13 + .../subagent-replay-v2-completed-tool.json | 83 +++++ .../data/subagent-replay-v2-running-tool.json | 65 ++++ .../data/subagent-replay-v2-transcript.json | 21 ++ .../standard-subagent-live.test.ts | 4 +- .../standard-subagent-replay.test.ts | 342 ++++++++++++++++++ .../standard-subagent-live.test.ts | 7 + src/subagents/StandardSubagentEventRouter.ts | 44 ++- src/tool-calls/reporters/CommandReporter.ts | 12 + 16 files changed, 825 insertions(+), 51 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v1-completed-tool.json create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v1-running-tool.json create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v1-transcript.json create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v2-running-tool.json create mode 100644 src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json create mode 100644 src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts diff --git a/readme-dev.md b/readme-dev.md index 55884f2fc..c365e3769 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -259,9 +259,17 @@ client mutations are not enabled. Parent cancellation interrupts observed live d An observed work ending reports `idle`; loss of observability reports `unknown`, not a fabricated failure or a permanently closed conversation. -Replay currently restores recoverable associations, not recursive child transcripts or historical -work outcomes. A parent advisory makes that gap visible. Rejoining a loaded child requires current -provider parentage; copied fork history alone cannot acquire the original runtime's subscriptions. +Replay follows both recorded spawn formats and replays available child transcripts once per conversation, +including nested children, using legacy or paginated history. Announcements and all selected history +precede the response. Completed snapshots reconcile tool and rendering state before duplicate live +payloads are suppressed; live bookkeeping still runs. In-progress tools retain classification without +resetting streamed output. Incomplete text has no safe delta cursor and is omitted rather than replayed +over newer buffered content. Agent-owned input is not 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. After replay, current child work/model 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 af2885408..018334a2f 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -85,6 +85,7 @@ 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}; /** * The slice of `thread/resume` the session layer consumes, plus whether Codex @@ -827,6 +828,18 @@ 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}> { + const {thread} = await this.codexClient.threadRead({threadId}); + if (thread.id !== threadId) throw new Error("Subagent history metadata names a different thread"); + return { + thread, + items: thread.historyMode === "legacy" + ? legacySubagentItems(this.codexClient, threadId) + : paginatedSubagentItems(this.codexClient, threadId), + }; + } + restoreSubagentSubscription(rootSessionId: string, parentThreadId: string, thread: Thread): Promise { return this.subagents.restoreChild(rootSessionId, parentThreadId, thread); } @@ -1794,3 +1807,19 @@ async function* noItems(): AsyncGenerator {} async function* oneItemPage(items: T[]): AsyncGenerator { if (items.length > 0) yield items; } + +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: "status" in item ? item.status !== "inProgress" : turn.status !== "inProgress", + }))); +} + +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: "status" in item ? item.status !== "inProgress" : completedAtMs !== null, + })); + } +} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 5ef4a025e..e58455812 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -15,6 +15,7 @@ import { OPENAI_PROVIDER_ID, type SessionMetadata, type SessionMetadataWithThread, + type RecordedSubagentItem, type UrlElicitationRequester } from "./CodexAcpClient"; import { @@ -217,7 +218,7 @@ import type {SessionIndexTitleState} from "./SessionIndexTitles"; import {ASYNC_TASK_STOP_METHOD} from "./async-tasks/AsyncTaskExtension"; import {CodexBackgroundTerminalTasks} from "./async-tasks/CodexBackgroundTerminalTasks"; import {clientSupportsCompaction, CodexSessionCompactions, createCompactionUpdate} from "./CodexSessionCompactions"; -import {CodexSessionToolCalls} from "./CodexSessionToolCalls"; +import {CodexSessionToolCalls, PLAIN_TOOL_CALL_ITEM_TYPES} from "./CodexSessionToolCalls"; import { type AgentFileChangeReport, type AgentFileChangeReportRequest, @@ -3496,7 +3497,7 @@ export class CodexAcpServer { isOpen, ); if (this.usesStandardSubagents()) { - await this.streamStandardAssociations(sessionId, itemPages, sessionState, isOpen); + await this.streamStandardHistory(sessionId, standardHistoryItems(itemPages), sessionState, isOpen); return; } if (this.supportsNativeSubagents()) { @@ -3689,62 +3690,87 @@ export class CodexAcpServer { } } - /** Association-only replay: child transcripts and recorded outcomes are a separate slice. */ - private async streamStandardAssociations( - sessionId: string, pages: AsyncIterable, state: SessionState, isOpen: () => boolean, + /** Recorded conversation data is independent of verified runtime ownership and current work. */ + private async streamStandardHistory( + 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; if (!(router instanceof StandardSubagentEventRouter)) return; const session = new ACPSessionConnection(this.connection, sessionId); const started = new Set(); - const restored = new Set(); - const restore = async (threadId: string, childId: string | null) => { - if (childId === null || restored.has(threadId)) return; - restored.add(threadId); + const replayChild = async (childThreadId: string, childId: string | null) => { + if (childId === null || replayed.has(childThreadId)) return; + if (ancestry.has(childThreadId)) { gaps.add(childThreadId); return; } + replayed.add(childThreadId); try { - const metadata = await this.codexAcpClient.appServerClient.threadRead({threadId}); + const history = await this.codexAcpClient.readSubagentHistory(childThreadId); if (!isOpen()) throw new SessionClosedDuringLoadError(); - const parent = router.parentThreadId(threadId); + const parent = router.parentThreadId(childThreadId); if (parent !== undefined) { - const runtime = await this.codexAcpClient.restoreSubagentSubscription(sessionId, parent, metadata.thread); - if (runtime) await router.restoreRuntime(runtime, this.codexAcpClient.appServerClient.runningTurnId(threadId)); + try { + const runtime = await this.codexAcpClient.restoreSubagentSubscription(state.sessionId, parent, history.thread); + if (runtime) await router.restoreRuntime(runtime, this.codexAcpClient.appServerClient.runningTurnId(childThreadId)); + } catch (error) { logger.error(`Failed to restore subagent ${childThreadId}`, error); } } + await this.streamStandardHistory(childId, history.items, state, isOpen, childThreadId, + new Set([...ancestry, childThreadId]), replayed, gaps); } catch (error) { if (error instanceof SessionClosedDuringLoadError) throw error; - logger.error(`Failed to restore subagent ${threadId}`, error); + gaps.add(childThreadId); + logger.error(`Failed to replay subagent ${childThreadId}`, error); } }; - const unresolved: Array = []; + const replayActivity = async (item: ThreadItem & {type: "subAgentActivity"}): Promise => { + const childId = await router.announceHistory(item.agentThreadId, item.agentPath); + if (childId === null) return false; + await reportSubagentModel(session, childId, this.subagentModels(state), item.model || undefined); + await replayChild(item.agentThreadId, childId); + return true; + }; + let unresolved: Array = []; for await (const items of pages) { - for (const item of items) { + for (const {item, completed, turnId} of items) { if (!isOpen()) throw new SessionClosedDuringLoadError(); if (item.type === "subAgentActivity") { - if (isRootAgentPath(item.agentPath) || item.agentThreadId === sessionId) continue; - const childId = await router.announceHistory(item.agentThreadId, item.agentPath); - if (childId === null) unresolved.push(item); - else { - await reportSubagentModel(session, childId, this.subagentModels(state), item.model || undefined); - await restore(item.agentThreadId, childId); - } + if (isRootAgentPath(item.agentPath)) continue; + if (ancestry.has(item.agentThreadId)) { gaps.add(item.agentThreadId); continue; } + if (!await replayActivity(item)) unresolved.push(item); continue; } if (item.type === "collabAgentToolCall" && item.tool === "spawnAgent") { - if (item.senderThreadId === sessionId) for (const child of item.receiverThreadIds) { - if (child.trim() && child !== sessionId) await restore(child, await router.announce(child, sessionId, undefined, item.prompt ?? undefined)); + if (item.senderThreadId === threadId) for (const child of item.receiverThreadIds) { + if (!child.trim() || ancestry.has(child)) { gaps.add(child); continue; } + await replayChild(child, await router.announce(child, sessionId, undefined, item.prompt ?? undefined)); } continue; } + if (threadId !== state.sessionId) { + // Agent-owned input is not an ordinary human message. Incomplete text has no safe delta cursor. + if (item.type === "userMessage" || (!completed && !PLAIN_TOOL_CALL_ITEM_TYPES.has(item.type))) { + gaps.add(threadId); + continue; + } + if (!completed) gaps.add(threadId); + if (!await router.reconcileHistoryItem(threadId, item, completed, turnId)) { gaps.add(threadId); continue; } + if (!isOpen()) throw new SessionClosedDuringLoadError(); + } for (const update of await this.createHistoryUpdates(item, state)) { if (this.protocolVersion === 2) await this.sendReplayMessageStart(session, update, started); await session.update(update); } } } - for (const item of unresolved) await restore(item.agentThreadId, await router.announceHistory(item.agentThreadId, item.agentPath)); + while (unresolved.length > 0) { + const pending: typeof unresolved = []; + for (const item of unresolved) if (!await replayActivity(item)) pending.push(item); + if (pending.length === unresolved.length) { for (const item of pending) gaps.add(item.agentThreadId); break; } + unresolved = pending; + } if (!isOpen()) throw new SessionClosedDuringLoadError(); - if (restored.size > 0) { + if (threadId === state.sessionId && gaps.size > 0) { const update = {sessionUpdate: "agent_thought_chunk" as const, messageId: `${sessionId}:subagent-history-gap`, - content: {type: "text" as const, text: "Recorded child transcripts are not included in this replay."}}; + content: {type: "text" as const, text: "Some recorded child history is unavailable or incomplete."}}; if (this.protocolVersion === 2) await this.sendReplayMessageStart(session, update, started); await session.update(update); } @@ -5546,6 +5572,12 @@ async function* itemsOfEntries(pages: AsyncIterable): AsyncGe } } +async function* standardHistoryItems(pages: AsyncIterable): AsyncGenerator { + for await (const page of pages) { + yield page.map(item => ({item, completed: !("status" in item) || item.status !== "inProgress"})); + } +} + function isUuidV7(id: string): boolean { return /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id); } diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index b79c366a3..a178e68df 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -41,6 +41,7 @@ import {CompactionReporter} from "./tool-calls/reporters/CompactionReporter"; import {DynamicToolReporter} from "./tool-calls/reporters/DynamicToolReporter"; import {FileChangeReporter} from "./tool-calls/reporters/FileChangeReporter"; import {CodexItemStreamState} from "./CodexItemStreamState"; +import {PLAIN_TOOL_CALL_ITEM_TYPES} from "./CodexSessionToolCalls"; import {ImageGenerationReporter} from "./tool-calls/reporters/ImageGenerationReporter"; import {ImageViewReporter} from "./tool-calls/reporters/ImageViewReporter"; import {McpToolReporter} from "./tool-calls/reporters/McpToolReporter"; @@ -180,22 +181,6 @@ const STRUCTURED_CODEX_ERROR_CATEGORIES: Partial; -/** - * Item types that render as a plain `tool_call`/`tool_call_update` and have no outstanding-item - * tracker of their own (`contextCompaction` has `CodexSessionCompactions`; `collabAgentToolCall`/ - * `subAgentActivity` have `CodexSubagentEventRouter`). Used to track which open items a provider - * restart's close-out needs to fail. - */ -const PLAIN_TOOL_CALL_ITEM_TYPES = new Set([ - "fileChange", - "commandExecution", - "mcpToolCall", - "dynamicToolCall", - "webSearch", - "imageView", - "imageGeneration", -]); - /** Prompt failures whose text the client has already received as an agent message. */ const failuresShownAsMessages = new WeakSet(); diff --git a/src/CodexSessionToolCalls.ts b/src/CodexSessionToolCalls.ts index 129eef46f..67cb750cb 100644 --- a/src/CodexSessionToolCalls.ts +++ b/src/CodexSessionToolCalls.ts @@ -1,4 +1,11 @@ import type {UpdateSessionEvent} from "./ACPSessionConnection"; +import type {ThreadItem} from "./app-server/v2"; +import {commandUsesTerminal} from "./tool-calls/reporters/CommandReporter"; + +/** Item types tracked as ordinary tool calls, without a separate lifecycle tracker. */ +export const PLAIN_TOOL_CALL_ITEM_TYPES = new Set([ + "fileChange", "commandExecution", "mcpToolCall", "dynamicToolCall", "webSearch", "imageView", "imageGeneration", +]); /** * Tracks tool-call items that reported `item/started` but not yet `item/completed`, for the item @@ -19,6 +26,12 @@ export class CodexSessionToolCalls { this.open.delete(itemId); } + restore(item: ThreadItem, completed: boolean, turnId?: string): void { + if (!PLAIN_TOOL_CALL_ITEM_TYPES.has(item.type)) return; + if (completed) this.complete(item.id); + else if (!this.open.has(item.id)) this.start(item.id, item.type === "commandExecution" && commandUsesTerminal(item), turnId); + } + /** Drops all tracked entries without emitting anything, e.g. when their turn ends. */ clear(): void { this.open.clear(); diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v1-completed-tool.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-completed-tool.json new file mode 100644 index 000000000..79396917b --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-completed-tool.json @@ -0,0 +1,68 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "title": "sleep 12", + "kind": "execute", + "status": "completed", + "content": [ + { + "terminalId": "command", + "type": "terminal" + } + ], + "rawInput": { + "command": "sleep 12", + "cwd": "/workspace" + }, + "_meta": { + "terminal_info": { + "cwd": "/workspace", + "terminal_id": "command" + } + }, + "sessionUpdate": "tool_call" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "_meta": { + "terminal_output_delta": { + "data": "Recorded output", + "terminal_id": "command" + }, + "terminal_exit": { + "exit_code": 0, + "signal": null, + "terminal_id": "command" + } + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "Saved answer", + "type": "text" + }, + "messageId": "saved", + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "New answer", + "type": "text" + }, + "messageId": "new", + "sessionUpdate": "agent_message_chunk" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v1-running-tool.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-running-tool.json new file mode 100644 index 000000000..eae7f1a99 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-running-tool.json @@ -0,0 +1,56 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "title": "sleep 12", + "kind": "execute", + "status": "in_progress", + "content": [ + { + "terminalId": "command", + "type": "terminal" + } + ], + "rawInput": { + "command": "sleep 12", + "cwd": "/workspace" + }, + "_meta": { + "terminal_info": { + "cwd": "/workspace", + "terminal_id": "command" + } + }, + "sessionUpdate": "tool_call" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "_meta": { + "terminal_output_delta": { + "data": "After", + "terminal_id": "command" + } + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "status": "completed", + "_meta": { + "terminal_exit": { + "exit_code": 0, + "signal": null, + "terminal_id": "command" + } + }, + "sessionUpdate": "tool_call_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v1-transcript.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-transcript.json new file mode 100644 index 000000000..0ecf33d9c --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v1-transcript.json @@ -0,0 +1,13 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "content": { + "text": "Saved child answer", + "type": "text" + }, + "messageId": "saved", + "sessionUpdate": "agent_message_chunk" + } + } +] \ 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 new file mode 100644 index 000000000..c241adb64 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-completed-tool.json @@ -0,0 +1,83 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "command", + "command": "sleep 12", + "cwd": "/workspace", + "sessionUpdate": "terminal_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "title": "sleep 12", + "kind": "execute", + "status": "completed", + "content": [ + { + "terminalId": "command", + "type": "terminal" + } + ], + "rawInput": { + "command": "sleep 12", + "cwd": "/workspace" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "command", + "output": { + "data": "UmVjb3JkZWQgb3V0cHV0" + }, + "exitStatus": { + "exitCode": 0, + "signal": null + }, + "sessionUpdate": "terminal_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "saved", + "content": [], + "sessionUpdate": "agent_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "saved", + "content": { + "text": "Saved answer", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "new", + "content": { + "text": "New answer", + "type": "text" + }, + "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/data/subagent-replay-v2-running-tool.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-running-tool.json new file mode 100644 index 000000000..a9f70ba61 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-running-tool.json @@ -0,0 +1,65 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "command", + "command": "sleep 12", + "cwd": "/workspace", + "sessionUpdate": "terminal_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "title": "sleep 12", + "kind": "execute", + "status": "in_progress", + "content": [ + { + "terminalId": "command", + "type": "terminal" + } + ], + "rawInput": { + "command": "sleep 12", + "cwd": "/workspace" + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "command", + "data": "QWZ0ZXI=", + "sessionUpdate": "terminal_output_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "state": "running", + "sessionUpdate": "state_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "command", + "exitStatus": { + "exitCode": 0, + "signal": null + }, + "sessionUpdate": "terminal_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "command", + "status": "completed", + "sessionUpdate": "tool_call_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json new file mode 100644 index 000000000..2f892592a --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/subagent-replay-v2-transcript.json @@ -0,0 +1,21 @@ +[ + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "saved", + "content": [], + "sessionUpdate": "agent_message" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "messageId": "saved", + "content": { + "text": "Saved child answer", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts index fc945cd32..9d100a01e 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts @@ -348,7 +348,7 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { expect(options).toEqual(["latest-settings"]); }); - it("replays associations and parent history without claiming complete child history", async () => { + it("replays parent history and empty child associations without inventing a gap or current runtime", async () => { await connect(); const item: ThreadItem = {type: "agentMessage", id: "parent-answer", text: "Parent history", phase: null, memoryCitation: null, delivery: null, questions: null}; @@ -358,7 +358,7 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { expect(h.scopedUpdates.some(entry => entry.sessionId === sessionId && entry.update.sessionUpdate === "agent_message_chunk" && entry.update.messageId === "parent-answer")).toBe(true); expect(h.scopedUpdates.some(entry => entry.sessionId === sessionId && entry.update.sessionUpdate === "agent_thought_chunk" - && entry.update.messageId === `${sessionId}:subagent-history-gap`)).toBe(true); + && entry.update.messageId === `${sessionId}:subagent-history-gap`)).toBe(false); expect(childUpdates().filter(entry => entry.sessionId === childId)).toEqual([]); }); diff --git a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts new file mode 100644 index 000000000..d9dad0bf6 --- /dev/null +++ b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts @@ -0,0 +1,342 @@ +import {afterEach, describe, expect, it, vi} from "vitest"; +import type {ServerNotification} from "../../app-server"; +import type {ThreadItem, TurnStatus} from "../../app-server/v2"; +import { + agentMessageDelta, commandExecutionItem, connectSession, codexResponse, createThread, createTurn, cwd, + itemStarted, sessionId, turnCompleted, turnStarted, userMessageItem, type PromptSession, +} from "./v2-prompt-harness"; +import {expectConformingV2SessionUpdates} from "./v2-session-update-guard"; + +const worker = "worker"; +const childId = `${sessionId}:subagent:${worker}`; +const message = (id: string, text: string): ThreadItem => ({ + type: "agentMessage", id, text, phase: null, memoryCitation: null, delivery: null, questions: null, +}); +function activity(id = worker, path = "/root/worker"): Extract { + return {type: "subAgentActivity", id: `activity-${id}`, agentThreadId: id, agentPath: path, + kind: "started", model: null, reasoningEffort: null}; +} +function spawn(sender = sessionId, receiver = worker): Extract { + return {type: "collabAgentToolCall", id: `spawn-${receiver}`, tool: "spawnAgent", status: "completed", + senderThreadId: sender, receiverThreadIds: [receiver], prompt: null, model: null, reasoningEffort: null, agentsStates: {}}; +} +const onWorker = (event: ServerNotification): ServerNotification => ({...event, params: {...event.params, threadId: worker}} as ServerNotification); +const command = (status: "inProgress" | "completed") => + ({...commandExecutionItem("command", status), aggregatedOutput: status === "completed" ? "Recorded output" : "Before", exitCode: status === "completed" ? 0 : null}); +type Stored = {items: ThreadItem[]; parent?: string | null; status?: TurnStatus; paginated?: boolean; unavailable?: boolean; metadataId?: string}; + +async function flush(h: PromptSession) { + await h.waitForNotifications(); + await h.request("session/list", {}); +} + +function history(h: PromptSession, nodes: Record, onRead?: (id: string, includeTurns: boolean) => void | Promise) { + const reads: Array<{id: string; includeTurns: boolean}> = []; + const resumes: string[] = []; + const thread = (id: string) => ({ + ...createThread(), id: nodes[id]?.metadataId ?? id, parentThreadId: nodes[id]?.parent ?? null, + status: id === sessionId ? {type: "idle" as const} : {type: "notLoaded" as const}, + historyMode: nodes[id]?.paginated ? "paginated" as const : "legacy" as const, + }); + h.setCodexResponse("thread/resume", async params => { + const id = (params as {threadId: string}).threadId; + resumes.push(id); + return {...codexResponse("thread/start") as object, thread: thread(id)}; + }); + h.setCodexResponse("thread/read", async params => { + const {threadId: id, includeTurns = false} = params as {threadId: string; includeTurns?: boolean}; + reads.push({id, includeTurns}); + await onRead?.(id, includeTurns); + if (includeTurns && nodes[id]?.unavailable) throw new Error("Private missing transcript diagnostic"); + return {thread: {...thread(id), turns: includeTurns && !nodes[id]?.paginated ? [{ + ...createTurn(nodes[id]?.status ?? "completed", "child-turn"), itemsView: "full", items: nodes[id]?.items ?? [], + }] : []}}; + }); + return {reads, resumes}; +} + +describe.each([1, 2] as const)("ACP v%s recorded child replay", 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 childUpdates = (id = childId) => h.scopedUpdates.filter(entry => entry.sessionId === id); + const gap = () => h.scopedUpdates.filter(entry => entry.sessionId === sessionId + && entry.update.sessionUpdate === "agent_thought_chunk" && entry.update.messageId === `${sessionId}:subagent-history-gap`); + const text = (value: string) => h.scopedUpdates.some(entry => entry.update.sessionUpdate === "agent_message_chunk" + && entry.update.content !== null && typeof entry.update.content === "object" + && "text" in entry.update.content && entry.update.content.text === value); + + it.each(["activity", "collab"] as const)("replays retained child content before the response without runtime authority (%s)", async format => { + await connect(); + const {reads, resumes} = history(h, { + [sessionId]: {items: [format === "activity" ? activity() : spawn(), message("parent", "Parent history")]}, + [worker]: {parent: "foreign-root", items: [message("saved", "Saved child answer")]}, + }); + const start = h.wireMessages.length; + await load(); + const boundary = h.scopedUpdates.length; + await flush(h); + expect(text("Saved child answer")).toBe(true); + expect(text("Parent history")).toBe(true); + expect(resumes).toEqual([sessionId]); + expect(reads.filter(read => read.id === worker && read.includeTurns)).toHaveLength(1); + expect(h.scopedUpdates.slice(boundary).filter(entry => entry.sessionId === childId)).toEqual([]); + expect(h.transcript.some(entry => "permissionRequest" in entry || "elicitationRequest" in entry)).toBe(false); + 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 replay = frames.findIndex(frame => "method" in frame && frame.method === "session/update" + && (frame.params as {sessionId: string; update: {messageId?: string}}).sessionId === childId + && (frame.params as {update: {messageId?: string}}).update.messageId === "saved"); + expect(replay).toBeGreaterThanOrEqual(0); + expect(response).toBeGreaterThan(replay); + expect(gap()).toEqual([]); + await expect(JSON.stringify(childUpdates(), null, 2)).toMatchFileSnapshot(`data/subagent-replay-v${version}-transcript.json`); + }); + + it("visits a conversation once even when both spawn formats and repeated assignments name it", async () => { + await connect(); + const {reads} = history(h, { + [sessionId]: {items: [activity(), spawn(), {...activity(), id: "later-assignment", kind: "interacted"}]}, + [worker]: {items: [message("one-answer", "One retained conversation")]}, + }); + await load(); + await flush(h); + expect(reads.filter(read => read.id === worker && read.includeTurns)).toHaveLength(1); + expect(childUpdates().filter(entry => entry.update.sessionUpdate === "agent_message_chunk")).toHaveLength(1); + }); + + it("recurses through immediate parents without attributing agent input to the human", async () => { + await connect(); + history(h, { + [sessionId]: {items: [activity("coordinator", "/root/coordinator")]}, + coordinator: {parent: sessionId, items: [spawn("coordinator")]}, + [worker]: {parent: "coordinator", items: [userMessageItem(null, "Agent-owned input"), message("nested", "Nested answer")]}, + }); + await load(); + await flush(h); + const known = new Set([sessionId]); + for (const entry of h.scopedUpdates) { + expect(known.has(entry.sessionId)).toBe(true); + if (entry.update.sessionUpdate === "subagent_update" && typeof entry.update.sessionId === "string") known.add(entry.update.sessionId); + } + expect(text("Nested answer")).toBe(true); + expect(childUpdates().some(entry => entry.update.sessionUpdate === "user_message_chunk")).toBe(false); + expect(gap()).toHaveLength(1); + }); + + it("resolves recoverable multi-level ancestry recorded in reverse order", async () => { + await connect(); + history(h, { + [sessionId]: {items: [activity(worker, "/root/a/b/worker"), activity("b", "/root/a/b"), activity("a", "/root/a")]}, + a: {items: []}, b: {items: []}, [worker]: {items: [message("deep", "Deep answer")]}, + }); + await load(); + await flush(h); + expect(text("Deep answer")).toBe(true); + expect(gap()).toEqual([]); + }); + + it("reads paginated child items in order without requesting legacy turns or work outcomes", 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 => + ({item, turnId: "stored-turn", startedAtMs: 0, completedAtMs: 1})); + h.setCodexResponse("thread/items/list", async params => { + const {sortDirection, cursor} = params as {sortDirection: string; cursor: string | null}; + 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"); }); + 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(); + const ids = childUpdates().flatMap(entry => entry.update.sessionUpdate === "agent_message_chunk" ? [entry.update.messageId] : []); + expect(ids).toEqual(["first-page", "last-page"]); + }); + + it.each([false, true])("retains associations and sibling history when a child snapshot is unavailable or misidentified (%s)", async mismatch => { + await connect(); + history(h, { + [sessionId]: {items: [spawn(), activity("sibling", "/root/sibling")]}, + [worker]: {items: [message("wrong", "Do not publish")], unavailable: !mismatch, ...(mismatch ? {metadataId: "other-thread"} : {})}, + sibling: {items: [message("retained", "Sibling answer")]}, + }); + await load(); + await flush(h); + expect(text("Do not publish")).toBe(false); + expect(text("Sibling answer")).toBe(true); + expect(h.scopedUpdates.some(entry => entry.update.sessionUpdate === "subagent_update" && entry.update.sessionId === childId)).toBe(true); + expect(gap()).toHaveLength(1); + expect(h.scopedUpdates.some(entry => JSON.stringify(entry).includes("Private missing transcript diagnostic"))).toBe(false); + }); + + async function startCommand() { + await connect(); + h.emit(itemStarted(activity())); + h.emit(onWorker(turnStarted("child-turn"))); + h.emit(onWorker(itemStarted(command("inProgress"), "child-turn"))); + await flush(h); + } + + it("reconciles completed tools before suppressing overlapping live completion and cleanup", async () => { + await startCommand(); + const before = h.scopedUpdates.length; + history(h, { + [sessionId]: {items: [activity()]}, + [worker]: {items: [command("completed"), message("saved", "Saved answer")]}, + }, (id, full) => { + if (id === worker && full) { + h.emit({method: "item/completed", params: {threadId: worker, turnId: "child-turn", item: command("completed"), completedAtMs: 0}}); + h.emit(onWorker(agentMessageDelta("Saved answer", "child-turn", "saved"))); + h.emit(onWorker(agentMessageDelta("New answer", "child-turn", "new"))); + h.emit(onWorker(turnCompleted("child-turn"))); + } + }); + await load(); + await vi.waitFor(() => expect(text("New answer")).toBe(true)); + await flush(h); + const updates = h.scopedUpdates.slice(before).filter(entry => entry.sessionId === childId); + expect(updates.some(entry => "status" in entry.update && entry.update.status === "failed")).toBe(false); + expect(updates.filter(entry => entry.update.sessionUpdate === "agent_message_chunk" && entry.update.messageId === "saved")).toHaveLength(1); + await expect(JSON.stringify(updates, null, 2)).toMatchFileSnapshot(`data/subagent-replay-v${version}-completed-tool.json`); + }); + + it("restores running command classification without resetting streamed bytes on a duplicate start", async () => { + await startCommand(); + h.emit({method: "item/commandExecution/outputDelta", params: {threadId: worker, turnId: "child-turn", itemId: "command", delta: "Before"}}); + await flush(h); + history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [command("inProgress")], status: "inProgress"}}, (id, full) => { + if (id === worker && full) { + h.emit(onWorker(itemStarted(command("inProgress"), "child-turn"))); + h.emit({method: "item/commandExecution/outputDelta", params: {threadId: worker, turnId: "child-turn", itemId: "command", delta: "After"}}); + } + }); + const before = h.scopedUpdates.length; + await load(); + await vi.waitFor(() => expect(h.scopedUpdates.slice(before).some(entry => + entry.update.sessionUpdate === "subagent_update" && entry.update.sessionId === childId + && entry.update.state && typeof entry.update.state === "object" + && (entry.update.state as {state: string}).state === "running")).toBe(true)); + h.emit({method: "item/completed", params: {threadId: worker, turnId: "child-turn", completedAtMs: 0, + item: {...command("completed"), aggregatedOutput: "BeforeAfter"}}}); + await flush(h); + const updates = h.scopedUpdates.slice(before).filter(entry => entry.sessionId === childId); + expect(JSON.stringify(updates)).not.toContain("BeforeAfter"); + expect(gap()).toHaveLength(1); + await expect(JSON.stringify(updates, null, 2)).toMatchFileSnapshot(`data/subagent-replay-v${version}-running-tool.json`); + }); + + it("does not duplicate incomplete text snapshots over buffered live deltas", async () => { + await connect(); + h.emit(itemStarted(activity())); + h.emit(onWorker(turnStarted("child-turn"))); + await flush(h); + history(h, { + [sessionId]: {items: [activity()]}, [worker]: {status: "inProgress", items: [message("partial", "Prefix")]}, + }, (id, full) => { + if (id === worker && full) { + h.emit(onWorker(agentMessageDelta("Prefix", "child-turn", "partial"))); + h.emit(onWorker(agentMessageDelta(" suffix", "child-turn", "partial"))); + } + }); + await load(); + await flush(h); + await vi.waitFor(() => { + const chunks = childUpdates().flatMap(entry => entry.update.sessionUpdate === "agent_message_chunk" + && entry.update.content !== null && typeof entry.update.content === "object" + && "text" in entry.update.content ? [entry.update.content.text] : []); + expect(chunks.join("")).toBe("Prefix suffix"); + }); + expect(gap()).toHaveLength(1); + }); + + it("does not resurrect a completed live tool from an older running snapshot", async () => { + await startCommand(); + h.emit({method: "item/completed", params: {threadId: worker, turnId: "child-turn", item: command("completed"), completedAtMs: 0}}); + h.emit(onWorker(turnCompleted("child-turn"))); + await flush(h); + history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [command("inProgress")], status: "inProgress"}}); + const before = h.scopedUpdates.length; + await load(); + await flush(h); + expect(h.scopedUpdates.slice(before).some(entry => entry.sessionId === childId + && (entry.update.sessionUpdate === "tool_call" || entry.update.sessionUpdate === "tool_call_update"))).toBe(false); + expect(gap()).toHaveLength(1); + }); + + it("removes completed runnable items without reviving or withdrawing a live permission", async () => { + let answer!: () => void; + await connect({onRequestPermission: request => new Promise(resolve => { + answer = () => resolve({outcome: {outcome: "selected", optionId: request.options[0]!.optionId}}); + })}); + h.emit(itemStarted(activity())); + h.emit(onWorker(turnStarted("child-turn"))); + h.emit(onWorker(itemStarted(command("inProgress"), "child-turn"))); + await flush(h); + const pending = h.triggerApproval("item/commandExecution/requestApproval", { + threadId: worker, turnId: "child-turn", itemId: "waiting", command: "echo next", cwd, + commandActions: [], reason: null, availableDecisions: ["accept", "decline"], + }); + try { + await vi.waitFor(() => expect(answer).toBeTypeOf("function")); + history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [command("completed")], status: "inProgress"}}); + await load(); + await vi.waitFor(() => expect(h.scopedUpdates.some(entry => entry.update.sessionUpdate === "subagent_update" + && entry.update.sessionId === childId && entry.update.state && typeof entry.update.state === "object" + && (entry.update.state as {state: string}).state === "requires_action")).toBe(true)); + answer(); + await expect(pending).resolves.toEqual({decision: "accept"}); + } finally { answer?.(); } + }); + + it("bounds cyclic history without reparenting a known ancestor", async () => { + await connect(); + const {reads} = history(h, { + [sessionId]: {items: [spawn()]}, + [worker]: {items: [spawn(worker, sessionId), message("cycle-safe", "Retained despite cycle")]}, + }); + await load(); + await flush(h); + expect(text("Retained despite cycle")).toBe(true); + expect(reads.filter(read => read.id === worker && read.includeTurns)).toHaveLength(1); + expect(gap()).toHaveLength(1); + }); + + it("does not publish selected child history after a parent closes during the read", async () => { + await connect(); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + let reading = false; + history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [message("too-late", "Do not publish")]}}, + async (id, full) => { if (id === worker && full) { reading = true; await held; } }); + 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(text("Do not publish")).toBe(false); + } finally { release(); } + }); + + it("does not fetch child transcripts when replay was not requested", async () => { + await connect(); + const {reads} = history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [message("saved", "Not selected")]}}); + await h.request("session/resume", {sessionId, cwd, mcpServers: []}); + await flush(h); + expect(reads.some(read => read.id === worker)).toBe(false); + expect(text("Not selected")).toBe(false); + }); +}); diff --git a/src/__tests__/integration/standard-subagent-live.test.ts b/src/__tests__/integration/standard-subagent-live.test.ts index 9dc3e324e..87affd153 100644 --- a/src/__tests__/integration/standard-subagent-live.test.ts +++ b/src/__tests__/integration/standard-subagent-live.test.ts @@ -150,6 +150,13 @@ describeIntegration("integration: live standard subagents", () => { expect(message, stderr).toBeGreaterThan(announcement); expect(updates.some(entry => state(entry)?.state === "idle")).toBe(true); }, {timeout: 15_000}); + if (ending === "complete" || ending === "terminal") { + 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)); + } expect(updates.some(entry => ["subagent_spawned", "subagent_state_update", "_subagent_update"].includes(entry.update.sessionUpdate))).toBe(false); expect(mock.unexpectedRequests).toBe(0); } finally { diff --git a/src/subagents/StandardSubagentEventRouter.ts b/src/subagents/StandardSubagentEventRouter.ts index 90692ac6d..8b689fada 100644 --- a/src/subagents/StandardSubagentEventRouter.ts +++ b/src/subagents/StandardSubagentEventRouter.ts @@ -53,6 +53,8 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { private readonly deferredLive: PendingNotificationBuffer; private readonly legacyActivities = new Set(); private readonly announced = new Set(); + private readonly replayedItems = new Map>(); + private readonly completedItems = new Map>(); private readonly models = new Map(); private readonly pendingModels = new Set(); private readonly spawnsInFlight = new Set(); @@ -155,10 +157,13 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { if (this.retired) return true; await this.interactionObserver?.(notification, child.sessionId); child.live = true; + const params = notification.params as {itemId?: string; item?: {id: string}}; + const itemId = params.itemId ?? params.item?.id; + const replayed = itemId === undefined ? undefined : this.replayedItems.get(threadId)?.get(itemId); const progressTurnId = (notification.params as {turnId?: string}).turnId; const progress = notification.method === "item/agentMessage/delta" || notification.method === "item/plan/delta" || notification.method === "item/reasoning/textDelta" || notification.method === "item/reasoning/summaryTextDelta"; - if (progressTurnId !== undefined && progress && !child.endedTurns.has(progressTurnId) + if (replayed !== true && progressTurnId !== undefined && progress && !child.endedTurns.has(progressTurnId) && (child.turnId === null || child.turnId === progressTurnId)) { child.turnId ??= progressTurnId; if (this.isCurrentForeground(child, progressTurnId)) { @@ -220,15 +225,23 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { } return true; case "item/started": + if (replayed === true) return true; if (notification.params.item.type === "userMessage") return true; if (child.turnId !== null && child.foregroundTurnId === notification.params.turnId) { await child.foreground.startItem(notification.params.item.id); } if (notification.params.item.type === "plan") return true; + if (replayed === false) return true; break; case "item/completed": { if (child.foregroundTurnId === notification.params.turnId) await child.foreground.completeItem(notification.params.item.id); const item = notification.params.item; + this.rememberCompletedItem(threadId, item.id); + if (replayed === true) { + child.toolCalls.complete(item.id); + if (item.type === "commandExecution") child.streams.commands.restore(item); + return true; + } if (item.type === "userMessage") return true; if (item.type === "plan") { const plans = new CodexPlanStream(this.session.forSession(child.sessionId), this.capabilities); @@ -241,7 +254,7 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { case "item/plan/delta": return true; } - return false; + return replayed === true; } private pendingFor(threadId: string): Pending { @@ -476,6 +489,33 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { const parentId = this.children.get(threadId)?.parentSessionId; return parentId === this.rootSessionId ? parentId : parentId === undefined ? undefined : this.associations.get(parentId)?.threadId; } + + /** Reconcile bookkeeping before suppressing duplicate presentation; history does not start work. */ + async reconcileHistoryItem(threadId: string, item: ThreadItem, completed: boolean, turnId?: string): Promise { + const child = this.children.get(threadId); + if (!child || this.retired) return false; + if (!completed && (this.completedItems.get(threadId)?.has(item.id) || (turnId !== undefined && child.endedTurns.has(turnId)))) return false; + child.toolCalls.restore(item, completed, turnId); + if (item.type === "commandExecution") child.streams.commands.restore(item); + if (item.type === "agentMessage") child.streams.agentMessagePhases.set(item.id, item.phase); + if (item.type === "imageView" && !completed) child.streams.emittedImageViewItems.add(item.id); + if (item.type === "imageGeneration" && !completed) child.streams.activeImageGenerationItems.add(item.id); + const items = this.replayedItems.get(threadId) ?? new Map(); + items.set(item.id, completed); + this.replayedItems.set(threadId, items); + if (completed) { + this.rememberCompletedItem(threadId, item.id); + await child.foreground.completeItem(item.id); + } + return true; + } + + private rememberCompletedItem(threadId: string, itemId: string): void { + const items = this.completedItems.get(threadId) ?? new Set(); + items.add(itemId); + this.completedItems.set(threadId, items); + } + async restoreRuntime(thread: Thread, turnId: string | undefined): Promise { const child = this.children.get(thread.id); if (!child || this.retired) return; diff --git a/src/tool-calls/reporters/CommandReporter.ts b/src/tool-calls/reporters/CommandReporter.ts index cf8b242a0..82ad50179 100644 --- a/src/tool-calls/reporters/CommandReporter.ts +++ b/src/tool-calls/reporters/CommandReporter.ts @@ -49,6 +49,18 @@ export class CommandReporter { return startFacts(item); } + /** Replay restores classification without forgetting bytes that already streamed live. */ + restore(item: CommandItem): void { + if (item.status === "inProgress") { + if (!this.terminalCommands.has(item.id) && !this.nonTerminalCommands.has(item.id)) this.started(item); + } else { + this.nonTerminalCommands.delete(item.id); + this.terminalCommands.delete(item.id); + this.streamedCommands.delete(item.id); + this.standardStreamedCommands.delete(item.id); + } + } + /** A chunk of command output. AIR gets the output of a read, search or list command at the end. */ outputDelta(itemId: string, delta: string): ToolFacts { if (delta.length > 0) this.standardStreamedCommands.add(itemId); From c3f2068e588d477a2f80d2781381dbdc035c1f37 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 26 Oct 2026 14:33:47 +0100 Subject: [PATCH 2/3] fix: classify recorded partial items conservatively --- src/CodexAcpClient.ts | 18 ++++++++++++++++-- src/CodexAcpServer.ts | 16 ++++++++++++++-- .../standard-subagent-replay.test.ts | 15 +++++++++++++++ src/subagents/StandardSubagentEventRouter.ts | 2 -- 4 files changed, 45 insertions(+), 6 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 018334a2f..6f88c3286 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -1812,14 +1812,28 @@ async function* legacySubagentItems(client: CodexAppServerClient, threadId: stri 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: "status" in item ? item.status !== "inProgress" : turn.status !== "inProgress", + item, turnId: turn.id, completed: recordedItemCompleted(item, turn.status !== "inProgress"), }))); } 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: "status" in item ? item.status !== "inProgress" : completedAtMs !== null, + item, turnId, completed: recordedItemCompleted(item, completedAtMs !== null), })); } } + +function recordedItemCompleted(item: ThreadItem, completed: boolean): boolean { + switch (item.type) { + case "commandExecution": + case "fileChange": + case "mcpToolCall": + case "dynamicToolCall": + case "collabAgentToolCall": + return item.status !== "inProgress"; + default: + // Open-ended provider strings (for example image generation status) are not completion flags. + return completed; + } +} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index e58455812..46a27faec 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -218,7 +218,7 @@ import type {SessionIndexTitleState} from "./SessionIndexTitles"; import {ASYNC_TASK_STOP_METHOD} from "./async-tasks/AsyncTaskExtension"; import {CodexBackgroundTerminalTasks} from "./async-tasks/CodexBackgroundTerminalTasks"; import {clientSupportsCompaction, CodexSessionCompactions, createCompactionUpdate} from "./CodexSessionCompactions"; -import {CodexSessionToolCalls, PLAIN_TOOL_CALL_ITEM_TYPES} from "./CodexSessionToolCalls"; +import {CodexSessionToolCalls} from "./CodexSessionToolCalls"; import { type AgentFileChangeReport, type AgentFileChangeReportRequest, @@ -3747,7 +3747,7 @@ export class CodexAcpServer { } if (threadId !== state.sessionId) { // Agent-owned input is not an ordinary human message. Incomplete text has no safe delta cursor. - if (item.type === "userMessage" || (!completed && !PLAIN_TOOL_CALL_ITEM_TYPES.has(item.type))) { + if (item.type === "userMessage" || (!completed && !supportsPartialHistory(item))) { gaps.add(threadId); continue; } @@ -5578,6 +5578,18 @@ async function* standardHistoryItems(pages: AsyncIterable): AsyncG } } +function supportsPartialHistory(item: ThreadItem): boolean { + switch (item.type) { + case "commandExecution": + case "fileChange": + case "mcpToolCall": + case "dynamicToolCall": + return true; + default: + return false; + } +} + function isUuidV7(id: string): boolean { return /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id); } diff --git a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts index d9dad0bf6..3d31132d0 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-replay.test.ts @@ -275,6 +275,21 @@ describe.each([1, 2] as const)("ACP v%s recorded child replay", version => { expect(gap()).toHaveLength(1); }); + it.each([false, true])("does not infer completion from an open-ended image status (paginated: %s)", async paginated => { + await connect(); + const image: ThreadItem = {type: "imageGeneration", id: "unfinished-image", + status: "provider-specific-running", revisedPrompt: null, failure: null, result: ""}; + history(h, {[sessionId]: {items: [activity()]}, [worker]: {items: [image], status: "inProgress", paginated}}); + if (paginated) h.setCodexResponse("thread/items/list", async () => ({ + data: [{item: image, turnId: "image-turn", startedAtMs: 0, completedAtMs: null}], + nextCursor: null, backwardsCursor: null, + })); + await load(); + await flush(h); + expect(childUpdates()).toEqual([]); + expect(gap()).toHaveLength(1); + }); + it("removes completed runnable items without reviving or withdrawing a live permission", async () => { let answer!: () => void; await connect({onRequestPermission: request => new Promise(resolve => { diff --git a/src/subagents/StandardSubagentEventRouter.ts b/src/subagents/StandardSubagentEventRouter.ts index 8b689fada..5b2e1e3d7 100644 --- a/src/subagents/StandardSubagentEventRouter.ts +++ b/src/subagents/StandardSubagentEventRouter.ts @@ -498,8 +498,6 @@ export class StandardSubagentEventRouter implements SubagentEventRouter { child.toolCalls.restore(item, completed, turnId); if (item.type === "commandExecution") child.streams.commands.restore(item); if (item.type === "agentMessage") child.streams.agentMessagePhases.set(item.id, item.phase); - if (item.type === "imageView" && !completed) child.streams.emittedImageViewItems.add(item.id); - if (item.type === "imageGeneration" && !completed) child.streams.activeImageGenerationItems.add(item.id); const items = this.replayedItems.get(threadId) ?? new Map(); items.set(item.id, completed); this.replayedItems.set(threadId, items); From 4b7120e47e881283e5fbb4d9c0ac3b1a0b0fc561 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 26 Oct 2026 14:34:46 +0100 Subject: [PATCH 3/3] docs: clarify conservative partial replay limits --- readme-dev.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/readme-dev.md b/readme-dev.md index c365e3769..fb434269a 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -262,10 +262,11 @@ failure or a permanently closed conversation. Replay follows both recorded spawn formats and replays available child transcripts once per conversation, including nested children, using legacy or paginated history. Announcements and all selected history precede the response. Completed snapshots reconcile tool and rendering state before duplicate live -payloads are suppressed; live bookkeeping still runs. In-progress tools retain classification without -resetting streamed output. Incomplete text has no safe delta cursor and is omitted rather than replayed -over newer buffered content. Agent-owned input is not rendered as human input. Missing, omitted, or -incomplete child history produces a parent advisory; an empty transcript alone does not invent a gap. +payloads are suppressed; live bookkeeping still runs. In-progress commands retain classification without +resetting streamed output. Incomplete text has no safe delta cursor; it and other records without a safe +partial representation are omitted rather than replayed over newer content. Agent-owned input is not +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