feat: replay recorded subagent conversations

readme-dev.md

@@ -259,9 +259,18 @@ 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 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
+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.
 

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<RecordedSubagentItem[]>}> {
+        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<Thread | null> {
         return this.subagents.restoreChild(rootSessionId, parentThreadId, thread);
     }
@@ -1794,3 +1807,33 @@ async function* noItems<T>(): AsyncGenerator<T[]> {}
 async function* oneItemPage<T>(items: T[]): AsyncGenerator<T[]> {
     if (items.length > 0) yield items;
 }
+
+async function* legacySubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator<RecordedSubagentItem[]> {
+    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"),
+    })));
+}
+
+async function* paginatedSubagentItems(client: CodexAppServerClient, threadId: string): AsyncGenerator<RecordedSubagentItem[]> {
+    for await (const page of client.threadItemEntryPages(threadId)) {
+        yield page.map(({item, turnId, completedAtMs}) => ({
+            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;
+    }
+}

src/CodexAcpServer.ts

@@ -15,6 +15,7 @@ import {
     OPENAI_PROVIDER_ID,
     type SessionMetadata,
     type SessionMetadataWithThread,
+    type RecordedSubagentItem,
     type UrlElicitationRequester
 } from "./CodexAcpClient";
 import {
@@ -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<ThreadItem[]>, state: SessionState, isOpen: () => boolean,
+    /** Recorded conversation data is independent of verified runtime ownership and current work. */
+    private async streamStandardHistory(
+        sessionId: string, pages: AsyncIterable<RecordedSubagentItem[]>, state: SessionState, isOpen: () => boolean,
+        threadId = sessionId, ancestry = new Set([threadId]), replayed = new Set<string>(), gaps = new Set<string>(),
     ): Promise<void> {
         const router = state.subagents;
         if (!(router instanceof StandardSubagentEventRouter)) return;
         const session = new ACPSessionConnection(this.connection, sessionId);
         const started = new Set<string>();
-        const restored = new Set<string>();
-        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<ThreadItem & {type: "subAgentActivity"}> = [];
+        const replayActivity = async (item: ThreadItem & {type: "subAgentActivity"}): Promise<boolean> => {
+            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<ThreadItem & {type: "subAgentActivity"}> = [];
         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 && !supportsPartialHistory(item))) {
+                        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,24 @@ async function* itemsOfEntries(pages: AsyncIterable<ThreadItemEntry[]>): AsyncGe
     }
 }
 
+async function* standardHistoryItems(pages: AsyncIterable<ThreadItem[]>): AsyncGenerator<RecordedSubagentItem[]> {
+    for await (const page of pages) {
+        yield page.map(item => ({item, completed: !("status" in item) || item.status !== "inProgress"}));
+    }
+}
+
+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);
 }

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<Record<StructuredCodexErrorKind
     activeTurnNotSteerable: "provider_error",
 } satisfies Record<StructuredCodexErrorKind, CodexFailureKind>;
 
-/**
- * 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<ThreadItem["type"]>([
-    "fileChange",
-    "commandExecution",
-    "mcpToolCall",
-    "dynamicToolCall",
-    "webSearch",
-    "imageView",
-    "imageGeneration",
-]);
-
 /** Prompt failures whose text the client has already received as an agent message. */
 const failuresShownAsMessages = new WeakSet<RequestError>();
 

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<ThreadItem["type"]>([
+    "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();

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"
+    }
+  }
+]