Do not replay rolled-back turns from history fallback

src/ResponseItemHistoryFallback.ts

@@ -80,12 +80,7 @@ export function parseResponseItemHistoryFallback(
         }
     };
 
-    for (const line of contents.split(/\r?\n/)) {
-        const record = parseJsonRecord(line);
-        if (!record) {
-            continue;
-        }
-
+    for (const record of activeHistoryRecords(contents)) {
         const eventMsgUpdates = createEventMsgUpdates(record);
         if (eventMsgUpdates) {
             pushUpdates(eventMsgUpdates);
@@ -152,6 +147,69 @@ export function parseResponseItemHistoryFallback(
     return recoveredFunctionCall ? updates : null;
 }
 
+function activeHistoryRecords(contents: string): JsonRecord[] {
+    const records: JsonRecord[] = [];
+    const turnStartIndices: number[] = [];
+    let hasExplicitTurnStart = false;
+
+    for (const line of contents.split(/\r?\n/)) {
+        const record = parseJsonRecord(line);
+        if (!record) {
+            continue;
+        }
+
+        const payload = eventMsgPayload(record);
+        switch (payload?.["type"]) {
+            case "task_started":
+                turnStartIndices.push(records.length);
+                hasExplicitTurnStart = true;
+                break;
+            case "task_complete":
+                hasExplicitTurnStart = false;
+                break;
+            case "thread_rolled_back": {
+                const numTurns = positiveInteger(payload["num_turns"]);
+                if (numTurns !== null) {
+                    rollbackTurns(records, turnStartIndices, numTurns);
+                }
+                hasExplicitTurnStart = false;
+                continue;
+            }
+            case "user_message":
+                if (!hasExplicitTurnStart) {
+                    turnStartIndices.push(records.length);
+                }
+                break;
+        }
+
+        records.push(record);
+    }
+
+    return records;
+}
+
+function eventMsgPayload(record: JsonRecord): JsonRecord | null {
+    return record["type"] === "event_msg" ? asRecord(record["payload"]) : null;
+}
+
+function positiveInteger(value: unknown): number | null {
+    return typeof value === "number" && Number.isInteger(value) && value > 0 ? value : null;
+}
+
+function rollbackTurns(records: JsonRecord[], turnStartIndices: number[], numTurns: number): void {
+    let retainedLength = records.length;
+    for (let index = 0; index < numTurns; index += 1) {
+        const turnStart = turnStartIndices.pop();
+        if (turnStart === undefined) {
+            retainedLength = 0;
+            turnStartIndices.length = 0;
+            break;
+        }
+        retainedLength = turnStart;
+    }
+    records.length = retainedLength;
+}
+
 function toolCallIdsFromThread(thread: Thread): Set<string> {
     const ids = new Set<string>();
     for (const turn of thread.turns) {

src/__tests__/CodexACPAgent/response-item-history-fallback.test.ts

@@ -5,6 +5,42 @@ import { parseResponseItemHistoryFallback } from "../../ResponseItemHistoryFallb
 type ToolCallUpdate = Extract<UpdateSessionEvent, { sessionUpdate: "tool_call_update" }>;
 
 describe("ResponseItemHistoryFallback", () => {
+    it("does not replay turns removed by thread rollback", () => {
+        const updates = parseResponseItemHistoryFallback(jsonl([
+            taskStarted("turn-obsolete"),
+            userMessage("unfinished prompt"),
+            assistantMessage("obsolete answer"),
+            functionCall("call-obsolete", "pwd"),
+            functionCallOutput("call-obsolete", "/workspace\n"),
+            taskComplete("turn-obsolete"),
+            threadRolledBack(1),
+            taskStarted("turn-current"),
+            userMessage("edited prompt"),
+            assistantMessage("current answer"),
+            functionCall("call-current", "pwd"),
+            functionCallOutput("call-current", "/workspace\n"),
+        ]), "terminal_output");
+
+        expect(messageTexts(updates, "user_message_chunk")).toEqual(["edited prompt"]);
+        expect(messageTexts(updates, "agent_message_chunk")).toEqual(["current answer"]);
+        expect(toolCallIds(updates)).toEqual(["call-current"]);
+    });
+
+    it("uses user messages as turn boundaries for legacy history", () => {
+        const updates = parseResponseItemHistoryFallback(jsonl([
+            userMessage("unfinished prompt"),
+            assistantMessage("obsolete answer"),
+            threadRolledBack(1),
+            userMessage("edited prompt"),
+            assistantMessage("current answer"),
+            functionCall("call-current", "pwd"),
+            functionCallOutput("call-current", "/workspace\n"),
+        ]), "terminal_output");
+
+        expect(messageTexts(updates, "user_message_chunk")).toEqual(["edited prompt"]);
+        expect(messageTexts(updates, "agent_message_chunk")).toEqual(["current answer"]);
+    });
+
     it("recovers only missing function calls for mixed parsed histories", () => {
         const updates = parseResponseItemHistoryFallback(jsonl([
             functionCall("call-existing", "rg \"Existing\" src"),
@@ -129,6 +165,46 @@ function functionCallOutput(callId: string, output: string): unknown {
     };
 }
 
+function taskStarted(turnId: string): unknown {
+    return {
+        type: "event_msg",
+        payload: { type: "task_started", turn_id: turnId },
+    };
+}
+
+function taskComplete(turnId: string): unknown {
+    return {
+        type: "event_msg",
+        payload: { type: "task_complete", turn_id: turnId },
+    };
+}
+
+function threadRolledBack(numTurns: number): unknown {
+    return {
+        type: "event_msg",
+        payload: { type: "thread_rolled_back", num_turns: numTurns },
+    };
+}
+
+function userMessage(message: string): unknown {
+    return {
+        type: "event_msg",
+        payload: { type: "user_message", message, images: [], local_images: [] },
+    };
+}
+
+function assistantMessage(text: string): unknown {
+    return {
+        type: "response_item",
+        payload: {
+            type: "message",
+            role: "assistant",
+            content: [{ type: "output_text", text }],
+            phase: "final_answer",
+        },
+    };
+}
+
 function toolCallIds(updates: UpdateSessionEvent[] | null): string[] {
     return (updates ?? [])
         .filter((update): update is Extract<UpdateSessionEvent, { sessionUpdate: "tool_call" }> => (
@@ -154,6 +230,17 @@ function thoughtTexts(updates: UpdateSessionEvent[] | null): string[] {
         .flatMap((update) => update.content.type === "text" ? [update.content.text] : []);
 }
 
+function messageTexts(
+    updates: UpdateSessionEvent[] | null,
+    kind: "user_message_chunk" | "agent_message_chunk",
+): string[] {
+    return (updates ?? []).flatMap((update) => (
+        update.sessionUpdate === kind && update.content.type === "text"
+            ? [update.content.text]
+            : []
+    ));
+}
+
 function agentMessageMetas(updates: UpdateSessionEvent[] | null): unknown[] {
     return (updates ?? [])
         .filter((update): update is Extract<UpdateSessionEvent, { sessionUpdate: "agent_message_chunk" }> => (