Load full thread history on session resume (#208)

* Load full thread history on session resume

Closes #206

* Preserve failed tool call status in history fallback

* Restore agent reasoning from event history

* Test thread read during session load

* Merge fallback history with parsed thread updates

Author: benbrandt

src/CodexAcpClient.ts

@@ -237,14 +237,18 @@ export class CodexAcpClient {
             threadId: request.sessionId,
         });
         onSubscribed?.();
+        const historyResponse = await this.codexClient.threadRead({
+            threadId: response.thread.id,
+            includeTurns: true,
+        });
         const codexModels = await this.fetchAvailableModels();
         const currentModelId = this.createModelId(codexModels, response.model, response.reasoningEffort).toString();
         return {
             sessionId: request.sessionId,
             currentModelId: currentModelId,
             models: codexModels,
             currentServiceTier: response.serviceTier as ServiceTier ?? null,
-            thread: response.thread,
+            thread: historyResponse.thread,
             additionalDirectories,
         };
     }

src/CodexAcpServer.ts

@@ -33,6 +33,7 @@ import {CodexCommands} from "./CodexCommands";
 import type {QuotaMeta} from "./QuotaMeta";
 import {logger} from "./Logger";
 import {sanitizeMcpServerName} from "./McpServerName";
+import {createResponseItemHistoryFallbackUpdates} from "./ResponseItemHistoryFallback";
 import {
     type LegacyLoadSessionResponse,
     type LegacyNewSessionResponse,
@@ -44,6 +45,7 @@ import {
     LEGACY_SET_SESSION_MODEL_METHOD,
 } from "./AcpExtensions";
 import {
+    createCommandExecutionCompleteUpdate,
     createCommandExecutionUpdate,
     createDynamicToolCallUpdate,
     createFileChangeUpdate,
@@ -853,17 +855,29 @@ export class CodexAcpServer {
 
     private async streamThreadHistory(sessionId: string, thread: Thread): Promise<void> {
         const session = new ACPSessionConnection(this.connection, sessionId);
+        const sessionState = this.getSessionState(sessionId);
+        const responseItemFallbackUpdates = await createResponseItemHistoryFallbackUpdates(
+            thread,
+            sessionState.terminalOutputMode,
+        );
+
+        const threadUpdates: UpdateSessionEvent[] = [];
         for (const turn of thread.turns) {
             for (const item of turn.items) {
-                const updates = await this.createHistoryUpdates(item);
-                for (const update of updates) {
-                    await session.update(update);
-                }
+                const updates = await this.createHistoryUpdates(item, sessionState);
+                threadUpdates.push(...updates);
             }
         }
+
+        const updates = responseItemFallbackUpdates
+            ? mergeHistoryUpdates(responseItemFallbackUpdates, threadUpdates)
+            : threadUpdates;
+        for (const update of updates) {
+            await session.update(update);
+        }
     }
 
-    private async createHistoryUpdates(item: ThreadItem): Promise<UpdateSessionEvent[]> {
+    private async createHistoryUpdates(item: ThreadItem, sessionState: SessionState): Promise<UpdateSessionEvent[]> {
         switch (item.type) {
             case "userMessage":
                 return this.createUserMessageUpdates(item);
@@ -880,8 +894,14 @@ export class CodexAcpServer {
                 return this.createReasoningUpdates(item);
             case "fileChange":
                 return [await createFileChangeUpdate(item)];
-            case "commandExecution":
-                return [await createCommandExecutionUpdate(item)];
+            case "commandExecution": {
+                const updates = [await createCommandExecutionUpdate(item)];
+                const completeUpdate = createCommandExecutionCompleteUpdate(item, sessionState.terminalOutputMode);
+                if (completeUpdate) {
+                    updates.push(completeUpdate);
+                }
+                return updates;
+            }
             case "mcpToolCall":
                 return [await createMcpToolCallUpdate(item)];
             case "dynamicToolCall":
@@ -1481,6 +1501,71 @@ export class CodexAcpServer {
     }
 }
 
+function mergeHistoryUpdates(
+    responseItemFallbackUpdates: UpdateSessionEvent[],
+    threadUpdates: UpdateSessionEvent[],
+): UpdateSessionEvent[] {
+    const merged: UpdateSessionEvent[] = [];
+    const seen = new Set<string>();
+    let fallbackIndex = 0;
+
+    const pushUpdate = (update: UpdateSessionEvent) => {
+        const key = historyUpdateKey(update);
+        if (key && seen.has(key)) {
+            return;
+        }
+        if (key) {
+            seen.add(key);
+        }
+        merged.push(update);
+    };
+
+    const flushFallbackThrough = (targetKey: string): boolean => {
+        const matchIndex = responseItemFallbackUpdates.findIndex((update, index) => (
+            index >= fallbackIndex && historyUpdateKey(update) === targetKey
+        ));
+        if (matchIndex === -1) {
+            return false;
+        }
+
+        while (fallbackIndex <= matchIndex) {
+            pushUpdate(responseItemFallbackUpdates[fallbackIndex]!);
+            fallbackIndex += 1;
+        }
+        return true;
+    };
+
+    for (const update of threadUpdates) {
+        const key = historyUpdateKey(update);
+        if (key && flushFallbackThrough(key)) {
+            continue;
+        }
+        pushUpdate(update);
+    }
+
+    while (fallbackIndex < responseItemFallbackUpdates.length) {
+        pushUpdate(responseItemFallbackUpdates[fallbackIndex]!);
+        fallbackIndex += 1;
+    }
+
+    return merged;
+}
+
+function historyUpdateKey(update: UpdateSessionEvent): string | null {
+    switch (update.sessionUpdate) {
+        case "user_message_chunk":
+        case "agent_message_chunk":
+        case "agent_thought_chunk":
+            return `${update.sessionUpdate}:${JSON.stringify(update.content)}`;
+        case "tool_call":
+            return `tool_call:${update.toolCallId}:start`;
+        case "tool_call_update":
+            return `tool_call:${update.toolCallId}:update`;
+        default:
+            return null;
+    }
+}
+
 function getRequestedMcpServerNames(mcpServers: Array<acp.McpServer>): Array<string> {
     return Array.from(new Set(mcpServers.map(server => sanitizeMcpServerName(server.name))));
 }

src/CodexToolCallMapper.ts

@@ -27,6 +27,10 @@ import type {
 } from "./app-server/v2";
 import type { JsonValue } from "./app-server/serde_json/JsonValue";
 import {logger} from "./Logger";
+import {
+    createTerminalOutputMeta,
+    type TerminalOutputMode,
+} from "./TerminalOutputMode";
 
 type CodexItemStatus = CommandExecutionStatus | PatchApplyStatus | McpToolCallStatus | DynamicToolCallStatus;
 type AcpToolCallStatus = "pending" | "in_progress" | "completed" | "failed";
@@ -87,6 +91,47 @@ export async function createCommandExecutionUpdate(item: CommandExecutionItem):
     }, item.id, item.cwd);
 }
 
+export function createCommandExecutionCompleteUpdate(
+    item: CommandExecutionItem,
+    terminalOutputMode: TerminalOutputMode,
+): UpdateSessionEvent | null {
+    if (item.status === "inProgress") {
+        return null;
+    }
+
+    const update: UpdateSessionEvent = {
+        sessionUpdate: "tool_call_update",
+        toolCallId: item.id,
+        status: item.status === "completed" ? "completed" : "failed",
+        rawOutput: {
+            formatted_output: item.aggregatedOutput ?? "",
+            exit_code: item.exitCode,
+        },
+    };
+
+    if (!commandExecutionUsesTerminalOutput(item)) {
+        return update;
+    }
+
+    const terminalMeta: Record<string, unknown> = {};
+    if (item.aggregatedOutput) {
+        Object.assign(
+            terminalMeta,
+            createTerminalOutputMeta(terminalOutputMode, item.id, item.aggregatedOutput),
+        );
+    }
+    terminalMeta["terminal_exit"] = {
+        exit_code: item.exitCode,
+        signal: null,
+        terminal_id: item.id,
+    };
+
+    return {
+        ...update,
+        _meta: terminalMeta,
+    };
+}
+
 export async function createMcpToolCallUpdate(
     item: ThreadItem & { type: "mcpToolCall" }
 ): Promise<UpdateSessionEvent> {
@@ -334,12 +379,12 @@ export function formatWebSearchTitle(item: WebSearchItem): string {
     }
 }
 
-function createCommandActionEvent(
+export function createCommandActionEvent(
     id: string,
     status: CommandExecutionStatus,
     cwd: string,
     commandAction: CommandAction
-): UpdateSessionEvent {
+): AcpToolCallEvent {
     const acpStatus = toAcpStatus(status);
     switch (commandAction.type) {
         case "read":
@@ -395,7 +440,7 @@ function createTerminalCommandEvent(
     event: AcpToolCallEvent,
     terminalId: string,
     cwd: string,
-): UpdateSessionEvent {
+): AcpToolCallEvent {
     const { rawInput, ...eventWithoutRawInput } = event;
     return {
         ...eventWithoutRawInput,