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/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/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 d3a24a3140..a99ad2ca22 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([]) @@ -347,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", @@ -354,6 +357,7 @@ describe("History resume delegation - parent metadata transitions", () => { ]), taskId: "p1", globalStoragePath: "/storage", + merge: true, }), ) @@ -362,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({ @@ -373,6 +378,7 @@ describe("History resume delegation - parent metadata transitions", () => { ]), taskId: "p1", globalStoragePath: "/storage", + merge: true, }), ) @@ -384,6 +390,155 @@ 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, + }) + + 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).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", + childTaskId: "child-merge", + completionResultSummary: "Done", + }) + + 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, + ) + }) + + 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 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, + 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(readApiMessages).not.toHaveBeenCalled() + expect(saveTaskMessages).not.toHaveBeenCalled() + expect(saveApiMessages).not.toHaveBeenCalled() + 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__/apiMessages.spec.ts b/src/core/task-persistence/__tests__/apiMessages.spec.ts index aa725f4744..c74ef6b150 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" -import { readApiMessages } from "../apiMessages" +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,11 +75,109 @@ describe("apiMessages.readApiMessages", () => { const oldPath = path.join(taskDir, "claude_messages.json") await fs.writeFile(oldPath, JSON.stringify({ key: "value" }), "utf8") - const result = await readApiMessages({ + await expect(readApiMessages({ taskId, globalStoragePath: tmpBaseDir })).rejects.toMatchObject({ + kind: "invalid", + }) + }) + + 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("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" }) + 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) + }) +}) + +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", + ) + + const savedMessages = 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 }), + ]) + 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 () => { + 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(result).toEqual([]) + 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 new file mode 100644 index 0000000000..0365dae435 --- /dev/null +++ b/src/core/task-persistence/__tests__/mergeMessageSnapshots.spec.ts @@ -0,0 +1,241 @@ +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) + }) + + 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", () => { + 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("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("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 }], + [{ 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( + [{ 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("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") + }) +}) + +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( + [ + { 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 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( + [ + { 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 c6bc360c05..e494b5b594 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,10 +80,29 @@ 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" }] + 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, + }) + + 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" }), + ]) + }) }) 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 +110,65 @@ 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", }) + }) - expect(result).toEqual([]) + 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") + + 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..51749e50fe 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 { fileExistsAtPath } from "../../utils/fs" - +import { safeWriteJson } from "../../utils/safeWriteJson" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" +import { ensureMessageIdentifiers, mergeApiMessageSnapshots } from "./mergeMessageSnapshots" +import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" export type ApiMessage = Anthropic.MessageParam & { + messageId?: string ts?: number isSummary?: boolean id?: string @@ -37,6 +38,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, @@ -47,75 +101,63 @@ 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") - - 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 currentMessages = await readApiMessagesFile(taskId, filePath) + if (currentMessages !== undefined) { + return currentMessages } - // 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}`, + const oldPath = path.join(taskDir, "claude_messages.json") + const legacyMessages = await readApiMessagesFile(taskId, oldPath) + if (legacyMessages === undefined) { + 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. + console.warn( + `[readApiMessages] Migrating legacy API conversation history for task ${taskId} from claude_messages.json to api_conversation_history.json.`, ) - return [] + await safeWriteJson(filePath, legacyMessages, { merge: mergeApiMessageSnapshots }) + + 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({ messages, taskId, globalStoragePath, + merge = false, }: { messages: ApiMessage[] 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) + 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 463df8a0bb..14adeedc68 100644 --- a/src/core/task-persistence/index.ts +++ b/src/core/task-persistence/index.ts @@ -1,6 +1,18 @@ -export { type ApiMessage, readApiMessages, saveApiMessages } from "./apiMessages" -export { readTaskMessages, saveTaskMessages } from "./taskMessages" +export { + type ApiMessage, + ApiMessagesReadError, + type ApiMessagesReadErrorKind, + readApiMessages, + saveApiMessages, +} from "./apiMessages" +export { + readTaskMessages, + saveTaskMessages, + TaskMessagesReadError, + 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 new file mode 100644 index 0000000000..fa93be2e29 --- /dev/null +++ b/src/core/task-persistence/mergeMessageSnapshots.ts @@ -0,0 +1,135 @@ +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[] { + // 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" + 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) + usedIds.add(candidate) + message.messageId = candidate + } + return messages +} + +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 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]) => { + const consumed = consumedByTimestamp.get(timestamp) + return messages.filter((_message, index) => !consumed?.has(index)) + }) + + 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)) + + timestampMerged.push(...existingLegacy.slice(incomingLegacyCount)) + return timestampMerged +} + +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 + } + + if (disk.partial !== true && next.partial === true) { + return disk + } + const merged = { ...disk, ...next } + if (disk.isAnswered === true) { + merged.isAnswered = true + } + return merged + }) +} + +export function mergeApiMessageSnapshots(existing: unknown, incoming: unknown): unknown { + return mergeTimestampedSnapshots(existing, incoming) +} 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 cee66432d9..89ec372c61 100644 --- a/src/core/task-persistence/taskMessages.ts +++ b/src/core/task-persistence/taskMessages.ts @@ -1,13 +1,25 @@ -import { safeWriteJson } from "../../utils/safeWriteJson" import * as path from "path" -import * as fs from "fs/promises" import type { ClineMessage } from "@roo-code/types" -import { fileExistsAtPath } from "../../utils/fs" - +import { safeWriteJson } from "../../utils/safeWriteJson" import { GlobalFileNames } from "../../shared/globalFileNames" import { getTaskDirectoryPath } from "../../utils/storage" +import { ensureMessageIdentifiers, mergeClineMessageSnapshots } from "./mergeMessageSnapshots" +import { getErrorCode, readFileWithMissingRetry } from "./readFileWithMissingRetry" + +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 @@ -20,37 +32,60 @@ 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 readFileWithMissingRetry(filePath) + } catch (error) { + const kind = getErrorCode(error) === "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) } - return [] + if (!Array.isArray(parsedData)) { + throw new TaskMessagesReadError( + "invalid", + `Task messages for ${taskId} at ${filePath} must be an array, got ${typeof parsedData}`, + ) + } + + return parsedData } 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): Promise { + ensureMessageIdentifiers(messages) const taskDir = await getTaskDirectoryPath(globalStoragePath, taskId) const filePath = path.join(taskDir, GlobalFileNames.uiMessages) - await safeWriteJson(filePath, messages) + 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 4c2d77b5ae..543084571d 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() + 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() @@ -1104,12 +1108,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) { @@ -1149,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. @@ -1181,21 +1187,31 @@ export class Task extends EventEmitter implements TaskLike { } } - public async overwriteClineMessages(newMessages: ClineMessage[]) { - this.clineMessages = newMessages + public async overwriteClineMessages(newMessages: ClineMessage[], persist = true) { + this.hydrateClineMessages(newMessages) + if (persist) { + await this.saveClineMessages(false) + } + } + + private hydrateClineMessages(messages: ClineMessage[]) { + this.clineMessages = ensureMessageIdentifiers(messages) restoreTodoListForTask(this) - await this.saveClineMessages() - // 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) } } } + 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 }) @@ -1215,12 +1231,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) { @@ -2108,7 +2125,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 +2141,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, @@ -2144,6 +2155,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 @@ -2162,8 +2184,17 @@ export class Task extends EventEmitter implements TaskLike { } } - await this.overwriteClineMessages(modifiedClineMessages) - this.clineMessages = await this.getSavedClineMessages() + // 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) // Now present the cline messages to the user and ask if they want to // resume (NOTE: we ran into a bug before where the @@ -2171,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.apiConversationHistory = await this.getSavedApiConversationHistory() + this.hydrateApiConversationHistory(savedApiConversationHistory) if ( this.pendingAction && this.apiConversationHistory.some( @@ -2193,6 +2224,10 @@ export class Task extends EventEmitter implements TaskLike { return } + if (this.abort || this.abandoned) { + return + } + const lastClineMessage = this.clineMessages .slice() .reverse() @@ -2673,7 +2708,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 671bd7d4b7..9956b74fb7 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, @@ -309,6 +309,36 @@ 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("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 () => { @@ -360,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, @@ -391,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, @@ -402,6 +432,36 @@ 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("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 () => { @@ -419,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, @@ -447,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", @@ -767,13 +827,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 +1125,193 @@ 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() + }) + + 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 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 }>>() + 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) + // 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() + }) + }) + // ── flushPendingToolResultsToHistory — save failure/success ─────────── describe("flushPendingToolResultsToHistory persistence", () => { @@ -1104,7 +1348,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/__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/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 394da7c10f..7f21a049e7 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" @@ -110,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, @@ -4063,18 +4065,24 @@ 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[] = [] + let parentApiMessages: ApiMessage[] = [] try { - parentApiMessages = (await readApiMessages({ + parentApiMessages = await readApiMessages({ 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 @@ -4085,6 +4093,7 @@ export class ClineProvider if (!Array.isArray(parentApiMessages)) parentApiMessages = [] const subtaskUiMessage: ClineMessage = { + messageId: crypto.randomUUID(), type: "say", say: "subtask_result", text: completionResultSummary, @@ -4098,7 +4107,12 @@ export class ClineProvider ) { parentClineMessages.push(subtaskUiMessage) } - await saveTaskMessages({ messages: parentClineMessages, taskId: parentTaskId, globalStoragePath }) + parentClineMessages = 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 @@ -4137,6 +4151,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: [ { @@ -4171,6 +4186,7 @@ export class ClineProvider ) if (!alreadyHasFallback) { parentApiMessages.push({ + messageId: crypto.randomUUID(), role: "user", content: [ { @@ -4183,7 +4199,12 @@ export class ClineProvider } } - await saveApiMessages({ messages: parentApiMessages as any, taskId: parentTaskId, globalStoragePath }) + parentApiMessages = await saveApiMessages({ + messages: parentApiMessages, + 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 @@ -4251,12 +4272,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, false) } catch { // non-fatal } diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index bc397656a4..381cf0c1e0 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": { @@ -1026,7 +1026,7 @@ }, "core/webview/ClineProvider.ts": { "@typescript-eslint/no-explicit-any": { - "count": 12 + "count": 8 } }, "core/webview/__tests__/ClineProvider.apiHandlerRebuild.spec.ts": {