diff --git a/src/ResponseItemHistoryFallback.ts b/src/ResponseItemHistoryFallback.ts index 7292df20..e6c54577 100644 --- a/src/ResponseItemHistoryFallback.ts +++ b/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 { const ids = new Set(); for (const turn of thread.turns) { diff --git a/src/__tests__/CodexACPAgent/response-item-history-fallback.test.ts b/src/__tests__/CodexACPAgent/response-item-history-fallback.test.ts index aaeef097..3d63e013 100644 --- a/src/__tests__/CodexACPAgent/response-item-history-fallback.test.ts +++ b/src/__tests__/CodexACPAgent/response-item-history-fallback.test.ts @@ -5,6 +5,42 @@ import { parseResponseItemHistoryFallback } from "../../ResponseItemHistoryFallb type ToolCallUpdate = Extract; 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 => ( @@ -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 => (