From 2e2df0c8cc544b6c4acd6745a83c1da56c09769d Mon Sep 17 00:00:00 2001 From: Roomote Date: Fri, 21 Aug 2026 03:21:59 +0000 Subject: [PATCH 1/8] fix(task): preserve history during resume hydration --- .../__tests__/taskMessages.spec.ts | 31 ++++--- src/core/task-persistence/index.ts | 7 +- src/core/task-persistence/taskMessages.ts | 59 +++++++----- src/core/task/Task.ts | 37 ++++---- .../task/__tests__/Task.persistence.spec.ts | 90 +++++++++++++++++-- .../Task.resume-eviction-race.spec.ts | 9 +- src/core/webview/ClineProvider.ts | 7 +- 7 files changed, 179 insertions(+), 61 deletions(-) diff --git a/src/core/task-persistence/__tests__/taskMessages.spec.ts b/src/core/task-persistence/__tests__/taskMessages.spec.ts index c6bc360c05..6956fe667d 100644 --- a/src/core/task-persistence/__tests__/taskMessages.spec.ts +++ b/src/core/task-persistence/__tests__/taskMessages.spec.ts @@ -68,7 +68,7 @@ describe("taskMessages.saveTaskMessages", () => { }) describe("taskMessages.readTaskMessages", () => { - it("returns empty array when file contains invalid JSON", async () => { + it("rejects invalid JSON without treating it as empty history", async () => { const taskId = "task-corrupt-json" // Manually create the task directory and write corrupted JSON const taskDir = path.join(tmpBaseDir, "tasks", taskId) @@ -76,26 +76,35 @@ describe("taskMessages.readTaskMessages", () => { const filePath = path.join(taskDir, "ui_messages.json") await fs.writeFile(filePath, "{not valid json!!!", "utf8") - const result = await readTaskMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readTaskMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) - - expect(result).toEqual([]) }) - it("returns [] when file contains valid JSON that is not an array", async () => { + it("rejects valid non-array JSON without treating it as empty history", async () => { const taskId = "task-non-array-json" const taskDir = path.join(tmpBaseDir, "tasks", taskId) await fs.mkdir(taskDir, { recursive: true }) const filePath = path.join(taskDir, "ui_messages.json") await fs.writeFile(filePath, JSON.stringify("hello"), "utf8") - const result = await readTaskMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readTaskMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) + }) + + it("distinguishes a missing history file from an empty history", async () => { + await expect(readTaskMessages({ taskId: "task-missing", globalStoragePath: tmpBaseDir })).rejects.toMatchObject( + { kind: "not_found" }, + ) + }) + + it("returns an explicitly persisted empty history", async () => { + const taskId = "task-empty" + const taskDir = path.join(tmpBaseDir, "tasks", taskId) + await fs.mkdir(taskDir, { recursive: true }) + await fs.writeFile(path.join(taskDir, "ui_messages.json"), "[]", "utf8") - expect(result).toEqual([]) + await expect(readTaskMessages({ taskId, globalStoragePath: tmpBaseDir })).resolves.toEqual([]) }) }) diff --git a/src/core/task-persistence/index.ts b/src/core/task-persistence/index.ts index 463df8a0bb..5dda196592 100644 --- a/src/core/task-persistence/index.ts +++ b/src/core/task-persistence/index.ts @@ -1,5 +1,10 @@ export { type ApiMessage, readApiMessages, saveApiMessages } from "./apiMessages" -export { readTaskMessages, saveTaskMessages } from "./taskMessages" +export { + readTaskMessages, + saveTaskMessages, + TaskMessagesReadError, + type TaskMessagesReadErrorKind, +} from "./taskMessages" export { taskMetadata } from "./taskMetadata" export { TaskHistoryStore } from "./TaskHistoryStore" export { diff --git a/src/core/task-persistence/taskMessages.ts b/src/core/task-persistence/taskMessages.ts index cee66432d9..900335ed04 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -4,11 +4,22 @@ import * as fs from "fs/promises" import type { ClineMessage } from "@roo-code/types" -import { fileExistsAtPath } from "../../utils/fs" - import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" +export type TaskMessagesReadErrorKind = "not_found" | "invalid" | "io_error" + +export class TaskMessagesReadError extends Error { + constructor( + public readonly kind: TaskMessagesReadErrorKind, + message: string, + public readonly originalError?: unknown, + ) { + super(message) + this.name = "TaskMessagesReadError" + } +} + export type ReadTaskMessagesOptions = { taskId: string globalStoragePath: string @@ -20,27 +31,33 @@ export async function readTaskMessages({ }: ReadTaskMessagesOptions): Promise { const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.uiMessages) - const fileExists = await fileExistsAtPath(filePath) - - if (fileExists) { - try { - const parsedData = JSON.parse(await fs.readFile(filePath, "utf8")) - if (!Array.isArray(parsedData)) { - console.warn( - `[readTaskMessages] Parsed data is not an array (got ${typeof parsedData}), returning empty. TaskId: ${taskId}, Path: ${filePath}`, - ) - return [] - } - return parsedData - } catch (error) { - console.warn( - `[readTaskMessages] Failed to parse ${filePath} for task ${taskId}, returning empty: ${error instanceof Error ? error.message : String(error)}`, - ) - return [] - } + + let fileContent: string + try { + fileContent = await fs.readFile(filePath, "utf8") + } catch (error) { + const kind = + typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT" + ? "not_found" + : "io_error" + throw new TaskMessagesReadError(kind, `Failed to read task messages for ${taskId} at ${filePath}`, error) + } + + let parsedData: unknown + try { + parsedData = JSON.parse(fileContent) + } catch (error) { + throw new TaskMessagesReadError("invalid", `Failed to parse task messages for ${taskId} at ${filePath}`, error) + } + + if (!Array.isArray(parsedData)) { + throw new TaskMessagesReadError( + "invalid", + `Task messages for ${taskId} at ${filePath} must be an array, got ${typeof parsedData}`, + ) } - return [] + return parsedData } export type SaveTaskMessagesOptions = { diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 4c2d77b5ae..8c88461ab0 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -1182,14 +1182,18 @@ export class Task extends EventEmitter implements TaskLike { } public async overwriteClineMessages(newMessages: ClineMessage[]) { - this.clineMessages = newMessages - restoreTodoListForTask(this) + this.hydrateClineMessages(newMessages) await this.saveClineMessages() + } + + private hydrateClineMessages(messages: ClineMessage[]) { + this.clineMessages = messages + restoreTodoListForTask(this) - // When overwriting messages (e.g., during task resume), repopulate the cloud sync tracking Set + // When hydrating or overwriting messages, repopulate the cloud sync tracking Set // with timestamps from all non-partial messages to prevent re-syncing previously synced messages this.cloudSyncedMessageTimestamps.clear() - for (const msg of newMessages) { + for (const msg of messages) { if (msg.partial !== true) { this.cloudSyncedMessageTimestamps.add(msg.ts) } @@ -2108,7 +2112,11 @@ export class Task extends EventEmitter implements TaskLike { private async resumeTaskFromHistory() { try { - const modifiedClineMessages = await this.getSavedClineMessages() + const modifiedClineMessages = [...(await this.getSavedClineMessages())] + + if (this.abort || this.abandoned) { + return + } // Remove any resume messages that may have been added before. const lastRelevantMessageIndex = findLastIndex( @@ -2120,16 +2128,6 @@ export class Task extends EventEmitter implements TaskLike { modifiedClineMessages.splice(lastRelevantMessageIndex + 1) } - // Remove any trailing reasoning-only UI messages that were not part of the persisted API conversation - while (modifiedClineMessages.length > 0) { - const last = modifiedClineMessages[modifiedClineMessages.length - 1] - if (last.type === "say" && last.say === "reasoning") { - modifiedClineMessages.pop() - } else { - break - } - } - if (this.pendingAction) { const pendingAskIndex = findLastIndex( modifiedClineMessages, @@ -2162,8 +2160,9 @@ export class Task extends EventEmitter implements TaskLike { } } - await this.overwriteClineMessages(modifiedClineMessages) - this.clineMessages = await this.getSavedClineMessages() + // Avoid a standalone write during hydration. The resume ask will persist only + // after all history reads succeed and the task is still active. + this.hydrateClineMessages(modifiedClineMessages) // Now present the cline messages to the user and ask if they want to // resume (NOTE: we ran into a bug before where the @@ -2193,6 +2192,10 @@ export class Task extends EventEmitter implements TaskLike { return } + if (this.abort || this.abandoned) { + return + } + const lastClineMessage = this.clineMessages .slice() .reverse() diff --git a/src/core/task/__tests__/Task.persistence.spec.ts b/src/core/task/__tests__/Task.persistence.spec.ts index 671bd7d4b7..46fe4c9706 100644 --- a/src/core/task/__tests__/Task.persistence.spec.ts +++ b/src/core/task/__tests__/Task.persistence.spec.ts @@ -767,13 +767,10 @@ describe("Task persistence", () => { expect(replay).toHaveBeenCalledWith(pendingAction) expect(ask).not.toHaveBeenCalled() - expect(mockSaveTaskMessages).toHaveBeenCalledWith( - expect.objectContaining({ - messages: expect.not.arrayContaining([ - expect.objectContaining({ text: pendingAction.approvalText }), - ]), - }), + expect(task.clineMessages).not.toEqual( + expect.arrayContaining([expect.objectContaining({ text: pendingAction.approvalText })]), ) + expect(mockSaveTaskMessages).not.toHaveBeenCalled() }) it("reconciles an already-persisted tool result before generic resume", async () => { @@ -1068,6 +1065,87 @@ describe("Task persistence", () => { }) }) + describe("resumeTaskFromHistory", () => { + it.each(["not_found", "invalid", "io_error"] as const)( + "does not persist when hydration fails with %s", + async (kind) => { + mockReadTaskMessages.mockRejectedValue(Object.assign(new Error(`history ${kind}`), { kind })) + + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + historyItem: { + id: `issue-1279-${kind}`, + number: 1, + ts: 1, + task: "Original task", + status: "completed", + tokensIn: 10, + tokensOut: 5, + totalCost: 0.001, + }, + initialStatus: "completed", + startTask: false, + }) + const askSpy = vi.spyOn(task, "ask") + + await expect(getTaskPersistenceAccess(task).resumeTaskFromHistory()).rejects.toThrow(`history ${kind}`) + await task.abortTask(true) + + expect(askSpy).not.toHaveBeenCalled() + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + expect(mockProvider.updateTaskHistory).not.toHaveBeenCalled() + }, + ) + + it("preserves finalized trailing reasoning without rewriting history during hydration", async () => { + const messages = [ + { ts: 1, type: "say" as const, say: "text" as const, text: "Original task" }, + { ts: 2, type: "say" as const, say: "completion_result" as const, text: "Initial result" }, + { ts: 3, type: "ask" as const, ask: "resume_completed_task" as const }, + { ts: 4, type: "say" as const, say: "user_feedback" as const, text: "Continue investigating" }, + { + ts: 5, + type: "say" as const, + say: "reasoning" as const, + text: "Critical current conclusion", + partial: false, + }, + ] + mockReadTaskMessages.mockResolvedValue(messages) + mockReadApiMessages.mockResolvedValue([ + { role: "user", content: [{ type: "text", text: "Continue investigating" }] }, + ]) + + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + historyItem: { + id: "issue-1279-current", + number: 1, + ts: 5, + task: "Original task", + status: "completed", + tokensIn: 10, + tokensOut: 5, + totalCost: 0.001, + }, + initialStatus: "completed", + startTask: false, + }) + vi.spyOn(task, "ask").mockImplementation(async (type) => { + expect(type).toBe("resume_completed_task") + expect(task.clineMessages).toContainEqual( + expect.objectContaining({ text: "Critical current conclusion", partial: false }), + ) + throw new Error("stop after hydration") + }) + + await expect(getTaskPersistenceAccess(task).resumeTaskFromHistory()).rejects.toThrow("stop after hydration") + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + }) + }) + // ── flushPendingToolResultsToHistory — save failure/success ─────────── describe("flushPendingToolResultsToHistory persistence", () => { diff --git a/src/core/task/__tests__/Task.resume-eviction-race.spec.ts b/src/core/task/__tests__/Task.resume-eviction-race.spec.ts index 8766f38d5b..334fd4c02e 100644 --- a/src/core/task/__tests__/Task.resume-eviction-race.spec.ts +++ b/src/core/task/__tests__/Task.resume-eviction-race.spec.ts @@ -185,9 +185,7 @@ describe("Task resume/eviction race (Work #1 (no message) regression)", () => { // Hold the disk read open so the task is aborted while clineMessages is // still empty — the same window a user hits by navigating away quickly. const readDeferred = createDeferred() - mockReadTaskMessages - .mockReturnValueOnce(readDeferred.promise) // first read: held open to simulate the race window - .mockResolvedValue([]) // second read (resumeTaskFromHistory:2023): post-abort, safe fallback + mockReadTaskMessages.mockReturnValueOnce(readDeferred.promise) const updateTaskHistory = vi.fn().mockResolvedValue([]) const mockProvider = makeMockProvider(updateTaskHistory) @@ -222,5 +220,10 @@ describe("Task resume/eviction race (Work #1 (no message) regression)", () => { { ts: historyItem.ts + 1, type: "say", say: "completion_result", text: "Done." }, ]) await runPromise + + // The abandoned hydration must not resume and persist after its read settles. + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + expect(updateTaskHistory).not.toHaveBeenCalled() + expect(mockReadTaskMessages).toHaveBeenCalledTimes(1) }) }) diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 394da7c10f..6f1ccfeb12 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -4063,8 +4063,11 @@ export class ClineProvider taskId: parentTaskId, globalStoragePath, }) - } catch { - parentClineMessages = [] + } catch (error) { + this.log( + `[reopenParentFromDelegation] Failed to read messages for parent ${parentTaskId}: ${error instanceof Error ? error.message : String(error)}`, + ) + return false } let parentApiMessages: any[] = [] From fe015371d787b1812ef8941397aa84adec10faaf Mon Sep 17 00:00:00 2001 From: Roomote Date: Sat, 22 Aug 2026 13:28:32 +0000 Subject: [PATCH 2/8] fix(task): merge concurrent history snapshots --- .../history-resume-delegation.spec.ts | 36 ++++++ .../__tests__/apiMessages.spec.ts | 35 +++++- .../__tests__/mergeMessageSnapshots.spec.ts | 106 ++++++++++++++++++ .../__tests__/taskMessages.spec.ts | 64 ++++++++++- src/core/task-persistence/apiMessages.ts | 5 +- .../task-persistence/mergeMessageSnapshots.ts | 81 +++++++++++++ src/core/task-persistence/taskMessages.ts | 41 +++++-- src/core/task/Task.ts | 10 +- .../task/__tests__/Task.persistence.spec.ts | 59 ++++++++++ src/core/webview/ClineProvider.ts | 14 ++- 10 files changed, 435 insertions(+), 16 deletions(-) create mode 100644 src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts create mode 100644 src/core/task-persistence/mergeMessageSnapshots.ts diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index d3a24a3140..48410feb37 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -354,6 +354,7 @@ describe("History resume delegation - parent metadata transitions", () => { ]), taskId: "p1", globalStoragePath: "/storage", + merge: true, }), ) @@ -373,6 +374,7 @@ describe("History resume delegation - parent metadata transitions", () => { ]), taskId: "p1", globalStoragePath: "/storage", + merge: true, }), ) @@ -384,6 +386,40 @@ describe("History resume delegation - parent metadata transitions", () => { expect(apiCall.messages).toHaveLength(2) // 1 original + 1 injected }) + it("does not reopen or overwrite a parent when its UI history cannot be read", async () => { + const parentItem = { + id: "parent-read-failure", + status: "delegated", + awaitingChildId: "child-read-failure", + childIds: ["child-read-failure"], + ts: 100, + task: "Parent", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + const log = vi.fn() + const provider = makeProviderStub({ + contextProxy: { globalStorageUri: { fsPath: "/storage" } }, + getTaskWithId: vi.fn().mockResolvedValue({ historyItem: parentItem }), + getCurrentTask: vi.fn(() => ({ taskId: "child-read-failure" })), + taskHistoryStore: makeTaskHistoryStoreStub({ id: "child-read-failure", status: "active" }, parentItem), + log, + }) + vi.mocked(readTaskMessages).mockRejectedValue(new Error("history unavailable")) + + const result = await ClineProvider.prototype.reopenParentFromDelegation.call(provider, { + parentTaskId: "parent-read-failure", + childTaskId: "child-read-failure", + completionResultSummary: "Child done", + }) + + expect(result).toBe(false) + expect(log).toHaveBeenCalledWith(expect.stringContaining("history unavailable")) + expect(saveTaskMessages).not.toHaveBeenCalled() + expect(saveApiMessages).not.toHaveBeenCalled() + }) + it("reopenParentFromDelegation injects tool_result when new_task tool_use exists in API history", async () => { const parentItem = { id: "p-tool", diff --git a/src/core/task-persistence/__tests__/apiMessages.spec.ts b/src/core/task-persistence/__tests__/apiMessages.spec.ts index aa725f4744..b154cafaae 100644 --- a/src/core/task-persistence/__tests__/apiMessages.spec.ts +++ b/src/core/task-persistence/__tests__/apiMessages.spec.ts @@ -4,7 +4,7 @@ import * as os from "os" import * as path from "path" import * as fs from "fs/promises" -import { readApiMessages } from "../apiMessages" +import { readApiMessages, saveApiMessages } from "../apiMessages" let tmpBaseDir: string @@ -84,3 +84,36 @@ describe("apiMessages.readApiMessages", () => { expect(result).toEqual([]) }) }) + +describe("apiMessages.saveApiMessages", () => { + it("merges a concurrent disk suffix when requested", async () => { + const taskId = "task-merge-api" + const taskDir = path.join(tmpBaseDir, "tasks", taskId) + await fs.mkdir(taskDir, { recursive: true }) + const filePath = path.join(taskDir, "api_conversation_history.json") + await fs.writeFile( + filePath, + JSON.stringify([ + { role: "user", content: "disk prefix", ts: 1 }, + { role: "assistant", content: "disk suffix", ts: 3 }, + ]), + "utf8", + ) + + await saveApiMessages({ + taskId, + globalStoragePath: tmpBaseDir, + merge: true, + messages: [ + { role: "user", content: "updated prefix", ts: 1 }, + { role: "assistant", content: "incoming", ts: 2 }, + ], + }) + + expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual([ + expect.objectContaining({ content: "updated prefix", ts: 1 }), + expect.objectContaining({ content: "incoming", ts: 2 }), + expect.objectContaining({ content: "disk suffix", ts: 3 }), + ]) + }) +}) diff --git a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts new file mode 100644 index 0000000000..a9b1a31093 --- /dev/null +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -0,0 +1,106 @@ +import { mergeApiMessageSnapshots, mergeClineMessageSnapshots } from "../mergeMessageSnapshots" + +describe("mergeClineMessageSnapshots", () => { + it("preserves disk-only messages and applies incoming updates in timestamp order", () => { + const result = mergeClineMessageSnapshots( + [ + { ts: 1, type: "say", say: "text", text: "old" }, + { ts: 3, type: "say", say: "text", text: "newer disk suffix" }, + ], + [ + { ts: 1, type: "say", say: "text", text: "updated" }, + { ts: 2, type: "say", say: "text", text: "incoming" }, + ], + ) + + expect(result).toEqual([ + expect.objectContaining({ ts: 1, text: "updated" }), + expect.objectContaining({ ts: 2, text: "incoming" }), + expect.objectContaining({ ts: 3, text: "newer disk suffix" }), + ]) + }) + + it("does not regress completed or answered message state", () => { + const result = mergeClineMessageSnapshots( + [{ ts: 1, type: "ask", ask: "tool", partial: false, isAnswered: true }], + [{ ts: 1, type: "ask", ask: "tool", partial: true, isAnswered: false }], + ) + + expect(result).toEqual([expect.objectContaining({ ts: 1, partial: false, isAnswered: true })]) + }) + + it("uses the incoming message when a timestamp is reused for a different message identity", () => { + expect( + mergeClineMessageSnapshots( + [{ ts: 1, type: "say", say: "text", text: "old" }], + [{ ts: 1, type: "ask", ask: "followup", text: "new" }], + ), + ).toEqual([{ ts: 1, type: "ask", ask: "followup", text: "new" }]) + }) + + it("returns incoming data when either snapshot is not an array", () => { + expect(mergeClineMessageSnapshots(null, [{ ts: 1 }])).toEqual([{ ts: 1 }]) + expect(mergeClineMessageSnapshots([], "invalid")).toBe("invalid") + }) +}) + +describe("mergeApiMessageSnapshots", () => { + it("retains equal-timestamp records and keeps tool calls before their results", () => { + const result = mergeApiMessageSnapshots( + [ + { role: "assistant", content: "old", ts: 1 }, + { role: "user", content: "same timestamp sibling", ts: 1 }, + { role: "user", content: [{ type: "tool_result", tool_use_id: "call-1", content: "ok" }], ts: 3 }, + ], + [ + { role: "assistant", content: "updated", ts: 1 }, + { + role: "assistant", + content: [{ type: "tool_use", id: "call-1", name: "read_file", input: {} }], + ts: 2, + }, + ], + ) + + expect(result).toEqual([ + expect.objectContaining({ role: "assistant", content: "updated", ts: 1 }), + expect.objectContaining({ role: "user", content: "same timestamp sibling", ts: 1 }), + expect.objectContaining({ role: "assistant", ts: 2 }), + expect.objectContaining({ role: "user", ts: 3 }), + ]) + }) + + it("preserves only the unmatched legacy disk tail", () => { + const result = mergeApiMessageSnapshots( + [ + { role: "user", content: "old prefix" }, + { role: "assistant", content: "disk tail" }, + ], + [{ role: "user", content: "updated prefix" }], + ) + + expect(result).toEqual([ + { role: "user", content: "updated prefix" }, + { role: "assistant", content: "disk tail" }, + ]) + }) + + it("keeps legacy prefixes ahead of newer timestamped messages", () => { + const result = mergeApiMessageSnapshots( + [ + { role: "user", content: "legacy prefix" }, + { role: "assistant", content: "disk suffix", ts: 3 }, + ], + [ + { role: "user", content: "updated legacy prefix" }, + { role: "assistant", content: "incoming", ts: 2 }, + ], + ) + + expect(result).toEqual([ + { role: "user", content: "updated legacy prefix" }, + expect.objectContaining({ content: "incoming", ts: 2 }), + expect.objectContaining({ content: "disk suffix", ts: 3 }), + ]) + }) +}) diff --git a/src/core/task-persistence/__tests__/taskMessages.spec.ts b/src/core/task-persistence/__tests__/taskMessages.spec.ts index 6956fe667d..61cecdb353 100644 --- a/src/core/task-persistence/__tests__/taskMessages.spec.ts +++ b/src/core/task-persistence/__tests__/taskMessages.spec.ts @@ -1,11 +1,18 @@ -import { describe, it, expect, vi, beforeEach } from "vitest" +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest" import * as os from "os" import * as path from "path" import * as fs from "fs/promises" +import type { ClineMessage } from "@roo-code/types" + // Mocks (use hoisted to avoid initialization ordering issues) const hoisted = vi.hoisted(() => ({ safeWriteJsonMock: vi.fn().mockResolvedValue(undefined), + readFileMock: vi.fn(), +})) +vi.mock("fs/promises", async (importOriginal) => ({ + ...(await importOriginal()), + readFile: hoisted.readFileMock, })) vi.mock("../../../utils/safeWriteJson", () => ({ safeWriteJson: hoisted.safeWriteJsonMock, @@ -18,10 +25,17 @@ let tmpBaseDir: string beforeEach(async () => { hoisted.safeWriteJsonMock.mockClear() + const actualFs = await vi.importActual("fs/promises") + hoisted.readFileMock.mockReset().mockImplementation(actualFs.readFile) // Create a unique, writable temp directory to act as globalStoragePath tmpBaseDir = await fs.mkdtemp(path.join(os.tmpdir(), "roo-test-")) }) +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() +}) + describe("taskMessages.saveTaskMessages", () => { beforeEach(() => { hoisted.safeWriteJsonMock.mockClear() @@ -48,6 +62,7 @@ describe("taskMessages.saveTaskMessages", () => { expect(hoisted.safeWriteJsonMock).toHaveBeenCalledTimes(1) const [, persisted] = hoisted.safeWriteJsonMock.mock.calls[0] expect(persisted).toEqual(messages) + expect(hoisted.safeWriteJsonMock.mock.calls[0][2]).toBeUndefined() }) it("persists messages without modification when no metadata", async () => { @@ -65,6 +80,23 @@ describe("taskMessages.saveTaskMessages", () => { const [, persisted] = hoisted.safeWriteJsonMock.mock.calls[0] expect(persisted).toEqual(messages) }) + + it("passes the history merge callback only when requested", async () => { + const messages: ClineMessage[] = [{ ts: 2, type: "say", say: "text", text: "incoming" }] + await saveTaskMessages({ + messages, + taskId: "task-merge", + globalStoragePath: tmpBaseDir, + merge: true, + }) + + const merge = hoisted.safeWriteJsonMock.mock.calls[0][2]?.merge + expect(merge).toBeTypeOf("function") + expect(merge([{ ts: 1, type: "say", say: "text", text: "disk" }], messages)).toEqual([ + expect.objectContaining({ ts: 1, text: "disk" }), + expect.objectContaining({ ts: 2, text: "incoming" }), + ]) + }) }) describe("taskMessages.readTaskMessages", () => { @@ -107,4 +139,34 @@ describe("taskMessages.readTaskMessages", () => { await expect(readTaskMessages({ taskId, globalStoragePath: tmpBaseDir })).resolves.toEqual([]) }) + + it("retries one transient missing-file read after a jittered delay", async () => { + vi.spyOn(Math, "random").mockReturnValue(0) + const missing = Object.assign(new Error("missing"), { code: "ENOENT" }) + hoisted.readFileMock.mockRejectedValueOnce(missing).mockResolvedValueOnce("[]") + + await expect(readTaskMessages({ taskId: "task-retry", globalStoragePath: tmpBaseDir })).resolves.toEqual([]) + expect(hoisted.readFileMock).toHaveBeenCalledTimes(2) + }) + + it("throws when the missing-file retry also fails", async () => { + vi.spyOn(Math, "random").mockReturnValue(0) + const missing = Object.assign(new Error("missing"), { code: "ENOENT" }) + hoisted.readFileMock.mockRejectedValue(missing) + + await expect( + readTaskMessages({ taskId: "task-still-missing", globalStoragePath: tmpBaseDir }), + ).rejects.toMatchObject({ kind: "not_found" }) + expect(hoisted.readFileMock).toHaveBeenCalledTimes(2) + }) + + it("does not retry non-ENOENT read failures", async () => { + const denied = Object.assign(new Error("denied"), { code: "EACCES" }) + hoisted.readFileMock.mockRejectedValueOnce(denied) + + await expect(readTaskMessages({ taskId: "task-denied", globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "io_error", + }) + expect(hoisted.readFileMock).toHaveBeenCalledTimes(1) + }) }) diff --git a/src/core/task-persistence/apiMessages.ts b/src/core/task-persistence/apiMessages.ts index 7672f6f7ee..3fdcd376a5 100644 --- a/src/core/task-persistence/apiMessages.ts +++ b/src/core/task-persistence/apiMessages.ts @@ -8,6 +8,7 @@ import { fileExistsAtPath } from "../../utils/fs" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" +import { mergeApiMessageSnapshots } from "./mergeMessageSnapshots" export type ApiMessage = Anthropic.MessageParam & { ts?: number @@ -110,12 +111,14 @@ export async function saveApiMessages({ messages, taskId, globalStoragePath, + merge = false, }: { messages: ApiMessage[] taskId: string globalStoragePath: string + merge?: boolean }) { const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.apiConversationHistory) - await safeWriteJson(filePath, messages) + await safeWriteJson(filePath, messages, merge ? { merge: mergeApiMessageSnapshots } : undefined) } diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts new file mode 100644 index 0000000000..1b158fecff --- /dev/null +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -0,0 +1,81 @@ +type MessageRecord = Record & { ts?: unknown } + +function isRecord(value: unknown): value is MessageRecord { + return typeof value === "object" && value !== null +} + +function mergeTimestampedSnapshots( + existing: unknown, + incoming: unknown, + mergeMatch: (disk: MessageRecord, next: MessageRecord) => MessageRecord, +): unknown { + if (!Array.isArray(existing) || !Array.isArray(incoming)) { + return incoming + } + + const existingGroups = new Map() + const existingLegacy: unknown[] = [] + + for (const message of existing) { + if (isRecord(message) && typeof message.ts === "number") { + const group = existingGroups.get(message.ts) ?? [] + group.push(message) + existingGroups.set(message.ts, group) + } else { + existingLegacy.push(message) + } + } + + const consumedByTimestamp = new Map() + let incomingLegacyCount = 0 + const merged = incoming.map((message) => { + if (!isRecord(message) || typeof message.ts !== "number") { + incomingLegacyCount++ + return message + } + + const consumed = consumedByTimestamp.get(message.ts) ?? 0 + consumedByTimestamp.set(message.ts, consumed + 1) + const diskMessage = existingGroups.get(message.ts)?.[consumed] + return diskMessage ? mergeMatch(diskMessage, message) : message + }) + + const diskOnlyTimestamped = [...existingGroups.entries()].flatMap(([timestamp, messages]) => + messages.slice(consumedByTimestamp.get(timestamp) ?? 0), + ) + + for (const diskMessage of diskOnlyTimestamped) { + const insertionIndex = merged.findIndex( + (message) => isRecord(message) && typeof message.ts === "number" && message.ts > (diskMessage.ts as number), + ) + if (insertionIndex === -1) { + merged.push(diskMessage) + } else { + merged.splice(insertionIndex, 0, diskMessage) + } + } + + merged.push(...existingLegacy.slice(incomingLegacyCount)) + return merged +} + +export function mergeClineMessageSnapshots(existing: unknown, incoming: unknown): unknown { + return mergeTimestampedSnapshots(existing, incoming, (disk, next) => { + if (disk.type !== next.type || disk.say !== next.say || disk.ask !== next.ask) { + return next + } + + const merged = { ...disk, ...next } + if (disk.partial === false && next.partial === true) { + merged.partial = false + } + if (disk.isAnswered === true) { + merged.isAnswered = true + } + return merged + }) +} + +export function mergeApiMessageSnapshots(existing: unknown, incoming: unknown): unknown { + return mergeTimestampedSnapshots(existing, incoming, (_disk, next) => next) +} diff --git a/src/core/task-persistence/taskMessages.ts b/src/core/task-persistence/taskMessages.ts index 900335ed04..13b0ae728f 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -6,6 +6,7 @@ import type { ClineMessage } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" +import { mergeClineMessageSnapshots } from "./mergeMessageSnapshots" export type TaskMessagesReadErrorKind = "not_found" | "invalid" | "io_error" @@ -25,6 +26,29 @@ export type ReadTaskMessagesOptions = { globalStoragePath: string } +const READ_RETRY_MIN_MS = 10 +const READ_RETRY_RANGE_MS = 291 + +function getErrorCode(error: unknown): string | undefined { + return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" + ? error.code + : undefined +} + +async function readFileWithMissingRetry(filePath: string): Promise { + try { + return await fs.readFile(filePath, "utf8") + } catch (error) { + if (getErrorCode(error) !== "ENOENT") { + throw error + } + + const retryDelay = READ_RETRY_MIN_MS + Math.floor(Math.random() * READ_RETRY_RANGE_MS) + await new Promise((resolve) => setTimeout(resolve, retryDelay)) + return fs.readFile(filePath, "utf8") + } +} + export async function readTaskMessages({ taskId, globalStoragePath, @@ -34,12 +58,9 @@ export async function readTaskMessages({ let fileContent: string try { - fileContent = await fs.readFile(filePath, "utf8") + fileContent = await readFileWithMissingRetry(filePath) } catch (error) { - const kind = - typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT" - ? "not_found" - : "io_error" + const kind = getErrorCode(error) === "ENOENT" ? "not_found" : "io_error" throw new TaskMessagesReadError(kind, `Failed to read task messages for ${taskId} at ${filePath}`, error) } @@ -64,10 +85,16 @@ export type SaveTaskMessagesOptions = { messages: ClineMessage[] taskId: string globalStoragePath: string + merge?: boolean } -export async function saveTaskMessages({ messages, taskId, globalStoragePath }: SaveTaskMessagesOptions) { +export async function saveTaskMessages({ + messages, + taskId, + globalStoragePath, + merge = false, +}: SaveTaskMessagesOptions) { const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.uiMessages) - await safeWriteJson(filePath, messages) + await safeWriteJson(filePath, messages, merge ? { merge: mergeClineMessageSnapshots } : undefined) } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 8c88461ab0..fc318759b3 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -1022,7 +1022,7 @@ export class Task extends EventEmitter implements TaskLike { async overwriteApiConversationHistory(newHistory: ApiMessage[]) { this.apiConversationHistory = newHistory - await this.saveApiConversationHistory() + await this.saveApiConversationHistory(false) } /** @@ -1104,12 +1104,13 @@ export class Task extends EventEmitter implements TaskLike { return saved } - private async saveApiConversationHistory(): Promise { + private async saveApiConversationHistory(merge = true): Promise { try { await saveApiMessages({ messages: structuredClone(this.apiConversationHistory), taskId: this.taskId, globalStoragePath: this.globalStoragePath, + merge, }) return true } catch (error) { @@ -1183,7 +1184,7 @@ export class Task extends EventEmitter implements TaskLike { public async overwriteClineMessages(newMessages: ClineMessage[]) { this.hydrateClineMessages(newMessages) - await this.saveClineMessages() + await this.saveClineMessages(false) } private hydrateClineMessages(messages: ClineMessage[]) { @@ -1219,12 +1220,13 @@ export class Task extends EventEmitter implements TaskLike { } } - private async saveClineMessages(): Promise { + private async saveClineMessages(merge = true): Promise { try { await saveTaskMessages({ messages: structuredClone(this.clineMessages), taskId: this.taskId, globalStoragePath: this.globalStoragePath, + merge, }) if (this._taskApiConfigName === undefined) { diff --git a/src/core/task/__tests__/Task.persistence.spec.ts b/src/core/task/__tests__/Task.persistence.spec.ts index 46fe4c9706..42ecbc640c 100644 --- a/src/core/task/__tests__/Task.persistence.spec.ts +++ b/src/core/task/__tests__/Task.persistence.spec.ts @@ -309,6 +309,20 @@ describe("Task persistence", () => { const result = await task.retrySaveApiConversationHistory() expect(result).toBe(true) + expect(mockSaveApiMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: true })) + }) + + it("uses authoritative replacement for explicit API history overwrites", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "test task", + startTask: false, + }) + + await task.overwriteApiConversationHistory([{ role: "user", content: "replacement" }]) + + expect(mockSaveApiMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: false })) }) it("returns false on failure", async () => { @@ -402,6 +416,20 @@ describe("Task persistence", () => { const result = await (task as Record).saveClineMessages() expect(result).toBe(true) + expect(mockSaveTaskMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: true })) + }) + + it("uses authoritative replacement for explicit UI history overwrites", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "test task", + startTask: false, + }) + + await task.overwriteClineMessages([{ ts: 1, type: "say", say: "text", text: "replacement" }]) + + expect(mockSaveTaskMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: false })) }) it("returns false on failure", async () => { @@ -1144,6 +1172,37 @@ describe("Task persistence", () => { await expect(getTaskPersistenceAccess(task).resumeTaskFromHistory()).rejects.toThrow("stop after hydration") expect(mockSaveTaskMessages).not.toHaveBeenCalled() }) + + it("stops after API history hydration when the task is aborted", async () => { + const apiMessagesDeferred = + createDeferred }>>() + mockReadTaskMessages.mockResolvedValue([{ ts: 1, type: "say", say: "text", text: "Original task" }]) + mockReadApiMessages.mockReturnValue(apiMessagesDeferred.promise) + + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + historyItem: { + id: "issue-1279-api-abort", + number: 1, + ts: 1, + task: "Original task", + tokensIn: 10, + tokensOut: 5, + totalCost: 0.001, + }, + startTask: false, + }) + const askSpy = vi.spyOn(task, "ask") + const resumePromise = getTaskPersistenceAccess(task).resumeTaskFromHistory() + await vi.waitFor(() => expect(mockReadApiMessages).toHaveBeenCalled()) + + await task.abortTask(true) + apiMessagesDeferred.resolve([{ role: "user", content: [{ type: "text", text: "Original task" }] }]) + await resumePromise + + expect(askSpy).not.toHaveBeenCalled() + }) }) // ── flushPendingToolResultsToHistory — save failure/success ─────────── diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 6f1ccfeb12..e73d54989a 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -4101,7 +4101,12 @@ export class ClineProvider ) { parentClineMessages.push(subtaskUiMessage) } - await saveTaskMessages({ messages: parentClineMessages, taskId: parentTaskId, globalStoragePath }) + await saveTaskMessages({ + messages: parentClineMessages, + taskId: parentTaskId, + globalStoragePath, + merge: true, + }) // Find the tool_use_id from the last assistant message's new_task tool_use let toolUseId: string | undefined @@ -4186,7 +4191,12 @@ export class ClineProvider } } - await saveApiMessages({ messages: parentApiMessages as any, taskId: parentTaskId, globalStoragePath }) + await saveApiMessages({ + messages: parentApiMessages as any, + taskId: parentTaskId, + globalStoragePath, + merge: true, + }) // 4) Close child instance if still open (single-open-task invariant). // This MUST happen BEFORE marking the child "completed" because From ecaf14e9b7fe43039a092d81120aabdd9c1b8a84 Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Wed, 26 Aug 2026 02:43:31 +0000 Subject: [PATCH 3/8] fix(task-history): harden task resume persistence --- .../src/suite/restart-persistence.test.ts | 23 ++++ .../history-resume-delegation.spec.ts | 5 +- .../__tests__/apiMessages.spec.ts | 89 ++++++++---- .../__tests__/mergeMessageSnapshots.spec.ts | 49 +++++++ src/core/task-persistence/apiMessages.ts | 128 ++++++++++-------- src/core/task-persistence/index.ts | 8 +- .../task-persistence/mergeMessageSnapshots.ts | 25 ++-- .../readFileWithMissingRetry.ts | 25 ++++ src/core/task-persistence/taskMessages.ts | 25 +--- src/core/task/Task.ts | 11 ++ .../task/__tests__/Task.persistence.spec.ts | 40 ++++++ 11 files changed, 313 insertions(+), 115 deletions(-) create mode 100644 src/core/task-persistence/readFileWithMissingRetry.ts diff --git a/apps/vscode-e2e/src/suite/restart-persistence.test.ts b/apps/vscode-e2e/src/suite/restart-persistence.test.ts index 29e7fa3ddd..1f03e39e85 100644 --- a/apps/vscode-e2e/src/suite/restart-persistence.test.ts +++ b/apps/vscode-e2e/src/suite/restart-persistence.test.ts @@ -73,11 +73,18 @@ async function runCreate(api: RooCodeAPI): Promise { } async function runVerify(api: RooCodeAPI): Promise { + const taskMessages: Array<{ type: string; ask?: string }> = [] + const messageHandler = ({ taskId, message }: { taskId: string; message: (typeof taskMessages)[number] }) => { + if (taskId === verifiedTaskId) taskMessages.push(message) + } + let verifiedTaskId: string | undefined try { const createResult = await readPhaseResult(getResultsDir(), "create") assert.strictEqual(createResult.status, "passed") const taskId = createResult.values?.taskId assert.ok(taskId, "Create phase should record a task ID") + verifiedTaskId = taskId + api.on(RooCodeEventName.Message, messageHandler) await waitFor(() => api.isReady()) assert.strictEqual(await api.isTaskInHistory(taskId), true, "Task should be present after restart") @@ -87,6 +94,20 @@ async function runVerify(api: RooCodeAPI): Promise { const conversationLength = await api.getTaskApiConversationHistoryLength(taskId) assert.ok(conversationLength > 0, "API conversation history should be available after restart") + await api.resumeTask(taskId) + await waitFor(() => taskMessages.some(({ type, ask }) => type === "ask" && ask === "resume_completed_task")) + assert.strictEqual(await api.isTaskInHistory(taskId), true, "Reopened task should remain in history") + const reopenedHistoryItem = await api.getTaskHistoryItem(taskId) + assert.ok(reopenedHistoryItem, "Reopened task should retain its history item") + assert.ok( + reopenedHistoryItem.task.includes("RESTART_PERSISTENCE_SMOKE"), + "Reopened task should retain its persisted history title", + ) + assert.ok( + (await api.getTaskApiConversationHistoryLength(taskId)) >= conversationLength, + "Reopened task should retain its persisted API conversation history", + ) + await writePhaseResult(getResultsDir(), { version: PHASE_RESULT_VERSION, phase: "verify", @@ -102,6 +123,8 @@ async function runVerify(api: RooCodeAPI): Promise { error: serializePhaseError(error), }) throw error + } finally { + api.off(RooCodeEventName.Message, messageHandler) } } diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index 48410feb37..37b3a0ea21 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -399,11 +399,12 @@ describe("History resume delegation - parent metadata transitions", () => { totalCost: 0, } const log = vi.fn() + const taskHistoryStore = makeTaskHistoryStoreStub({ id: "child-read-failure", status: "active" }, parentItem) const provider = makeProviderStub({ contextProxy: { globalStorageUri: { fsPath: "/storage" } }, getTaskWithId: vi.fn().mockResolvedValue({ historyItem: parentItem }), getCurrentTask: vi.fn(() => ({ taskId: "child-read-failure" })), - taskHistoryStore: makeTaskHistoryStoreStub({ id: "child-read-failure", status: "active" }, parentItem), + taskHistoryStore, log, }) vi.mocked(readTaskMessages).mockRejectedValue(new Error("history unavailable")) @@ -416,8 +417,10 @@ describe("History resume delegation - parent metadata transitions", () => { expect(result).toBe(false) expect(log).toHaveBeenCalledWith(expect.stringContaining("history unavailable")) + expect(readApiMessages).not.toHaveBeenCalled() expect(saveTaskMessages).not.toHaveBeenCalled() expect(saveApiMessages).not.toHaveBeenCalled() + expect(taskHistoryStore.atomicUpdatePair).not.toHaveBeenCalled() }) it("reopenParentFromDelegation injects tool_result when new_task tool_use exists in API history", async () => { diff --git a/src/core/task-persistence/__tests__/apiMessages.spec.ts b/src/core/task-persistence/__tests__/apiMessages.spec.ts index b154cafaae..1be0cdb871 100644 --- a/src/core/task-persistence/__tests__/apiMessages.spec.ts +++ b/src/core/task-persistence/__tests__/apiMessages.spec.ts @@ -4,31 +4,36 @@ import * as os from "os" import * as path from "path" import * as fs from "fs/promises" +const hoisted = vi.hoisted(() => ({ readFileMock: vi.fn() })) +vi.mock("fs/promises", async (importOriginal) => ({ + ...(await importOriginal()), + readFile: hoisted.readFileMock, +})) + import { readApiMessages, saveApiMessages } from "../apiMessages" let tmpBaseDir: string beforeEach(async () => { + const actualFs = await vi.importActual("fs/promises") + hoisted.readFileMock.mockReset().mockImplementation(actualFs.readFile) tmpBaseDir = await fs.mkdtemp(path.join(os.tmpdir(), "roo-test-api-")) }) describe("apiMessages.readApiMessages", () => { - it("returns empty array when api_conversation_history.json contains invalid JSON", async () => { + it("rejects invalid api_conversation_history.json without treating it as empty history", async () => { const taskId = "task-corrupt-api" const taskDir = path.join(tmpBaseDir, "tasks", taskId) await fs.mkdir(taskDir, { recursive: true }) const filePath = path.join(taskDir, "api_conversation_history.json") await fs.writeFile(filePath, "<<>>", "utf8") - const result = await readApiMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) - - expect(result).toEqual([]) }) - it("returns empty array when claude_messages.json fallback contains invalid JSON", async () => { + it("rejects invalid claude_messages.json without deleting it", async () => { const taskId = "task-corrupt-fallback" const taskDir = path.join(tmpBaseDir, "tasks", taskId) await fs.mkdir(taskDir, { recursive: true }) @@ -37,13 +42,10 @@ describe("apiMessages.readApiMessages", () => { const oldPath = path.join(taskDir, "claude_messages.json") await fs.writeFile(oldPath, "not json at all {[!", "utf8") - const result = await readApiMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) - expect(result).toEqual([]) - // The corrupted fallback file should NOT be deleted const stillExists = await fs .access(oldPath) @@ -52,22 +54,19 @@ describe("apiMessages.readApiMessages", () => { expect(stillExists).toBe(true) }) - it("returns [] when file contains valid JSON that is not an array", async () => { + it("rejects valid non-array JSON in the current file", async () => { const taskId = "task-non-array-api" const taskDir = path.join(tmpBaseDir, "tasks", taskId) await fs.mkdir(taskDir, { recursive: true }) const filePath = path.join(taskDir, "api_conversation_history.json") await fs.writeFile(filePath, JSON.stringify("hello"), "utf8") - const result = await readApiMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) - - expect(result).toEqual([]) }) - it("returns [] when fallback file contains valid JSON that is not an array", async () => { + it("rejects valid non-array JSON in the fallback file", async () => { const taskId = "task-non-array-fallback" const taskDir = path.join(tmpBaseDir, "tasks", taskId) await fs.mkdir(taskDir, { recursive: true }) @@ -76,12 +75,32 @@ describe("apiMessages.readApiMessages", () => { const oldPath = path.join(taskDir, "claude_messages.json") await fs.writeFile(oldPath, JSON.stringify({ key: "value" }), "utf8") - const result = await readApiMessages({ - taskId, - globalStoragePath: tmpBaseDir, + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", }) + }) - expect(result).toEqual([]) + it("returns empty history only when current and legacy files are both missing", async () => { + await expect(readApiMessages({ taskId: "task-missing", globalStoragePath: tmpBaseDir })).resolves.toEqual([]) + }) + + it("retries one transient missing-file read", async () => { + vi.spyOn(Math, "random").mockReturnValue(0) + const missing = Object.assign(new Error("missing"), { code: "ENOENT" }) + hoisted.readFileMock.mockRejectedValueOnce(missing).mockResolvedValueOnce("[]") + + await expect(readApiMessages({ taskId: "task-retry", globalStoragePath: tmpBaseDir })).resolves.toEqual([]) + expect(hoisted.readFileMock).toHaveBeenCalledTimes(2) + }) + + it("does not retry non-ENOENT read failures", async () => { + const denied = Object.assign(new Error("denied"), { code: "EACCES" }) + hoisted.readFileMock.mockRejectedValueOnce(denied) + + await expect(readApiMessages({ taskId: "task-denied", globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "io_error", + }) + expect(hoisted.readFileMock).toHaveBeenCalledTimes(1) }) }) @@ -116,4 +135,28 @@ describe("apiMessages.saveApiMessages", () => { expect.objectContaining({ content: "disk suffix", ts: 3 }), ]) }) + + it("replaces the persisted snapshot when merge is false", async () => { + const taskId = "task-replace-api" + const taskDir = path.join(tmpBaseDir, "tasks", taskId) + await fs.mkdir(taskDir, { recursive: true }) + const filePath = path.join(taskDir, "api_conversation_history.json") + await fs.writeFile( + filePath, + JSON.stringify([ + { role: "user", content: "A", ts: 1 }, + { role: "assistant", content: "B", ts: 2 }, + ]), + "utf8", + ) + + await saveApiMessages({ + taskId, + globalStoragePath: tmpBaseDir, + merge: false, + messages: [{ role: "user", content: "C", ts: 3 }], + }) + + expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual([{ role: "user", content: "C", ts: 3 }]) + }) }) diff --git a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts index a9b1a31093..d3c1a8ec3e 100644 --- a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -29,6 +29,15 @@ describe("mergeClineMessageSnapshots", () => { expect(result).toEqual([expect.objectContaining({ ts: 1, partial: false, isAnswered: true })]) }) + it("keeps incoming completed and answered state when disk state is stale", () => { + const result = mergeClineMessageSnapshots( + [{ ts: 1, type: "ask", ask: "tool", partial: true, isAnswered: false }], + [{ ts: 1, type: "ask", ask: "tool", partial: false, isAnswered: true }], + ) + + expect(result).toEqual([expect.objectContaining({ ts: 1, partial: false, isAnswered: true })]) + }) + it("uses the incoming message when a timestamp is reused for a different message identity", () => { expect( mergeClineMessageSnapshots( @@ -85,6 +94,46 @@ describe("mergeApiMessageSnapshots", () => { ]) }) + it("keeps all incoming legacy messages when incoming has more than disk", () => { + const result = mergeApiMessageSnapshots( + [{ role: "user", content: "disk prefix" }], + [ + { role: "user", content: "updated prefix" }, + { role: "assistant", content: "incoming tail" }, + ], + ) + + expect(result).toEqual([ + { role: "user", content: "updated prefix" }, + { role: "assistant", content: "incoming tail" }, + ]) + }) + + it("linearly interleaves multiple disk-only messages while preserving equal-timestamp siblings", () => { + const result = mergeApiMessageSnapshots( + [ + { role: "assistant", content: "disk one", ts: 1 }, + { role: "assistant", content: "matched old two", ts: 2 }, + { role: "user", content: "equal sibling", ts: 2 }, + { role: "assistant", content: "disk three", ts: 3 }, + { role: "assistant", content: "disk five", ts: 5 }, + ], + [ + { role: "assistant", content: "incoming two", ts: 2 }, + { role: "assistant", content: "incoming four", ts: 4 }, + ], + ) + + expect(result).toEqual([ + expect.objectContaining({ content: "disk one", ts: 1 }), + expect.objectContaining({ content: "incoming two", ts: 2 }), + expect.objectContaining({ content: "equal sibling", ts: 2 }), + expect.objectContaining({ content: "disk three", ts: 3 }), + expect.objectContaining({ content: "incoming four", ts: 4 }), + expect.objectContaining({ content: "disk five", ts: 5 }), + ]) + }) + it("keeps legacy prefixes ahead of newer timestamped messages", () => { const result = mergeApiMessageSnapshots( [ diff --git a/src/core/task-persistence/apiMessages.ts b/src/core/task-persistence/apiMessages.ts index 3fdcd376a5..d3ef89f9ce 100644 --- a/src/core/task-persistence/apiMessages.ts +++ b/src/core/task-persistence/apiMessages.ts @@ -4,11 +4,10 @@ import * as fs from "fs/promises" import { Anthropic } from "@anthropic-ai/sdk" -import { fileExistsAtPath } from "../../utils/fs" - import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" import { mergeApiMessageSnapshots } from "./mergeMessageSnapshots" +import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" export type ApiMessage = Anthropic.MessageParam & { ts?: number @@ -38,6 +37,59 @@ export type ApiMessage = Anthropic.MessageParam & { isTruncationMarker?: boolean } +export type ApiMessagesReadErrorKind = "invalid" | "io_error" + +export class ApiMessagesReadError extends Error { + constructor( + public readonly kind: ApiMessagesReadErrorKind, + message: string, + public readonly originalError?: unknown, + ) { + super(message) + this.name = "ApiMessagesReadError" + } +} + +function parseApiMessages(fileContent: string, taskId: string, filePath: string): ApiMessage[] { + let parsedData: unknown + try { + parsedData = JSON.parse(fileContent) + } catch (error) { + throw new ApiMessagesReadError( + "invalid", + `Failed to parse API conversation history for ${taskId} at ${filePath}`, + error, + ) + } + + if (!Array.isArray(parsedData)) { + throw new ApiMessagesReadError( + "invalid", + `API conversation history for ${taskId} at ${filePath} must be an array, got ${typeof parsedData}`, + ) + } + + return parsedData +} + +async function readApiMessagesFile(taskId: string, filePath: string): Promise { + let fileContent: string + try { + fileContent = await readFileWithMissingRetry(filePath) + } catch (error) { + if (getErrorCode(error) === "ENOENT") { + return undefined + } + throw new ApiMessagesReadError( + "io_error", + `Failed to read API conversation history for ${taskId} at ${filePath}`, + error, + ) + } + + return parseApiMessages(fileContent, taskId, filePath) +} + export async function readApiMessages({ taskId, globalStoragePath, @@ -48,63 +100,27 @@ export async function readApiMessages({ const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.apiConversationHistory) - if (await fileExistsAtPath(filePath)) { - const fileContent = await fs.readFile(filePath, "utf8") - try { - const parsedData = JSON.parse(fileContent) - if (!Array.isArray(parsedData)) { - console.warn( - `[readApiMessages] Parsed data is not an array (got ${typeof parsedData}), returning empty. TaskId: ${taskId}, Path: ${filePath}`, - ) - return [] - } - if (parsedData.length === 0) { - console.error( - `[Roo-Debug] readApiMessages: Found API conversation history file, but it's empty (parsed as []). TaskId: ${taskId}, Path: ${filePath}`, - ) - } - return parsedData - } catch (error) { - console.warn( - `[readApiMessages] Error parsing API conversation history file, returning empty. TaskId: ${taskId}, Path: ${filePath}, Error: ${error}`, - ) - return [] - } - } else { - const oldPath = path.join(taskDir, "claude_messages.json") + const currentMessages = await readApiMessagesFile(taskId, filePath) + if (currentMessages !== undefined) { + return currentMessages + } - if (await fileExistsAtPath(oldPath)) { - const fileContent = await fs.readFile(oldPath, "utf8") - try { - const parsedData = JSON.parse(fileContent) - if (!Array.isArray(parsedData)) { - console.warn( - `[readApiMessages] Parsed OLD data is not an array (got ${typeof parsedData}), returning empty. TaskId: ${taskId}, Path: ${oldPath}`, - ) - return [] - } - if (parsedData.length === 0) { - console.error( - `[Roo-Debug] readApiMessages: Found OLD API conversation history file (claude_messages.json), but it's empty (parsed as []). TaskId: ${taskId}, Path: ${oldPath}`, - ) - } - await fs.unlink(oldPath) - return parsedData - } catch (error) { - console.warn( - `[readApiMessages] Error parsing OLD API conversation history file (claude_messages.json), returning empty. TaskId: ${taskId}, Path: ${oldPath}, Error: ${error}`, - ) - // DO NOT unlink oldPath if parsing failed. - return [] - } - } + const oldPath = path.join(taskDir, "claude_messages.json") + const legacyMessages = await readApiMessagesFile(taskId, oldPath) + if (legacyMessages === undefined) { + return [] } - // If we reach here, neither the new nor the old history file was found. - console.error( - `[Roo-Debug] readApiMessages: API conversation history file not found for taskId: ${taskId}. Expected at: ${filePath}`, - ) - return [] + try { + await fs.unlink(oldPath) + } catch (error) { + throw new ApiMessagesReadError( + "io_error", + `Failed to remove migrated API conversation history for ${taskId} at ${oldPath}`, + error, + ) + } + return legacyMessages } export async function saveApiMessages({ diff --git a/src/core/task-persistence/index.ts b/src/core/task-persistence/index.ts index 5dda196592..5de1787533 100644 --- a/src/core/task-persistence/index.ts +++ b/src/core/task-persistence/index.ts @@ -1,4 +1,10 @@ -export { type ApiMessage, readApiMessages, saveApiMessages } from "./apiMessages" +export { + type ApiMessage, + ApiMessagesReadError, + type ApiMessagesReadErrorKind, + readApiMessages, + saveApiMessages, +} from "./apiMessages" export { readTaskMessages, saveTaskMessages, diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts index 1b158fecff..64713dda95 100644 --- a/src/core/task-persistence/mergeMessageSnapshots.ts +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -44,19 +44,24 @@ function mergeTimestampedSnapshots( messages.slice(consumedByTimestamp.get(timestamp) ?? 0), ) - for (const diskMessage of diskOnlyTimestamped) { - const insertionIndex = merged.findIndex( - (message) => isRecord(message) && typeof message.ts === "number" && message.ts > (diskMessage.ts as number), - ) - if (insertionIndex === -1) { - merged.push(diskMessage) - } else { - merged.splice(insertionIndex, 0, diskMessage) + const timestampMerged: unknown[] = [] + let diskOnlyIndex = 0 + for (const incomingMessage of merged) { + if (isRecord(incomingMessage) && typeof incomingMessage.ts === "number") { + while ( + diskOnlyIndex < diskOnlyTimestamped.length && + (diskOnlyTimestamped[diskOnlyIndex].ts as number) < incomingMessage.ts + ) { + timestampMerged.push(diskOnlyTimestamped[diskOnlyIndex]) + diskOnlyIndex++ + } } + timestampMerged.push(incomingMessage) } + timestampMerged.push(...diskOnlyTimestamped.slice(diskOnlyIndex)) - merged.push(...existingLegacy.slice(incomingLegacyCount)) - return merged + timestampMerged.push(...existingLegacy.slice(incomingLegacyCount)) + return timestampMerged } export function mergeClineMessageSnapshots(existing: unknown, incoming: unknown): unknown { diff --git a/src/core/task-persistence/readFileWithMissingRetry.ts b/src/core/task-persistence/readFileWithMissingRetry.ts new file mode 100644 index 0000000000..d3369fb44c --- /dev/null +++ b/src/core/task-persistence/readFileWithMissingRetry.ts @@ -0,0 +1,25 @@ +import * as fs from "fs/promises" + +const READ_RETRY_MIN_MS = 1 +const READ_RETRY_MAX_MS = 10 +const READ_RETRY_RANGE_MS = READ_RETRY_MAX_MS - READ_RETRY_MIN_MS + 1 + +export function getErrorCode(error: unknown): string | undefined { + return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" + ? error.code + : undefined +} + +export async function readFileWithMissingRetry(filePath: string): Promise { + try { + return await fs.readFile(filePath, "utf8") + } catch (error) { + if (getErrorCode(error) !== "ENOENT") { + throw error + } + + const retryDelay = READ_RETRY_MIN_MS + Math.floor(Math.random() * READ_RETRY_RANGE_MS) + await new Promise((resolve) => setTimeout(resolve, retryDelay)) + return fs.readFile(filePath, "utf8") + } +} diff --git a/src/core/task-persistence/taskMessages.ts b/src/core/task-persistence/taskMessages.ts index 13b0ae728f..cf76877d4b 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -1,12 +1,12 @@ import { safeWriteJson } from "../../utils/safeWriteJson" import * as path from "path" -import * as fs from "fs/promises" import type { ClineMessage } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" import { mergeClineMessageSnapshots } from "./mergeMessageSnapshots" +import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" export type TaskMessagesReadErrorKind = "not_found" | "invalid" | "io_error" @@ -26,29 +26,6 @@ export type ReadTaskMessagesOptions = { globalStoragePath: string } -const READ_RETRY_MIN_MS = 10 -const READ_RETRY_RANGE_MS = 291 - -function getErrorCode(error: unknown): string | undefined { - return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" - ? error.code - : undefined -} - -async function readFileWithMissingRetry(filePath: string): Promise { - try { - return await fs.readFile(filePath, "utf8") - } catch (error) { - if (getErrorCode(error) !== "ENOENT") { - throw error - } - - const retryDelay = READ_RETRY_MIN_MS + Math.floor(Math.random() * READ_RETRY_RANGE_MS) - await new Promise((resolve) => setTimeout(resolve, retryDelay)) - return fs.readFile(filePath, "utf8") - } -} - export async function readTaskMessages({ taskId, globalStoragePath, diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index fc318759b3..36ad9143f7 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -2144,6 +2144,17 @@ export class Task extends EventEmitter implements TaskLike { } } + // Incomplete reasoning has no matching API-history entry and would become + // an orphaned bubble when the resumed request starts fresh reasoning. + while (modifiedClineMessages.length > 0) { + const lastMessage = modifiedClineMessages[modifiedClineMessages.length - 1] + if (lastMessage.type === "say" && lastMessage.say === "reasoning" && lastMessage.partial === true) { + modifiedClineMessages.pop() + } else { + break + } + } + // Since we don't use `api_req_finished` anymore, we need to check if the // last `api_req_started` has a cost value, if it doesn't and no // cancellation reason to present, then we remove it since it indicates diff --git a/src/core/task/__tests__/Task.persistence.spec.ts b/src/core/task/__tests__/Task.persistence.spec.ts index 42ecbc640c..47c13a7bfd 100644 --- a/src/core/task/__tests__/Task.persistence.spec.ts +++ b/src/core/task/__tests__/Task.persistence.spec.ts @@ -1173,6 +1173,40 @@ describe("Task persistence", () => { expect(mockSaveTaskMessages).not.toHaveBeenCalled() }) + it("removes incomplete trailing reasoning before prompting to resume", async () => { + mockReadTaskMessages.mockResolvedValue([ + { ts: 1, type: "say", say: "text", text: "Original task" }, + { ts: 2, type: "say", say: "reasoning", text: "Incomplete conclusion", partial: true }, + ]) + mockReadApiMessages.mockResolvedValue([ + { role: "user", content: [{ type: "text", text: "Original task" }] }, + ]) + + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + historyItem: { + id: "issue-1279-partial-reasoning", + number: 1, + ts: 2, + task: "Original task", + tokensIn: 10, + tokensOut: 5, + totalCost: 0.001, + }, + startTask: false, + }) + vi.spyOn(task, "ask").mockImplementation(async () => { + expect(task.clineMessages).not.toContainEqual( + expect.objectContaining({ say: "reasoning", partial: true }), + ) + throw new Error("stop after hydration") + }) + + await expect(getTaskPersistenceAccess(task).resumeTaskFromHistory()).rejects.toThrow("stop after hydration") + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + }) + it("stops after API history hydration when the task is aborted", async () => { const apiMessagesDeferred = createDeferred }>>() @@ -1198,10 +1232,16 @@ describe("Task persistence", () => { await vi.waitFor(() => expect(mockReadApiMessages).toHaveBeenCalled()) await task.abortTask(true) + // abortTask persists its own cancellation state. Reset those calls so the + // assertions below isolate work performed by the resumed hydration path. + mockSaveTaskMessages.mockClear() + vi.mocked(mockProvider.updateTaskHistory).mockClear() apiMessagesDeferred.resolve([{ role: "user", content: [{ type: "text", text: "Original task" }] }]) await resumePromise expect(askSpy).not.toHaveBeenCalled() + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + expect(mockProvider.updateTaskHistory).not.toHaveBeenCalled() }) }) From 7cf79e29ecfe2c1fbea1f388f40368d41ddcd1ec Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Wed, 26 Aug 2026 22:31:36 +0000 Subject: [PATCH 4/8] fix(task-history): persist migrated API history --- .../task-persistence/__tests__/apiMessages.spec.ts | 14 ++++++++++++++ src/core/task-persistence/apiMessages.ts | 5 +++++ src/core/task-persistence/mergeMessageSnapshots.ts | 9 ++++++--- 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/src/core/task-persistence/__tests__/apiMessages.spec.ts b/src/core/task-persistence/__tests__/apiMessages.spec.ts index 1be0cdb871..5723190bfb 100644 --- a/src/core/task-persistence/__tests__/apiMessages.spec.ts +++ b/src/core/task-persistence/__tests__/apiMessages.spec.ts @@ -84,6 +84,20 @@ describe("apiMessages.readApiMessages", () => { await expect(readApiMessages({ taskId: "task-missing", globalStoragePath: tmpBaseDir })).resolves.toEqual([]) }) + it("migrates valid legacy history before deleting its source", async () => { + const taskId = "task-legacy-api" + const taskDir = path.join(tmpBaseDir, "tasks", taskId) + await fs.mkdir(taskDir, { recursive: true }) + const oldPath = path.join(taskDir, "claude_messages.json") + const currentPath = path.join(taskDir, "api_conversation_history.json") + const legacyMessages = [{ role: "user", content: "legacy", ts: 1 }] + await fs.writeFile(oldPath, JSON.stringify(legacyMessages), "utf8") + + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).resolves.toEqual(legacyMessages) + await expect(fs.readFile(currentPath, "utf8").then(JSON.parse)).resolves.toEqual(legacyMessages) + await expect(fs.access(oldPath)).rejects.toMatchObject({ code: "ENOENT" }) + }) + it("retries one transient missing-file read", async () => { vi.spyOn(Math, "random").mockReturnValue(0) const missing = Object.assign(new Error("missing"), { code: "ENOENT" }) diff --git a/src/core/task-persistence/apiMessages.ts b/src/core/task-persistence/apiMessages.ts index d3ef89f9ce..b7da8056c3 100644 --- a/src/core/task-persistence/apiMessages.ts +++ b/src/core/task-persistence/apiMessages.ts @@ -111,6 +111,11 @@ export async function readApiMessages({ return [] } + // Persist the successfully parsed legacy history before deleting its source. + // The next ordinary task-history save may not happen until after user input, + // so returning the in-memory data alone would leave a data-loss window. + await safeWriteJson(filePath, legacyMessages, { merge: mergeApiMessageSnapshots }) + try { await fs.unlink(oldPath) } catch (error) { diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts index 64713dda95..69882bb86a 100644 --- a/src/core/task-persistence/mergeMessageSnapshots.ts +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -7,7 +7,7 @@ function isRecord(value: unknown): value is MessageRecord { function mergeTimestampedSnapshots( existing: unknown, incoming: unknown, - mergeMatch: (disk: MessageRecord, next: MessageRecord) => MessageRecord, + mergeMatch?: (disk: MessageRecord, next: MessageRecord) => MessageRecord, ): unknown { if (!Array.isArray(existing) || !Array.isArray(incoming)) { return incoming @@ -37,7 +37,10 @@ function mergeTimestampedSnapshots( const consumed = consumedByTimestamp.get(message.ts) ?? 0 consumedByTimestamp.set(message.ts, consumed + 1) const diskMessage = existingGroups.get(message.ts)?.[consumed] - return diskMessage ? mergeMatch(diskMessage, message) : message + if (mergeMatch !== undefined && diskMessage !== undefined) { + return mergeMatch(diskMessage, message) + } + return message }) const diskOnlyTimestamped = [...existingGroups.entries()].flatMap(([timestamp, messages]) => @@ -82,5 +85,5 @@ export function mergeClineMessageSnapshots(existing: unknown, incoming: unknown) } export function mergeApiMessageSnapshots(existing: unknown, incoming: unknown): unknown { - return mergeTimestampedSnapshots(existing, incoming, (_disk, next) => next) + return mergeTimestampedSnapshots(existing, incoming) } From 7e97cf0d4b0e136ec4fbca7d75ac0351051eaab3 Mon Sep 17 00:00:00 2001 From: Roomote Date: Sat, 5 Sep 2026 00:40:22 +0000 Subject: [PATCH 5/8] fix(history): preserve concurrent equal-timestamp messages --- packages/types/src/message.ts | 1 + .../history-resume-delegation.spec.ts | 72 +++++++++++++++++-- .../__tests__/apiMessages.spec.ts | 11 ++- .../__tests__/mergeMessageSnapshots.spec.ts | 45 +++++++++++- .../__tests__/taskMessages.spec.ts | 10 +-- src/core/task-persistence/apiMessages.ts | 23 ++++-- src/core/task-persistence/index.ts | 1 + .../task-persistence/mergeMessageSnapshots.ts | 50 ++++++++++--- src/core/task-persistence/taskMessages.ts | 22 ++++-- src/core/task/Task.ts | 31 +++++--- .../task/__tests__/Task.persistence.spec.ts | 44 ++++++++++-- src/core/task/apiConversationHistory.ts | 4 +- src/core/webview/ClineProvider.ts | 8 +-- 13 files changed, 274 insertions(+), 48 deletions(-) diff --git a/packages/types/src/message.ts b/packages/types/src/message.ts index 28d5af82ac..01d7962266 100644 --- a/packages/types/src/message.ts +++ b/packages/types/src/message.ts @@ -248,6 +248,7 @@ export type ContextTruncation = z.infer * Note: These fields are mutually exclusive - a message will have at most one of them. */ export const clineMessageSchema = z.object({ + messageId: z.string().optional(), ts: z.number(), type: z.union([z.literal("ask"), z.literal("say")]), ask: clineAskSchema.optional(), diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index 37b3a0ea21..1bc42b3382 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -2,7 +2,9 @@ import { describe, it, expect, vi, beforeEach } from "vitest" import { RooCodeEventName } from "@roo-code/types" -import type { HistoryItem } from "@roo-code/types" +import type { ClineMessage, HistoryItem } from "@roo-code/types" + +import type { ApiMessage } from "../core/task-persistence" /* vscode mock for Task/Provider imports */ vi.mock("vscode", () => { @@ -44,8 +46,8 @@ vi.mock("../core/task-persistence", async (importOriginal) => { return { ...real, readApiMessages: vi.fn().mockResolvedValue([]), - saveApiMessages: vi.fn().mockResolvedValue(undefined), - saveTaskMessages: vi.fn().mockResolvedValue(undefined), + saveApiMessages: vi.fn(async ({ messages }: { messages: unknown[] }) => messages), + saveTaskMessages: vi.fn(async ({ messages }: { messages: unknown[] }) => messages), } }) @@ -237,7 +239,7 @@ describe("History resume delegation - parent metadata transitions", () => { removeClineFromStack, createTaskWithHistoryItem, taskHistoryStore, - } as any) + } as unknown as ClineProvider) vi.mocked(readTaskMessages).mockResolvedValue([]) vi.mocked(readApiMessages).mockResolvedValue([]) @@ -386,6 +388,68 @@ describe("History resume delegation - parent metadata transitions", () => { expect(apiCall.messages).toHaveLength(2) // 1 original + 1 injected }) + it("hydrates the reopened parent from locked merge results without authoritative rewrites", async () => { + const parentItem = { + id: "parent-merge", + status: "delegated", + awaitingChildId: "child-merge", + childIds: ["child-merge"], + ts: 100, + task: "Parent", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + const overwriteClineMessages = vi.fn() + const overwriteApiConversationHistory = vi.fn() + const taskHistoryStore = makeTaskHistoryStoreStub({ id: "child-merge", status: "active" }, parentItem) + const provider = makeProviderStub({ + contextProxy: { globalStorageUri: { fsPath: "/storage" } }, + getTaskWithId: vi.fn().mockResolvedValue({ historyItem: parentItem }), + emit: vi.fn(), + getCurrentTask: vi.fn(() => ({ taskId: "child-merge" })), + removeClineFromStack: vi.fn().mockResolvedValue(undefined), + createTaskWithHistoryItem: vi.fn().mockResolvedValue({ + overwriteClineMessages, + overwriteApiConversationHistory, + resumeAfterDelegation: vi.fn().mockResolvedValue(undefined), + }), + taskHistoryStore, + } as any) + + vi.mocked(readTaskMessages).mockResolvedValue([{ ts: 1, type: "say", say: "text", text: "initial UI" }]) + vi.mocked(readApiMessages).mockResolvedValue([{ ts: 1, role: "user", content: "initial API" }]) + vi.mocked(saveTaskMessages).mockResolvedValueOnce([ + { ts: 1, type: "say", say: "text", text: "initial UI" }, + { ts: 2, type: "say", say: "text", text: "concurrent UI" }, + ] satisfies ClineMessage[]) + vi.mocked(saveApiMessages).mockResolvedValueOnce([ + { ts: 1, role: "user", content: "initial API" }, + { ts: 2, role: "assistant", content: "concurrent API" }, + ] satisfies ApiMessage[]) + + await ClineProvider.prototype.reopenParentFromDelegation.call(provider, { + parentTaskId: "parent-merge", + childTaskId: "child-merge", + completionResultSummary: "Done", + }) + + expect(overwriteClineMessages).toHaveBeenCalledWith( + [ + { ts: 1, type: "say", say: "text", text: "initial UI" }, + { ts: 2, type: "say", say: "text", text: "concurrent UI" }, + ], + false, + ) + expect(overwriteApiConversationHistory).toHaveBeenCalledWith( + [ + { ts: 1, role: "user", content: "initial API" }, + { ts: 2, role: "assistant", content: "concurrent API" }, + ], + false, + ) + }) + it("does not reopen or overwrite a parent when its UI history cannot be read", async () => { const parentItem = { id: "parent-read-failure", diff --git a/src/core/task-persistence/__tests__/apiMessages.spec.ts b/src/core/task-persistence/__tests__/apiMessages.spec.ts index 5723190bfb..c74ef6b150 100644 --- a/src/core/task-persistence/__tests__/apiMessages.spec.ts +++ b/src/core/task-persistence/__tests__/apiMessages.spec.ts @@ -133,7 +133,7 @@ describe("apiMessages.saveApiMessages", () => { "utf8", ) - await saveApiMessages({ + const savedMessages = await saveApiMessages({ taskId, globalStoragePath: tmpBaseDir, merge: true, @@ -148,6 +148,11 @@ describe("apiMessages.saveApiMessages", () => { expect.objectContaining({ content: "incoming", ts: 2 }), expect.objectContaining({ content: "disk suffix", ts: 3 }), ]) + expect(savedMessages).toEqual([ + expect.objectContaining({ content: "updated prefix", ts: 1 }), + expect.objectContaining({ content: "incoming", ts: 2 }), + expect.objectContaining({ content: "disk suffix", ts: 3 }), + ]) }) it("replaces the persisted snapshot when merge is false", async () => { @@ -171,6 +176,8 @@ describe("apiMessages.saveApiMessages", () => { messages: [{ role: "user", content: "C", ts: 3 }], }) - expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual([{ role: "user", content: "C", ts: 3 }]) + expect(JSON.parse(await fs.readFile(filePath, "utf8"))).toEqual([ + expect.objectContaining({ role: "user", content: "C", ts: 3, messageId: expect.any(String) }), + ]) }) }) diff --git a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts index d3c1a8ec3e..e371b2cc4c 100644 --- a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -1,4 +1,23 @@ -import { mergeApiMessageSnapshots, mergeClineMessageSnapshots } from "../mergeMessageSnapshots" +import { + ensureMessageIdentifiers, + mergeApiMessageSnapshots, + mergeClineMessageSnapshots, +} from "../mergeMessageSnapshots" + +describe("ensureMessageIdentifiers", () => { + it("assigns the same upgrade identifiers to equivalent legacy snapshots", () => { + const legacyMessages: Array<{ messageId?: string; ts?: number; role: string }> = [ + { ts: 1, role: "user" }, + { ts: 1, role: "assistant" }, + { role: "user" }, + ] + const first = ensureMessageIdentifiers(structuredClone(legacyMessages)) + const second = ensureMessageIdentifiers(structuredClone(legacyMessages)) + + expect(first.map(({ messageId }) => messageId)).toEqual(second.map(({ messageId }) => messageId)) + expect(new Set(first.map(({ messageId }) => messageId)).size).toBe(3) + }) +}) describe("mergeClineMessageSnapshots", () => { it("preserves disk-only messages and applies incoming updates in timestamp order", () => { @@ -47,6 +66,18 @@ describe("mergeClineMessageSnapshots", () => { ).toEqual([{ ts: 1, type: "ask", ask: "followup", text: "new" }]) }) + it("preserves unrelated messages with the same timestamp", () => { + expect( + mergeClineMessageSnapshots( + [{ messageId: "disk", ts: 1, type: "say", say: "text", text: "disk message" }], + [{ messageId: "incoming", ts: 1, type: "say", say: "text", text: "incoming message" }], + ), + ).toEqual([ + { messageId: "incoming", ts: 1, type: "say", say: "text", text: "incoming message" }, + { messageId: "disk", ts: 1, type: "say", say: "text", text: "disk message" }, + ]) + }) + it("returns incoming data when either snapshot is not an array", () => { expect(mergeClineMessageSnapshots(null, [{ ts: 1 }])).toEqual([{ ts: 1 }]) expect(mergeClineMessageSnapshots([], "invalid")).toBe("invalid") @@ -54,6 +85,18 @@ describe("mergeClineMessageSnapshots", () => { }) describe("mergeApiMessageSnapshots", () => { + it("preserves unrelated messages with the same timestamp", () => { + expect( + mergeApiMessageSnapshots( + [{ messageId: "disk", role: "user", content: "disk message", ts: 1 }], + [{ messageId: "incoming", role: "user", content: "incoming message", ts: 1 }], + ), + ).toEqual([ + { messageId: "incoming", role: "user", content: "incoming message", ts: 1 }, + { messageId: "disk", role: "user", content: "disk message", ts: 1 }, + ]) + }) + it("retains equal-timestamp records and keeps tool calls before their results", () => { const result = mergeApiMessageSnapshots( [ diff --git a/src/core/task-persistence/__tests__/taskMessages.spec.ts b/src/core/task-persistence/__tests__/taskMessages.spec.ts index 61cecdb353..e494b5b594 100644 --- a/src/core/task-persistence/__tests__/taskMessages.spec.ts +++ b/src/core/task-persistence/__tests__/taskMessages.spec.ts @@ -83,16 +83,18 @@ describe("taskMessages.saveTaskMessages", () => { it("passes the history merge callback only when requested", async () => { const messages: ClineMessage[] = [{ ts: 2, type: "say", say: "text", text: "incoming" }] - await saveTaskMessages({ + hoisted.safeWriteJsonMock.mockImplementationOnce(async (_path, data, options) => + options.merge([{ ts: 1, type: "say", say: "text", text: "disk" }], data), + ) + const savedMessages = await saveTaskMessages({ messages, taskId: "task-merge", globalStoragePath: tmpBaseDir, merge: true, }) - const merge = hoisted.safeWriteJsonMock.mock.calls[0][2]?.merge - expect(merge).toBeTypeOf("function") - expect(merge([{ ts: 1, type: "say", say: "text", text: "disk" }], messages)).toEqual([ + expect(hoisted.safeWriteJsonMock.mock.calls[0][2]?.merge).toBeTypeOf("function") + expect(savedMessages).toEqual([ expect.objectContaining({ ts: 1, text: "disk" }), expect.objectContaining({ ts: 2, text: "incoming" }), ]) diff --git a/src/core/task-persistence/apiMessages.ts b/src/core/task-persistence/apiMessages.ts index b7da8056c3..079973240b 100644 --- a/src/core/task-persistence/apiMessages.ts +++ b/src/core/task-persistence/apiMessages.ts @@ -1,15 +1,16 @@ -import { safeWriteJson } from "../../utils/safeWriteJson" import * as path from "path" import * as fs from "fs/promises" import { Anthropic } from "@anthropic-ai/sdk" +import { safeWriteJson } from "../../utils/safeWriteJson" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" -import { mergeApiMessageSnapshots } from "./mergeMessageSnapshots" +import { ensureMessageIdentifiers, mergeApiMessageSnapshots } from "./mergeMessageSnapshots" import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" export type ApiMessage = Anthropic.MessageParam & { + messageId?: string ts?: number isSummary?: boolean id?: string @@ -138,8 +139,22 @@ export async function saveApiMessages({ taskId: string globalStoragePath: string merge?: boolean -}) { +}): Promise { + ensureMessageIdentifiers(messages) const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.apiConversationHistory) - await safeWriteJson(filePath, messages, merge ? { merge: mergeApiMessageSnapshots } : undefined) + let savedMessages = messages + await safeWriteJson( + filePath, + messages, + merge + ? { + merge: (existing, incoming) => { + savedMessages = mergeApiMessageSnapshots(existing, incoming) as ApiMessage[] + return savedMessages + }, + } + : undefined, + ) + return savedMessages } diff --git a/src/core/task-persistence/index.ts b/src/core/task-persistence/index.ts index 5de1787533..14adeedc68 100644 --- a/src/core/task-persistence/index.ts +++ b/src/core/task-persistence/index.ts @@ -12,6 +12,7 @@ export { type TaskMessagesReadErrorKind, } from "./taskMessages" export { taskMetadata } from "./taskMetadata" +export { ensureMessageIdentifiers } from "./mergeMessageSnapshots" export { TaskHistoryStore } from "./TaskHistoryStore" export { abandonDelegatedChild, diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts index 69882bb86a..4da55062f0 100644 --- a/src/core/task-persistence/mergeMessageSnapshots.ts +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -1,9 +1,23 @@ -type MessageRecord = Record & { ts?: unknown } +type MessageRecord = Record & { messageId?: unknown; ts?: unknown } +type IdentifiedMessage = { messageId?: string; ts?: unknown } function isRecord(value: unknown): value is MessageRecord { return typeof value === "object" && value !== null } +export function ensureMessageIdentifiers(messages: T[]): T[] { + const timestampOrdinals = new Map() + for (const message of messages) { + if (typeof message.messageId === "string") continue + + const timestampKey = typeof message.ts === "number" ? String(message.ts) : "none" + const ordinal = timestampOrdinals.get(timestampKey) ?? 0 + timestampOrdinals.set(timestampKey, ordinal + 1) + message.messageId = `legacy:${timestampKey}:${ordinal}` + } + return messages +} + function mergeTimestampedSnapshots( existing: unknown, incoming: unknown, @@ -26,7 +40,7 @@ function mergeTimestampedSnapshots( } } - const consumedByTimestamp = new Map() + const consumedByTimestamp = new Map>() let incomingLegacyCount = 0 const merged = incoming.map((message) => { if (!isRecord(message) || typeof message.ts !== "number") { @@ -34,18 +48,38 @@ function mergeTimestampedSnapshots( return message } - const consumed = consumedByTimestamp.get(message.ts) ?? 0 - consumedByTimestamp.set(message.ts, consumed + 1) - const diskMessage = existingGroups.get(message.ts)?.[consumed] + const existingGroup = existingGroups.get(message.ts) ?? [] + const consumed = consumedByTimestamp.get(message.ts) ?? new Set() + const messageId = typeof message.messageId === "string" ? message.messageId : undefined + let diskIndex = + messageId === undefined + ? existingGroup.findIndex( + (candidate, index) => !consumed.has(index) && candidate.messageId === undefined, + ) + : existingGroup.findIndex( + (candidate, index) => !consumed.has(index) && candidate.messageId === messageId, + ) + if (diskIndex === -1 && messageId !== undefined) { + // Match one legacy record while persisted histories are upgraded with identifiers. + diskIndex = existingGroup.findIndex( + (candidate, index) => !consumed.has(index) && candidate.messageId === undefined, + ) + } + const diskMessage = diskIndex === -1 ? undefined : existingGroup[diskIndex] + if (diskIndex !== -1) { + consumed.add(diskIndex) + consumedByTimestamp.set(message.ts, consumed) + } if (mergeMatch !== undefined && diskMessage !== undefined) { return mergeMatch(diskMessage, message) } return message }) - const diskOnlyTimestamped = [...existingGroups.entries()].flatMap(([timestamp, messages]) => - messages.slice(consumedByTimestamp.get(timestamp) ?? 0), - ) + const diskOnlyTimestamped = [...existingGroups.entries()].flatMap(([timestamp, messages]) => { + const consumed = consumedByTimestamp.get(timestamp) + return messages.filter((_message, index) => !consumed?.has(index)) + }) const timestampMerged: unknown[] = [] let diskOnlyIndex = 0 diff --git a/src/core/task-persistence/taskMessages.ts b/src/core/task-persistence/taskMessages.ts index cf76877d4b..89ec372c61 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -1,11 +1,11 @@ -import { safeWriteJson } from "../../utils/safeWriteJson" import * as path from "path" import type { ClineMessage } from "@roo-code/types" +import { safeWriteJson } from "../../utils/safeWriteJson" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" -import { mergeClineMessageSnapshots } from "./mergeMessageSnapshots" +import { ensureMessageIdentifiers, mergeClineMessageSnapshots } from "./mergeMessageSnapshots" import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" export type TaskMessagesReadErrorKind = "not_found" | "invalid" | "io_error" @@ -70,8 +70,22 @@ export async function saveTaskMessages({ taskId, globalStoragePath, merge = false, -}: SaveTaskMessagesOptions) { +}: SaveTaskMessagesOptions): Promise { + ensureMessageIdentifiers(messages) const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.uiMessages) - await safeWriteJson(filePath, messages, merge ? { merge: mergeClineMessageSnapshots } : undefined) + let savedMessages = messages + await safeWriteJson( + filePath, + messages, + merge + ? { + merge: (existing, incoming) => { + savedMessages = mergeClineMessageSnapshots(existing, incoming) as ClineMessage[] + return savedMessages + }, + } + : undefined, + ) + return savedMessages } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 36ad9143f7..63578c74b5 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -112,6 +112,7 @@ import { ClineProvider } from "../webview/ClineProvider" import { MultiSearchReplaceDiffStrategy } from "../diff/strategies/multi-search-replace" import { type ApiMessage, + ensureMessageIdentifiers, readApiMessages, saveApiMessages, readTaskMessages, @@ -979,7 +980,8 @@ export class Task extends EventEmitter implements TaskLike { // API Messages private async getSavedApiConversationHistory(): Promise { - return readApiMessages({ taskId: this.taskId, globalStoragePath: this.globalStoragePath }) + const messages = await readApiMessages({ taskId: this.taskId, globalStoragePath: this.globalStoragePath }) + return ensureMessageIdentifiers(messages) } private async addToApiConversationHistory(message: Anthropic.MessageParam, reasoning?: string) { @@ -1020,9 +1022,11 @@ export class Task extends EventEmitter implements TaskLike { // For API requests, consecutive same-role messages are merged via mergeConsecutiveApiMessages() // so rewind/edit behavior can still reference original message boundaries. - async overwriteApiConversationHistory(newHistory: ApiMessage[]) { - this.apiConversationHistory = newHistory - await this.saveApiConversationHistory(false) + async overwriteApiConversationHistory(newHistory: ApiMessage[], persist = true) { + this.hydrateApiConversationHistory(newHistory) + if (persist) { + await this.saveApiConversationHistory(false) + } } /** @@ -1087,7 +1091,7 @@ export class Task extends EventEmitter implements TaskLike { const lastEffective = effectiveHistoryForValidation[effectiveHistoryForValidation.length - 1] const historyForValidation = lastEffective?.role === "assistant" ? effectiveHistoryForValidation : [] const validatedMessage = validateAndFixToolResultIds(userMessage, historyForValidation) - const userMessageWithTs = { ...validatedMessage, ts: Date.now() } + const userMessageWithTs = { ...validatedMessage, messageId: crypto.randomUUID(), ts: Date.now() } this.apiConversationHistory.push(userMessageWithTs as ApiMessage) const saved = await this.saveApiConversationHistory() @@ -1150,6 +1154,7 @@ export class Task extends EventEmitter implements TaskLike { } private async addToClineMessages(message: ClineMessage) { + message.messageId ??= crypto.randomUUID() this.clineMessages.push(message) const provider = this.providerRef.deref() // Unanswered asks must reach the webview before Message listeners can respond against its state. @@ -1182,13 +1187,15 @@ export class Task extends EventEmitter implements TaskLike { } } - public async overwriteClineMessages(newMessages: ClineMessage[]) { + public async overwriteClineMessages(newMessages: ClineMessage[], persist = true) { this.hydrateClineMessages(newMessages) - await this.saveClineMessages(false) + if (persist) { + await this.saveClineMessages(false) + } } private hydrateClineMessages(messages: ClineMessage[]) { - this.clineMessages = messages + this.clineMessages = ensureMessageIdentifiers(messages) restoreTodoListForTask(this) // When hydrating or overwriting messages, repopulate the cloud sync tracking Set @@ -1201,6 +1208,10 @@ export class Task extends EventEmitter implements TaskLike { } } + private hydrateApiConversationHistory(messages: ApiMessage[]) { + this.apiConversationHistory = ensureMessageIdentifiers(messages) + } + private async updateClineMessage(message: ClineMessage) { const provider = this.providerRef.deref() await provider?.postMessageToWebview({ type: "messageUpdated", clineMessage: message }) @@ -2183,7 +2194,7 @@ export class Task extends EventEmitter implements TaskLike { // task, and it was because we were waiting for resume). // This is important in case the user deletes messages without resuming // the task first. - this.apiConversationHistory = await this.getSavedApiConversationHistory() + this.hydrateApiConversationHistory(await this.getSavedApiConversationHistory()) if ( this.pendingAction && this.apiConversationHistory.some( @@ -2689,7 +2700,7 @@ export class Task extends EventEmitter implements TaskLike { // Load conversation history if not already loaded if (this.apiConversationHistory.length === 0) { - this.apiConversationHistory = await this.getSavedApiConversationHistory() + this.hydrateApiConversationHistory(await this.getSavedApiConversationHistory()) } // Add environment details to the existing last user message (which contains the tool_result) diff --git a/src/core/task/__tests__/Task.persistence.spec.ts b/src/core/task/__tests__/Task.persistence.spec.ts index 47c13a7bfd..1850de5dac 100644 --- a/src/core/task/__tests__/Task.persistence.spec.ts +++ b/src/core/task/__tests__/Task.persistence.spec.ts @@ -293,7 +293,7 @@ describe("Task persistence", () => { describe("saveApiConversationHistory", () => { it("returns true on success", async () => { - mockSaveApiMessages.mockResolvedValueOnce(undefined) + mockSaveApiMessages.mockResolvedValueOnce([]) const task = new Task({ provider: mockProvider, @@ -325,6 +325,22 @@ describe("Task persistence", () => { expect(mockSaveApiMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: false })) }) + it("can hydrate API history without persisting it", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "test task", + startTask: false, + }) + + await task.overwriteApiConversationHistory([{ role: "user", content: "merged" }], false) + + expect(task.apiConversationHistory).toEqual([ + expect.objectContaining({ role: "user", content: "merged", messageId: expect.any(String) }), + ]) + expect(mockSaveApiMessages).not.toHaveBeenCalled() + }) + it("returns false on failure", async () => { vi.useFakeTimers() @@ -374,7 +390,7 @@ describe("Task persistence", () => { }) it("snapshots the array before passing to saveApiMessages", async () => { - mockSaveApiMessages.mockResolvedValueOnce(undefined) + mockSaveApiMessages.mockResolvedValueOnce([]) const task = new Task({ provider: mockProvider, @@ -405,7 +421,7 @@ describe("Task persistence", () => { describe("saveClineMessages", () => { it("returns true on success", async () => { - mockSaveTaskMessages.mockResolvedValueOnce(undefined) + mockSaveTaskMessages.mockResolvedValueOnce([]) const task = new Task({ provider: mockProvider, @@ -432,6 +448,22 @@ describe("Task persistence", () => { expect(mockSaveTaskMessages).toHaveBeenCalledWith(expect.objectContaining({ merge: false })) }) + it("can hydrate UI history without persisting it", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "test task", + startTask: false, + }) + + await task.overwriteClineMessages([{ ts: 1, type: "say", say: "text", text: "merged" }], false) + + expect(task.clineMessages).toEqual([ + expect.objectContaining({ ts: 1, text: "merged", messageId: expect.any(String) }), + ]) + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + }) + it("returns false on failure", async () => { mockSaveTaskMessages.mockRejectedValueOnce(new Error("write error")) @@ -447,7 +479,7 @@ describe("Task persistence", () => { }) it("snapshots the array before passing to saveTaskMessages", async () => { - mockSaveTaskMessages.mockResolvedValueOnce(undefined) + mockSaveTaskMessages.mockResolvedValueOnce([]) const task = new Task({ provider: mockProvider, @@ -475,7 +507,7 @@ describe("Task persistence", () => { }) it("preserves an existing lifecycle status during metadata saves", async () => { - mockSaveTaskMessages.mockResolvedValueOnce(undefined) + mockSaveTaskMessages.mockResolvedValueOnce([]) mockTaskMetadata.mockResolvedValueOnce({ historyItem: { id: "task-with-advanced-status", @@ -1281,7 +1313,7 @@ describe("Task persistence", () => { }) it("clears userMessageContent on save success", async () => { - mockSaveApiMessages.mockResolvedValueOnce(undefined) + mockSaveApiMessages.mockResolvedValueOnce([]) const task = new Task({ provider: mockProvider, diff --git a/src/core/task/apiConversationHistory.ts b/src/core/task/apiConversationHistory.ts index b0b9959f47..d1e609c8dc 100644 --- a/src/core/task/apiConversationHistory.ts +++ b/src/core/task/apiConversationHistory.ts @@ -1,4 +1,5 @@ import { Anthropic } from "@anthropic-ai/sdk" +import crypto from "crypto" import { type ProviderSettings, getApiProtocol, getModelId, isRetiredProvider } from "@roo-code/types" @@ -57,6 +58,7 @@ function prepareAssistantMessage( const messageWithTs: any = { ...message, + messageId: crypto.randomUUID(), ...(responseId ? { id: responseId } : {}), ts: Date.now(), } @@ -127,7 +129,7 @@ function prepareUserMessage(message: Anthropic.MessageParam, apiConversationHist } const validatedMessage = validateAndFixToolResultIds(messageToAdd, historyForValidation) - return { ...validatedMessage, ts: Date.now() } + return { ...validatedMessage, messageId: crypto.randomUUID(), ts: Date.now() } } function prependContentBlock(message: any, block: any): void { diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index e73d54989a..c4eec77858 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -4101,7 +4101,7 @@ export class ClineProvider ) { parentClineMessages.push(subtaskUiMessage) } - await saveTaskMessages({ + parentClineMessages = await saveTaskMessages({ messages: parentClineMessages, taskId: parentTaskId, globalStoragePath, @@ -4191,7 +4191,7 @@ export class ClineProvider } } - await saveApiMessages({ + parentApiMessages = await saveApiMessages({ messages: parentApiMessages as any, taskId: parentTaskId, globalStoragePath, @@ -4264,12 +4264,12 @@ export class ClineProvider // 8) Inject restored histories into the in-memory instance before resuming if (parentInstance) { try { - await parentInstance.overwriteClineMessages(parentClineMessages) + await parentInstance.overwriteClineMessages(parentClineMessages, false) } catch { // non-fatal } try { - await parentInstance.overwriteApiConversationHistory(parentApiMessages as any) + await parentInstance.overwriteApiConversationHistory(parentApiMessages as any, false) } catch { // non-fatal } From e0ac8180f2375b4fc00c85140932c5ee635f0f33 Mon Sep 17 00:00:00 2001 From: Roomote Date: Sat, 5 Sep 2026 00:45:13 +0000 Subject: [PATCH 6/8] fix(history): identify restored parent messages --- src/__tests__/history-resume-delegation.spec.ts | 2 ++ src/core/webview/ClineProvider.ts | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index 1bc42b3382..ca5a03a0fe 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -349,6 +349,7 @@ describe("History resume delegation - parent metadata transitions", () => { expect.objectContaining({ messages: expect.arrayContaining([ expect.objectContaining({ + messageId: expect.any(String), type: "say", say: "subtask_result", text: "Subtask completed successfully", @@ -365,6 +366,7 @@ describe("History resume delegation - parent metadata transitions", () => { expect.objectContaining({ messages: expect.arrayContaining([ expect.objectContaining({ + messageId: expect.any(String), role: "user", content: expect.arrayContaining([ expect.objectContaining({ diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index c4eec77858..d47482b15a 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -2,6 +2,7 @@ import os from "os" import * as path from "path" import fs from "fs/promises" import EventEmitter from "events" +import crypto from "crypto" import { Anthropic } from "@anthropic-ai/sdk" import delay from "delay" @@ -4088,6 +4089,7 @@ export class ClineProvider if (!Array.isArray(parentApiMessages)) parentApiMessages = [] const subtaskUiMessage: ClineMessage = { + messageId: crypto.randomUUID(), type: "say", say: "subtask_result", text: completionResultSummary, @@ -4145,6 +4147,7 @@ export class ClineProvider // If no existing tool_result found, create a NEW user message with the tool_result if (!alreadyHasToolResult) { parentApiMessages.push({ + messageId: crypto.randomUUID(), role: "user", content: [ { @@ -4179,6 +4182,7 @@ export class ClineProvider ) if (!alreadyHasFallback) { parentApiMessages.push({ + messageId: crypto.randomUUID(), role: "user", content: [ { From d481856eb74fea10fb480891993bb738e46c38bf Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 6 Sep 2026 00:00:41 +0000 Subject: [PATCH 7/8] fix(task-history): address review feedback --- docs/architecture/task-lifecycle-model.md | 20 ++--- .../history-resume-delegation.spec.ts | 76 +++++++++++++++---- .../__tests__/mergeMessageSnapshots.spec.ts | 9 +++ .../task-persistence/mergeMessageSnapshots.ts | 4 +- src/core/webview/ClineProvider.ts | 7 +- src/eslint-suppressions.json | 2 +- 6 files changed, 90 insertions(+), 28 deletions(-) diff --git a/docs/architecture/task-lifecycle-model.md b/docs/architecture/task-lifecycle-model.md index e330a9dba8..79ecca0a6c 100644 --- a/docs/architecture/task-lifecycle-model.md +++ b/docs/architecture/task-lifecycle-model.md @@ -96,16 +96,16 @@ These are safety claims within the documented bounds. The check does not claim l The following map separates issue observations from the architectural interpretation encoded here. Open issues can change after this document is written; follow each link for current status. -| Issue and directly observed evidence | Derived protocol rule | Production transition and current check | -| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): the issue report states that a barrier-controlled two-host run reproduced an old child completion clearing a newer handoff 25/25 times. | Completion is conditional on the parent still awaiting that exact child; a live-linked child must remain owned by its parent. | `completeDelegatedChild` rejects stale authoritative input. The lifecycle explorer checks that reducer rule, while the shared-store explorer reproduces the cross-host stale-cache counterexample with an exact causal witness. | -| [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): an in-flight `saveClineMessages` can restore parent/root IDs after abandonment cleared them. | Detachment should be monotonic: later lifecycle work must not reattach an abandoned child. | `abandonDelegatedChild` clears both sides. The shared-store explorer proves the detach commit occurs, then reproduces a refreshed-cache delta that preserves interrupted status while restoring stale live-task lineage. | -| [#1453](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1453), under user report [#1279](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1279): CI observed `TaskCompleted` before restart-visible API history once; 120 local repetitions did not reproduce it, while an Alloy abstraction permits the ordering. | A completion/readiness contract must define whether completion implies restart visibility. This is a liveness/durability boundary, not only a `HistoryItem` safety transition. | Not claimed by this checker. Add a controlled persistence barrier test after the contract decision; move to temporal model checking if eventual readiness and failure handling become protocol guarantees. | -| [#921](https://github.com/Zoo-Code-Org/Zoo-Code/issues/921): delegation across parallel tabs lacks coverage for different view-local mode/profile state. | Delegation must bind an explicit immutable execution-context snapshot rather than read whichever view is focused later. | The persisted ownership transition is covered; mode/profile snapshot isolation is outside this state model and belongs in a production adapter/model-based test. | -| [#920](https://github.com/Zoo-Code-Org/Zoo-Code/issues/920): issue analysis identifies a missing cross-instance history-update test and potential lost writes. | Distinct task writes must not overwrite one another, and same-task conflicts need an explicit merge/ownership rule. | The shared-store explorer checks distinct-task writes and same-record independent deltas. Cross-instance store tests retain production API coverage, and the synchronized real-filesystem smoke test exercises the actual lock/write path without claiming exhaustive filesystem proof. | -| [#369](https://github.com/Zoo-Code-Org/Zoo-Code/issues/369) and [#372](https://github.com/Zoo-Code-Org/Zoo-Code/issues/372): planned fan-out keeps a parent live while a child runs and requires completion routing by explicit parent ID, single-writer result readiness, permit release, and orphan cleanup. | Persisted `delegated` status is ownership, not proof that the parent instance is suspended. Completion must route by IDs; scheduler resources and live-instance state need separate invariants. | Nested and sibling lifecycle ownership are covered. Scheduler permits, live/suspended parent selection, orphan cancellation, and single-writer message readiness must be added when fan-out lands; they should not be folded into `HistoryItem` fields prematurely. | -| [#1468](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1468): a late chunk from one request combined tool identity with arguments from another request; rerun passed. | Every stream accumulator needs a request/task generation key, and late events cannot mutate another scope. | Separate protocol. The [native tool-call parser request-scope model](./native-tool-call-parser-scoping-model.md), whose source of truth is `scripts/check-native-tool-call-parser-scoping.ts`, exhaustively replays bounded production-parser interleavings without adding fields to this lifecycle model. | -| [#612](https://github.com/Zoo-Code-Org/Zoo-Code/issues/612): the CLI copied a status union and omitted `interrupted`. | Lifecycle vocabulary should have one type owner. | `HistoryItemStatus` is derived from `HistoryItem`, and production/checker transitions share `taskLifecycle.ts`; consumers should import rather than copy the union. | +| Issue and directly observed evidence | Derived protocol rule | Production transition and current check | +| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): the issue report states that a barrier-controlled two-host run reproduced an old child completion clearing a newer handoff 25/25 times. | Completion is conditional on the parent still awaiting that exact child; a live-linked child must remain owned by its parent. | `completeDelegatedChild` rejects stale authoritative input. The lifecycle explorer checks that reducer rule, while the shared-store explorer reproduces the cross-host stale-cache counterexample with an exact causal witness. | +| [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): an in-flight `saveClineMessages` can restore parent/root IDs after abandonment cleared them. | Detachment should be monotonic: later lifecycle work must not reattach an abandoned child. | `abandonDelegatedChild` clears both sides. The shared-store explorer proves the detach commit occurs, then reproduces a refreshed-cache delta that preserves interrupted status while restoring stale live-task lineage. | +| [#1453](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1453), under user report [#1279](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1279): CI observed `TaskCompleted` before restart-visible API history once; 120 local repetitions did not reproduce it, while an Alloy abstraction permits the ordering. | A completion/readiness contract must define whether completion implies restart visibility. This is a liveness/durability boundary, not only a `HistoryItem` safety transition. | Not claimed by this checker. #1279 is resolved: `reopenParentFromDelegation` reads both UI and API history and saves them under the per-file advisory lock before the lifecycle transition, so the histories are durable before `TaskDelegationCompleted` fires. `restart-persistence.test.ts` is the controlled barrier test; move to temporal model checking if eventual readiness and failure handling become protocol guarantees. | +| [#921](https://github.com/Zoo-Code-Org/Zoo-Code/issues/921): delegation across parallel tabs lacks coverage for different view-local mode/profile state. | Delegation must bind an explicit immutable execution-context snapshot rather than read whichever view is focused later. | The persisted ownership transition is covered; mode/profile snapshot isolation is outside this state model and belongs in a production adapter/model-based test. | +| [#920](https://github.com/Zoo-Code-Org/Zoo-Code/issues/920): issue analysis identifies a missing cross-instance history-update test and potential lost writes. | Distinct task writes must not overwrite one another, and same-task conflicts need an explicit merge/ownership rule. | The shared-store explorer checks distinct-task writes and same-record independent deltas. Cross-instance store tests retain production API coverage, and the synchronized real-filesystem smoke test exercises the actual lock/write path without claiming exhaustive filesystem proof. | +| [#369](https://github.com/Zoo-Code-Org/Zoo-Code/issues/369) and [#372](https://github.com/Zoo-Code-Org/Zoo-Code/issues/372): planned fan-out keeps a parent live while a child runs and requires completion routing by explicit parent ID, single-writer result readiness, permit release, and orphan cleanup. | Persisted `delegated` status is ownership, not proof that the parent instance is suspended. Completion must route by IDs; scheduler resources and live-instance state need separate invariants. | Nested and sibling lifecycle ownership are covered. Scheduler permits, live/suspended parent selection, orphan cancellation, and single-writer message readiness must be added when fan-out lands; they should not be folded into `HistoryItem` fields prematurely. | +| [#1468](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1468): a late chunk from one request combined tool identity with arguments from another request; rerun passed. | Every stream accumulator needs a request/task generation key, and late events cannot mutate another scope. | Separate protocol. The [native tool-call parser request-scope model](./native-tool-call-parser-scoping-model.md), whose source of truth is `scripts/check-native-tool-call-parser-scoping.ts`, exhaustively replays bounded production-parser interleavings without adding fields to this lifecycle model. | +| [#612](https://github.com/Zoo-Code-Org/Zoo-Code/issues/612): the CLI copied a status union and omitted `interrupted`. | Lifecycle vocabulary should have one type owner. | `HistoryItemStatus` is derived from `HistoryItem`, and production/checker transitions share `taskLifecycle.ts`; consumers should import rather than copy the union. | The issue-derived cases intentionally map to bug classes rather than issue-specific flags. In particular, stale event ownership, monotonic terminal/detached state, explicit scope, and single-writer boundaries generalize to future concurrent task work. diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index ca5a03a0fe..a99ad2ca22 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -417,18 +417,26 @@ describe("History resume delegation - parent metadata transitions", () => { resumeAfterDelegation: vi.fn().mockResolvedValue(undefined), }), taskHistoryStore, - } as any) + }) vi.mocked(readTaskMessages).mockResolvedValue([{ ts: 1, type: "say", say: "text", text: "initial UI" }]) vi.mocked(readApiMessages).mockResolvedValue([{ ts: 1, role: "user", content: "initial API" }]) - vi.mocked(saveTaskMessages).mockResolvedValueOnce([ - { ts: 1, type: "say", say: "text", text: "initial UI" }, - { ts: 2, type: "say", say: "text", text: "concurrent UI" }, - ] satisfies ClineMessage[]) - vi.mocked(saveApiMessages).mockResolvedValueOnce([ - { ts: 1, role: "user", content: "initial API" }, - { ts: 2, role: "assistant", content: "concurrent API" }, - ] satisfies ApiMessage[]) + vi.mocked(saveTaskMessages).mockImplementationOnce( + async ({ messages }) => + [ + { ts: 1, type: "say", say: "text", text: "initial UI" }, + { ts: 2, type: "say", say: "text", text: "concurrent UI" }, + messages.at(-1)!, // injected subtask_result + ] as ClineMessage[], + ) + vi.mocked(saveApiMessages).mockImplementationOnce( + async ({ messages }) => + [ + { ts: 1, role: "user", content: "initial API" }, + { ts: 2, role: "assistant", content: "concurrent API" }, + messages.at(-1)!, // injected tool_result / fallback + ] as ApiMessage[], + ) await ClineProvider.prototype.reopenParentFromDelegation.call(provider, { parentTaskId: "parent-merge", @@ -437,17 +445,19 @@ describe("History resume delegation - parent metadata transitions", () => { }) expect(overwriteClineMessages).toHaveBeenCalledWith( - [ + expect.arrayContaining([ { ts: 1, type: "say", say: "text", text: "initial UI" }, { ts: 2, type: "say", say: "text", text: "concurrent UI" }, - ], + expect.objectContaining({ type: "say", say: "subtask_result", text: "Done" }), + ]), false, ) expect(overwriteApiConversationHistory).toHaveBeenCalledWith( - [ + expect.arrayContaining([ { ts: 1, role: "user", content: "initial API" }, { ts: 2, role: "assistant", content: "concurrent API" }, - ], + expect.objectContaining({ role: "user" }), + ]), false, ) }) @@ -489,6 +499,46 @@ describe("History resume delegation - parent metadata transitions", () => { expect(taskHistoryStore.atomicUpdatePair).not.toHaveBeenCalled() }) + it("does not reopen or overwrite a parent when its API history cannot be read", async () => { + const parentItem = { + id: "parent-api-read-failure", + status: "delegated", + awaitingChildId: "child-api-read-failure", + childIds: ["child-api-read-failure"], + ts: 100, + task: "Parent", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + const log = vi.fn() + const taskHistoryStore = makeTaskHistoryStoreStub( + { id: "child-api-read-failure", status: "active" }, + parentItem, + ) + const provider = makeProviderStub({ + contextProxy: { globalStorageUri: { fsPath: "/storage" } }, + getTaskWithId: vi.fn().mockResolvedValue({ historyItem: parentItem }), + getCurrentTask: vi.fn(() => ({ taskId: "child-api-read-failure" })), + taskHistoryStore, + log, + }) + vi.mocked(readTaskMessages).mockResolvedValue([]) + vi.mocked(readApiMessages).mockRejectedValue(new Error("api history unavailable")) + + const result = await ClineProvider.prototype.reopenParentFromDelegation.call(provider, { + parentTaskId: "parent-api-read-failure", + childTaskId: "child-api-read-failure", + completionResultSummary: "Child done", + }) + + expect(result).toBe(false) + expect(log).toHaveBeenCalledWith(expect.stringContaining("api history unavailable")) + expect(saveTaskMessages).not.toHaveBeenCalled() + expect(saveApiMessages).not.toHaveBeenCalled() + expect(taskHistoryStore.atomicUpdatePair).not.toHaveBeenCalled() + }) + it("reopenParentFromDelegation injects tool_result when new_task tool_use exists in API history", async () => { const parentItem = { id: "p-tool", diff --git a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts index e371b2cc4c..5dac5018f8 100644 --- a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -48,6 +48,15 @@ describe("mergeClineMessageSnapshots", () => { expect(result).toEqual([expect.objectContaining({ ts: 1, partial: false, isAnswered: true })]) }) + it("preserves finalized disk text when the incoming match is a stale partial", () => { + const result = mergeClineMessageSnapshots( + [{ messageId: "msg-1", ts: 1, type: "say", say: "text", text: "finalized text", partial: false }], + [{ messageId: "msg-1", ts: 1, type: "say", say: "text", text: "stale partial", partial: true }], + ) + + expect(result).toEqual([expect.objectContaining({ ts: 1, text: "finalized text", partial: false })]) + }) + it("keeps incoming completed and answered state when disk state is stale", () => { const result = mergeClineMessageSnapshots( [{ ts: 1, type: "ask", ask: "tool", partial: true, isAnswered: false }], diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts index 4da55062f0..5eb1df5d82 100644 --- a/src/core/task-persistence/mergeMessageSnapshots.ts +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -107,10 +107,10 @@ export function mergeClineMessageSnapshots(existing: unknown, incoming: unknown) return next } - const merged = { ...disk, ...next } if (disk.partial === false && next.partial === true) { - merged.partial = false + return disk } + const merged = { ...disk, ...next } if (disk.isAnswered === true) { merged.isAnswered = true } diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index d47482b15a..4369887363 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -4077,8 +4077,11 @@ export class ClineProvider taskId: parentTaskId, globalStoragePath, })) as any[] - } catch { - parentApiMessages = [] + } catch (error) { + this.log( + `[reopenParentFromDelegation] Failed to read API messages for parent ${parentTaskId}: ${error instanceof Error ? error.message : String(error)}`, + ) + return false } // 2) Inject synthetic records: UI subtask_result and update API tool_result diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index bc397656a4..cece760e43 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -26,7 +26,7 @@ }, "__tests__/history-resume-delegation.spec.ts": { "@typescript-eslint/no-explicit-any": { - "count": 72 + "count": 71 } }, "__tests__/migrateSettings.spec.ts": { From b90b2d94973b2d416e203c74c6d61e779621f60d Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 6 Sep 2026 02:46:52 +0000 Subject: [PATCH 8/8] test(history): kill survived string-literal mutant in ensureMessageIdentifiers --- .../__tests__/mergeMessageSnapshots.spec.ts | 34 ++++++++++++++++++ src/core/task-persistence/apiMessages.ts | 3 ++ .../task-persistence/mergeMessageSnapshots.ts | 18 ++++++++-- src/core/task/Task.ts | 10 +++++- .../task/__tests__/Task.persistence.spec.ts | 35 +++++++++++++++++++ src/core/webview/ClineProvider.ts | 11 +++--- src/eslint-suppressions.json | 2 +- 7 files changed, 103 insertions(+), 10 deletions(-) diff --git a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts index 5dac5018f8..0365dae435 100644 --- a/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -17,6 +17,31 @@ describe("ensureMessageIdentifiers", () => { expect(first.map(({ messageId }) => messageId)).toEqual(second.map(({ messageId }) => messageId)) expect(new Set(first.map(({ messageId }) => messageId)).size).toBe(3) }) + + it("assigns unique IDs in a partially upgraded snapshot with an existing legacy ID at the same timestamp", () => { + const messages: Array<{ messageId?: string; ts: number; role: string }> = [ + { messageId: "legacy:1:0", ts: 1, role: "user" }, + { ts: 1, role: "assistant" }, + ] + ensureMessageIdentifiers(messages) + expect(messages[0]!.messageId).toBe("legacy:1:0") + expect(messages[1]!.messageId).toBe("legacy:1:1") + expect(new Set(messages.map((m) => m.messageId)).size).toBe(2) + }) + + it("preserves existing messageIds and does not reassign them", () => { + const messages: Array<{ messageId?: string; ts: number; role: string }> = [ + { messageId: "stable-id", ts: 1, role: "user" }, + ] + ensureMessageIdentifiers(messages) + expect(messages[0]!.messageId).toBe("stable-id") + }) + + it("uses 'none' as the timestamp key for messages without a ts field", () => { + const messages: Array<{ messageId?: string; role: string }> = [{ role: "user" }] + ensureMessageIdentifiers(messages) + expect(messages[0]!.messageId).toBe("legacy:none:0") + }) }) describe("mergeClineMessageSnapshots", () => { @@ -57,6 +82,15 @@ describe("mergeClineMessageSnapshots", () => { expect(result).toEqual([expect.objectContaining({ ts: 1, text: "finalized text", partial: false })]) }) + it("preserves a disk message whose partial field is omitted when the incoming match is a stale partial", () => { + const result = mergeClineMessageSnapshots( + [{ messageId: "msg-1", ts: 1, type: "say", say: "text", text: "finalized text" }], + [{ messageId: "msg-1", ts: 1, type: "say", say: "text", text: "stale partial", partial: true }], + ) + + expect(result).toEqual([expect.objectContaining({ ts: 1, text: "finalized text" })]) + }) + it("keeps incoming completed and answered state when disk state is stale", () => { const result = mergeClineMessageSnapshots( [{ ts: 1, type: "ask", ask: "tool", partial: true, isAnswered: false }], diff --git a/src/core/task-persistence/apiMessages.ts b/src/core/task-persistence/apiMessages.ts index 079973240b..51749e50fe 100644 --- a/src/core/task-persistence/apiMessages.ts +++ b/src/core/task-persistence/apiMessages.ts @@ -115,6 +115,9 @@ export async function readApiMessages({ // Persist the successfully parsed legacy history before deleting its source. // The next ordinary task-history save may not happen until after user input, // so returning the in-memory data alone would leave a data-loss window. + console.warn( + `[readApiMessages] Migrating legacy API conversation history for task ${taskId} from claude_messages.json to api_conversation_history.json.`, + ) await safeWriteJson(filePath, legacyMessages, { merge: mergeApiMessageSnapshots }) try { diff --git a/src/core/task-persistence/mergeMessageSnapshots.ts b/src/core/task-persistence/mergeMessageSnapshots.ts index 5eb1df5d82..fa93be2e29 100644 --- a/src/core/task-persistence/mergeMessageSnapshots.ts +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -6,14 +6,26 @@ function isRecord(value: unknown): value is MessageRecord { } export function ensureMessageIdentifiers(messages: T[]): T[] { + // Pre-scan: collect all IDs already present so generated ones stay unique. + const usedIds = new Set() + for (const message of messages) { + if (typeof message.messageId === "string") usedIds.add(message.messageId) + } + const timestampOrdinals = new Map() for (const message of messages) { if (typeof message.messageId === "string") continue const timestampKey = typeof message.ts === "number" ? String(message.ts) : "none" - const ordinal = timestampOrdinals.get(timestampKey) ?? 0 + let ordinal = timestampOrdinals.get(timestampKey) ?? 0 + let candidate = `legacy:${timestampKey}:${ordinal}` + while (usedIds.has(candidate)) { + ordinal++ + candidate = `legacy:${timestampKey}:${ordinal}` + } timestampOrdinals.set(timestampKey, ordinal + 1) - message.messageId = `legacy:${timestampKey}:${ordinal}` + usedIds.add(candidate) + message.messageId = candidate } return messages } @@ -107,7 +119,7 @@ export function mergeClineMessageSnapshots(existing: unknown, incoming: unknown) return next } - if (disk.partial === false && next.partial === true) { + if (disk.partial !== true && next.partial === true) { return disk } const merged = { ...disk, ...next } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 63578c74b5..543084571d 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -2184,6 +2184,14 @@ export class Task extends EventEmitter implements TaskLike { } } + // Read API history before hydrating either side. If the task is aborted + // or abandoned after the UI read completes but before this point, the + // abort guard below will fire and neither history will be written. + const savedApiConversationHistory = await this.getSavedApiConversationHistory() + if (this.abort || this.abandoned) { + return + } + // Avoid a standalone write during hydration. The resume ask will persist only // after all history reads succeed and the task is still active. this.hydrateClineMessages(modifiedClineMessages) @@ -2194,7 +2202,7 @@ export class Task extends EventEmitter implements TaskLike { // task, and it was because we were waiting for resume). // This is important in case the user deletes messages without resuming // the task first. - this.hydrateApiConversationHistory(await this.getSavedApiConversationHistory()) + this.hydrateApiConversationHistory(savedApiConversationHistory) if ( this.pendingAction && this.apiConversationHistory.some( diff --git a/src/core/task/__tests__/Task.persistence.spec.ts b/src/core/task/__tests__/Task.persistence.spec.ts index 1850de5dac..9956b74fb7 100644 --- a/src/core/task/__tests__/Task.persistence.spec.ts +++ b/src/core/task/__tests__/Task.persistence.spec.ts @@ -1239,6 +1239,41 @@ describe("Task persistence", () => { expect(mockSaveTaskMessages).not.toHaveBeenCalled() }) + it("stops before hydrating either history when the task is evicted during the API read", async () => { + const apiMessagesDeferred = + createDeferred }>>() + mockReadTaskMessages.mockResolvedValue([{ ts: 1, type: "say", say: "text", text: "UI message" }]) + mockReadApiMessages.mockReturnValue(apiMessagesDeferred.promise) + + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + historyItem: { + id: "issue-1279-evict-during-api-read", + number: 1, + ts: 1, + task: "Original task", + tokensIn: 10, + tokensOut: 5, + totalCost: 0.001, + }, + startTask: false, + }) + const resumePromise = getTaskPersistenceAccess(task).resumeTaskFromHistory() + await vi.waitFor(() => expect(mockReadApiMessages).toHaveBeenCalled()) + + await task.abortTask(true) + mockSaveTaskMessages.mockClear() + vi.mocked(mockProvider.updateTaskHistory).mockClear() + apiMessagesDeferred.resolve([{ role: "user", content: [{ type: "text", text: "API message" }] }]) + await resumePromise + + // Neither UI nor API history should have been hydrated or persisted. + expect(task.clineMessages).toHaveLength(0) + expect(task.apiConversationHistory).toHaveLength(0) + expect(mockSaveTaskMessages).not.toHaveBeenCalled() + }) + it("stops after API history hydration when the task is aborted", async () => { const apiMessagesDeferred = createDeferred }>>() diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 4369887363..7f21a049e7 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -111,6 +111,7 @@ import { Task } from "../task/Task" import { webviewMessageHandler } from "./webviewMessageHandler" import type { ClineMessage, TodoItem } from "@roo-code/types" import { + type ApiMessage, readApiMessages, saveApiMessages, saveTaskMessages, @@ -4071,12 +4072,12 @@ export class ClineProvider return false } - let parentApiMessages: any[] = [] + let parentApiMessages: ApiMessage[] = [] try { - parentApiMessages = (await readApiMessages({ + parentApiMessages = await readApiMessages({ taskId: parentTaskId, globalStoragePath, - })) as any[] + }) } catch (error) { this.log( `[reopenParentFromDelegation] Failed to read API messages for parent ${parentTaskId}: ${error instanceof Error ? error.message : String(error)}`, @@ -4199,7 +4200,7 @@ export class ClineProvider } parentApiMessages = await saveApiMessages({ - messages: parentApiMessages as any, + messages: parentApiMessages, taskId: parentTaskId, globalStoragePath, merge: true, @@ -4276,7 +4277,7 @@ export class ClineProvider // non-fatal } try { - await parentInstance.overwriteApiConversationHistory(parentApiMessages as any, false) + await parentInstance.overwriteApiConversationHistory(parentApiMessages, false) } catch { // non-fatal } diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index cece760e43..381cf0c1e0 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -1026,7 +1026,7 @@ }, "core/webview/ClineProvider.ts": { "@typescript-eslint/no-explicit-any": { - "count": 12 + "count": 8 } }, "core/webview/__tests__/ClineProvider.apiHandlerRebuild.spec.ts": {