From ac5c2e035a479b9b2fbd04726a319c273e3a8d29 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 18 Jul 2026 01:41:54 +0800 Subject: [PATCH 1/4] fix(core): serialize task history updates across parallel tabs --- src/core/task-persistence/TaskHistoryLock.ts | 100 ++++++++++++ .../__tests__/TaskHistoryLock.spec.ts | 153 ++++++++++++++++++ src/core/webview/ClineProvider.ts | 118 ++++++++------ .../ClineProvider.taskHistory.spec.ts | 127 +++++++++++++++ src/shared/globalFileNames.ts | 1 + 5 files changed, 447 insertions(+), 52 deletions(-) create mode 100644 src/core/task-persistence/TaskHistoryLock.ts create mode 100644 src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts diff --git a/src/core/task-persistence/TaskHistoryLock.ts b/src/core/task-persistence/TaskHistoryLock.ts new file mode 100644 index 0000000000..bbc0f7de7c --- /dev/null +++ b/src/core/task-persistence/TaskHistoryLock.ts @@ -0,0 +1,100 @@ +import * as fs from "fs/promises" +import * as path from "path" +import * as lockfile from "proper-lockfile" + +import { GlobalFileNames } from "../../shared/globalFileNames" +import { getStorageBasePath } from "../../utils/storage" + +/** + * Cross-process lock for task history mutations. + * + * Multiple `ClineProvider` instances may live in separate extension-host + * processes (for example VS Code windows) while sharing the same task history + * storage. Each process has its own `TaskHistoryStore`, so an in-memory mutex is + * not sufficient. This lock serializes mutations by taking an exclusive advisory + * lock on the shared `tasks/_history.lock` file. + */ +export class TaskHistoryLock { + private queue: Promise = Promise.resolve() + + /** + * Acquires the shared task-history lock and executes `fn` while holding it. + * + * The lock file is scoped to the effective storage root (including custom + * storage path resolution) so all windows/processes targeting the same history + * store contend on the same file. + */ + async withLock(globalStoragePath: string, fn: () => Promise): Promise { + const result = this.queue.then( + async () => { + const lockFilePath = await this.getLockFilePath(globalStoragePath) + return this.runWithFileLock(lockFilePath, fn) + }, + async () => { + const lockFilePath = await this.getLockFilePath(globalStoragePath) + return this.runWithFileLock(lockFilePath, fn) + }, + ) + + this.queue = result.then( + () => undefined, + () => undefined, + ) + + return result + } + + /** + * Clears in-process queues. File locks held by other processes are not affected. + */ + reset(): void { + this.queue = Promise.resolve() + } + + async getLockFilePath(globalStoragePath: string): Promise { + const basePath = await getStorageBasePath(globalStoragePath) + const tasksDir = path.join(basePath, "tasks") + await fs.mkdir(tasksDir, { recursive: true }) + const lockFilePath = path.join(tasksDir, GlobalFileNames.historyLock) + + try { + await fs.open(lockFilePath, "a").then((handle) => handle.close()) + } catch (error) { + console.error(`[TaskHistoryLock] Failed to create lock file at ${lockFilePath}:`, error) + throw error + } + + return lockFilePath + } + + private async runWithFileLock(lockFilePath: string, fn: () => Promise): Promise { + let releaseLock: (() => Promise) | undefined + + try { + releaseLock = await lockfile.lock(lockFilePath, { + stale: 31000, + update: 10000, + realpath: false, + retries: { + retries: 10, + factor: 2, + minTimeout: 50, + maxTimeout: 1000, + }, + onCompromised: (err) => { + console.error(`[TaskHistoryLock] Lock at ${lockFilePath} was compromised:`, err) + throw err + }, + }) + + return await fn() + } finally { + if (releaseLock) { + await releaseLock() + } + } + } +} + +// Singleton instance shared across all ClineProvider instances in this process. +export const taskHistoryLock = new TaskHistoryLock() diff --git a/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts new file mode 100644 index 0000000000..1b29d1b267 --- /dev/null +++ b/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts @@ -0,0 +1,153 @@ +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" +import { fork, type ChildProcess } from "child_process" + +import { TaskHistoryLock } from "../TaskHistoryLock" + +vi.mock("../../../utils/storage", () => ({ + getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath), +})) + +const waitForMessage = (child: ChildProcess, expected: string): Promise => + new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + child.off("message", onMessage) + reject(new Error(`Timed out waiting for child process message: ${expected}`)) + }, 5000) + + const onMessage = (message: unknown) => { + if (message === expected) { + clearTimeout(timeout) + child.off("message", onMessage) + resolve() + } + } + + child.on("message", onMessage) + }) + +describe("TaskHistoryLock", () => { + let tmpDir: string + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-lock-")) + }) + + afterEach(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }) + }) + + it("serializes concurrent operations", async () => { + const lock = new TaskHistoryLock() + let activeCount = 0 + let maxActiveCount = 0 + const order: string[] = [] + + const run = (id: string) => + lock.withLock(tmpDir, async () => { + activeCount++ + maxActiveCount = Math.max(maxActiveCount, activeCount) + order.push(`start:${id}`) + await new Promise((resolve) => setTimeout(resolve, 5)) + order.push(`end:${id}`) + activeCount-- + return id + }) + + const results = await Promise.all([run("a"), run("b"), run("c")]) + + expect(results).toEqual(["a", "b", "c"]) + expect(maxActiveCount).toBe(1) + expect(order).toHaveLength(6) + for (const id of ["a", "b", "c"]) { + expect(order).toContain(`start:${id}`) + expect(order).toContain(`end:${id}`) + expect(order.indexOf(`start:${id}`)).toBeLessThan(order.indexOf(`end:${id}`)) + } + }) + + it("continues processing after a previous operation rejects", async () => { + const lock = new TaskHistoryLock() + const order: string[] = [] + + const failed = lock.withLock(tmpDir, async () => { + order.push("start:fail") + throw new Error("simulated failure") + }) + + const succeeded = lock.withLock(tmpDir, async () => { + order.push("start:success") + return "ok" + }) + + await expect(failed).rejects.toThrow("simulated failure") + await expect(succeeded).resolves.toBe("ok") + expect(order).toEqual(["start:fail", "start:success"]) + }) + + it("waits for an independent process holding the same lock file", async () => { + const lock = new TaskHistoryLock() + const lockFilePath = await lock.getLockFilePath(tmpDir) + const childScriptPath = path.join(tmpDir, "hold-history-lock.cjs") + await fs.writeFile( + childScriptPath, + ` +const lockfile = require("proper-lockfile") + +let release + +async function main() { + release = await lockfile.lock(process.argv[2], { + stale: 31000, + update: 10000, + realpath: false, + }) + process.send?.("locked") +} + +process.on("message", async (message) => { + if (message === "release") { + await release?.() + process.send?.("released") + process.exit(0) + } +}) + +main().catch((error) => { + process.send?.({ error: error instanceof Error ? error.message : String(error) }) + process.exit(1) +}) +`, + "utf8", + ) + + const child = fork(childScriptPath, [lockFilePath], { stdio: ["ignore", "ignore", "ignore", "ipc"] }) + try { + await waitForMessage(child, "locked") + + let enteredCriticalSection = false + const blocked = lock.withLock(tmpDir, async () => { + enteredCriticalSection = true + return "acquired" + }) + + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(enteredCriticalSection).toBe(false) + + child.send("release") + await waitForMessage(child, "released") + await expect(blocked).resolves.toBe("acquired") + expect(enteredCriticalSection).toBe(true) + } finally { + if (!child.killed) { + child.kill() + } + } + }) + + it("reset is a no-op for file-based locking", () => { + const lock = new TaskHistoryLock() + expect(() => lock.reset()).not.toThrow() + }) +}) diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 6e5545a1f2..edb8f4b328 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -109,6 +109,7 @@ import { TaskHistoryStore, assertValidTransition, } from "../task-persistence" +import { taskHistoryLock } from "../task-persistence/TaskHistoryLock" import { readTaskMessages } from "../task-persistence/taskMessages" import { getNonce } from "./getNonce" import { getUri } from "./getUri" @@ -385,7 +386,9 @@ export class ClineProvider if (legacyHistory.length > 0) { this.log(`[initializeTaskHistoryStore] Migrating ${legacyHistory.length} entries from globalState`) - await this.taskHistoryStore.migrateFromGlobalState(legacyHistory) + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => + this.taskHistoryStore.migrateFromGlobalState(legacyHistory), + ) } await this.context.globalState.update(migrationKey, true) @@ -2049,9 +2052,13 @@ export class ClineProvider } } - // Delete all tasks from state in one batch - await this.taskHistoryStore.deleteMany(allIdsToDelete) - this.recentTasksCache = undefined + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { + // Delete all tasks from state in one batch + await this.taskHistoryStore.deleteMany(allIdsToDelete) + this.recentTasksCache = undefined + + await this.postStateToWebview() + }) // Delete associated shadow repositories or branches and task directories const globalStorageDir = this.contextProxy.globalStorageUri.fsPath @@ -2079,8 +2086,6 @@ export class ClineProvider ) } } - - await this.postStateToWebview() } catch (error) { // If task is not found, just remove it from state if (error instanceof Error && error.message === "Task not found") { @@ -2092,10 +2097,12 @@ export class ClineProvider } async deleteTaskFromState(id: string) { - await this.taskHistoryStore.delete(id) - this.recentTasksCache = undefined + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { + await this.taskHistoryStore.delete(id) + this.recentTasksCache = undefined - await this.postStateToWebview() + await this.postStateToWebview() + }) } async refreshWorkspace() { @@ -2736,17 +2743,20 @@ export class ClineProvider async updateTaskHistory(item: HistoryItem, options: { broadcast?: boolean } = {}): Promise { const { broadcast = true } = options - const history = await this.taskHistoryStore.upsert(item) - this.recentTasksCache = undefined + // Serialize all task history mutations across parallel tabs using the shared lock. + return taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { + const history = await this.taskHistoryStore.upsert(item) + this.recentTasksCache = undefined - // Broadcast the updated history to the webview if requested. - // Prefer per-item updates to avoid repeatedly cloning/sending the full history. - if (broadcast && this.isViewLaunched) { - const updatedItem = this.taskHistoryStore.get(item.id) ?? item - await this.postMessageToWebview({ type: "taskHistoryItemUpdated", taskHistoryItem: updatedItem }) - } + // Broadcast the updated history to the webview if requested. + // Prefer per-item updates to avoid repeatedly cloning/sending the full history. + if (broadcast && this.isViewLaunched) { + const updatedItem = this.taskHistoryStore.get(item.id) ?? item + await this.postMessageToWebview({ type: "taskHistoryItemUpdated", taskHistoryItem: updatedItem }) + } - return history + return history + }) } /** @@ -3579,17 +3589,19 @@ export class ClineProvider // write, and the pure updater cannot re-enter the lock (no deadlock). // Broadcast and cache invalidation happen outside the lock after it releases. try { - await this.taskHistoryStore.atomicReadAndUpdate(parentTaskId, (historyItem) => { - assertValidTransition(historyItem.status, "delegated") - const childIds = Array.from(new Set([...(historyItem.childIds ?? []), child.taskId])) - return { - ...historyItem, - status: "delegated", - delegatedToId: child.taskId, - awaitingChildId: child.taskId, - childIds, - } - }) + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => + this.taskHistoryStore.atomicReadAndUpdate(parentTaskId, (historyItem) => { + assertValidTransition(historyItem.status, "delegated") + const childIds = Array.from(new Set([...(historyItem.childIds ?? []), child.taskId])) + return { + ...historyItem, + status: "delegated", + delegatedToId: child.taskId, + awaitingChildId: child.taskId, + childIds, + } + }), + ) this.recentTasksCache = undefined if (this.isViewLaunched) { const updatedItem = this.taskHistoryStore.get(parentTaskId) @@ -3827,29 +3839,31 @@ export class ClineProvider // any concurrent write that landed between step 1 and the lock acquisition // is preserved rather than silently overwritten. let updatedHistory!: typeof historyItem - await this.taskHistoryStore.atomicUpdatePair( - childTaskId, - parentTaskId, - (child) => { - assertValidTransition(child.status, "completed") - return { ...child, status: "completed" as const, completionResultSummary } - }, - (parent) => { - if (parent.status !== "active") { - assertValidTransition(parent.status, "active") - } - const childIds = Array.from(new Set([...(parent.childIds ?? []), childTaskId])) - updatedHistory = { - ...parent, - status: "active" as const, - completedByChildId: childTaskId, - completionResultSummary, - awaitingChildId: undefined, - delegatedToId: undefined, - childIds, - } - return updatedHistory - }, + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => + this.taskHistoryStore.atomicUpdatePair( + childTaskId, + parentTaskId, + (child) => { + assertValidTransition(child.status, "completed") + return { ...child, status: "completed" as const, completionResultSummary } + }, + (parent) => { + if (parent.status !== "active") { + assertValidTransition(parent.status, "active") + } + const childIds = Array.from(new Set([...(parent.childIds ?? []), childTaskId])) + updatedHistory = { + ...parent, + status: "active" as const, + completedByChildId: childTaskId, + completionResultSummary, + awaitingChildId: undefined, + delegatedToId: undefined, + childIds, + } + return updatedHistory + }, + ), ) this.recentTasksCache = undefined diff --git a/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts b/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts index 4a76ab4858..f57c0bd6ab 100644 --- a/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts @@ -5,6 +5,7 @@ import type { HistoryItem, ExtensionMessage } from "@roo-code/types" import { TelemetryService } from "@roo-code/telemetry" import { ContextProxy } from "../../config/ContextProxy" +import { taskHistoryLock } from "../../task-persistence/TaskHistoryLock" import { ClineProvider } from "../ClineProvider" // Mock setup @@ -116,7 +117,11 @@ vi.mock("vscode", () => ({ showInformationMessage: vi.fn(), showWarningMessage: vi.fn(), showErrorMessage: vi.fn(), + createTextEditorDecorationType: vi.fn(() => ({ dispose: vi.fn() })), onDidChangeActiveTextEditor: vi.fn(() => ({ dispose: vi.fn() })), + tabGroups: { + onDidChangeTabs: vi.fn(() => ({ dispose: vi.fn() })), + }, }, workspace: { getConfiguration: vi.fn().mockReturnValue({ @@ -132,6 +137,12 @@ vi.mock("vscode", () => ({ onDidChangeTextDocument: vi.fn(() => ({ dispose: vi.fn() })), onDidOpenTextDocument: vi.fn(() => ({ dispose: vi.fn() })), onDidCloseTextDocument: vi.fn(() => ({ dispose: vi.fn() })), + createFileSystemWatcher: vi.fn(() => ({ + onDidCreate: vi.fn(() => ({ dispose: vi.fn() })), + onDidDelete: vi.fn(() => ({ dispose: vi.fn() })), + onDidChange: vi.fn(() => ({ dispose: vi.fn() })), + dispose: vi.fn(), + })), }, env: { uriScheme: "vscode", @@ -240,6 +251,7 @@ vi.mock("@roo-code/cloud", () => ({ getOrganizationMemberships: vi.fn().mockResolvedValue([]), getUserSettings: vi.fn().mockReturnValue(null), isTaskSyncEnabled: vi.fn().mockReturnValue(false), + off: vi.fn(), } }, }, @@ -260,6 +272,8 @@ describe("ClineProvider Task History Synchronization", () => { beforeEach(async () => { vi.clearAllMocks() + taskHistoryLock.reset() + vi.spyOn(taskHistoryLock, "withLock").mockImplementation(async (_globalStoragePath, fn) => fn()) if (!TelemetryService.hasInstance()) { TelemetryService.createInstance([]) @@ -779,5 +793,118 @@ describe("ClineProvider Task History Synchronization", () => { // The second write (tokensIn: 222) should be the last one since writes are serialized expect(item!.tokensIn).toBe(222) }) + + it("routes concurrent updateTaskHistory calls from two parallel instances through the shared lock", async () => { + const provider2 = new ClineProvider( + mockContext, + mockOutputChannel, + "sidebar", + new ContextProxy(mockContext), + ) + await provider2.taskHistoryStore.initialized + + const provider1Upsert = vi + .spyOn(provider.taskHistoryStore, "upsert") + .mockImplementation(async (item: HistoryItem) => [item]) + const provider2Upsert = vi + .spyOn(provider2.taskHistoryStore, "upsert") + .mockImplementation(async (item: HistoryItem) => [item]) + + try { + await Promise.all([ + provider.updateTaskHistory(createHistoryItem({ id: "parallel-provider-1", task: "Provider 1" }), { + broadcast: false, + }), + provider2.updateTaskHistory(createHistoryItem({ id: "parallel-provider-2", task: "Provider 2" }), { + broadcast: false, + }), + ]) + + expect(provider1Upsert).toHaveBeenCalledTimes(1) + expect(provider2Upsert).toHaveBeenCalledTimes(1) + expect(taskHistoryLock.withLock).toHaveBeenCalledWith( + mockContext.globalStorageUri.fsPath, + expect.any(Function), + ) + expect(taskHistoryLock.withLock).toHaveBeenCalledTimes(2) + } finally { + await provider2.dispose() + } + }) + + it("routes 5+ concurrent updateTaskHistory calls from different tabs through the shared lock", async () => { + const providers = [provider] + + for (let i = 1; i < 5; i++) { + const nextProvider = new ClineProvider( + mockContext, + mockOutputChannel, + "sidebar", + new ContextProxy(mockContext), + ) + await nextProvider.taskHistoryStore.initialized + providers.push(nextProvider) + } + + try { + providers.forEach((currentProvider) => { + vi.spyOn(currentProvider.taskHistoryStore, "upsert").mockImplementation( + async (item: HistoryItem) => [item], + ) + }) + + await Promise.all( + providers.map((currentProvider, index) => + currentProvider.updateTaskHistory( + createHistoryItem({ id: `parallel-tab-${index}`, task: `Parallel Tab ${index}` }), + { broadcast: false }, + ), + ), + ) + + expect(taskHistoryLock.withLock).toHaveBeenCalledTimes(5) + for (const call of vi.mocked(taskHistoryLock.withLock).mock.calls) { + expect(call[0]).toBe(mockContext.globalStorageUri.fsPath) + } + } finally { + for (const currentProvider of providers.slice(1)) { + await currentProvider.dispose() + } + } + }) + + it("routes normal deleteTaskWithId deleteMany mutations through the shared lock", async () => { + const deleteManySpy = vi.spyOn(provider.taskHistoryStore, "deleteMany").mockResolvedValue(undefined) + const postStateSpy = vi.spyOn(provider, "postStateToWebview").mockResolvedValue(undefined) + vi.spyOn(provider, "getTaskWithId").mockResolvedValue({ + historyItem: createHistoryItem({ id: "delete-normal", task: "Delete normal" }), + taskDirPath: "/test/task/path", + apiConversationHistoryFilePath: "/test/task/path/api_conversation_history.json", + uiMessagesFilePath: "/test/task/path/ui_messages.json", + apiConversationHistory: [], + }) + + await provider.deleteTaskWithId("delete-normal", false) + + expect(deleteManySpy).toHaveBeenCalledWith(["delete-normal"]) + expect(taskHistoryLock.withLock).toHaveBeenCalledWith( + mockContext.globalStorageUri.fsPath, + expect.any(Function), + ) + expect(postStateSpy).toHaveBeenCalled() + }) + + it("routes fallback deleteTaskFromState mutations through the shared lock", async () => { + const deleteSpy = vi.spyOn(provider.taskHistoryStore, "delete").mockResolvedValue(undefined) + vi.spyOn(provider, "postStateToWebview").mockResolvedValue(undefined) + + await provider.deleteTaskFromState("cross-delete") + + expect(deleteSpy).toHaveBeenCalledWith("cross-delete") + expect(taskHistoryLock.withLock).toHaveBeenCalledWith( + mockContext.globalStorageUri.fsPath, + expect.any(Function), + ) + }) }) }) diff --git a/src/shared/globalFileNames.ts b/src/shared/globalFileNames.ts index 0b54ff6809..4400dfa212 100644 --- a/src/shared/globalFileNames.ts +++ b/src/shared/globalFileNames.ts @@ -6,4 +6,5 @@ export const GlobalFileNames = { taskMetadata: "task_metadata.json", historyItem: "history_item.json", historyIndex: "_index.json", + historyLock: "_history.lock", } From 370e9c6f7452dddc479f0239d18c87f2f8d82f1d Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 18 Jul 2026 03:15:01 +0800 Subject: [PATCH 2/4] fix(task-persistence): guard concurrent history mutations --- .../history-resume-delegation.spec.ts | 72 +++++ src/__tests__/provider-delegation.spec.ts | 12 + src/core/task-persistence/TaskHistoryLock.ts | 6 +- src/core/task-persistence/TaskHistoryStore.ts | 246 ++++++++++-------- .../__tests__/TaskHistoryLock.spec.ts | 167 +++--------- .../__tests__/TaskHistoryStore.spec.ts | 31 +++ src/core/webview/ClineProvider.ts | 140 +++++----- .../ClineProvider.taskHistory.spec.ts | 80 +++++- 8 files changed, 444 insertions(+), 310 deletions(-) diff --git a/src/__tests__/history-resume-delegation.spec.ts b/src/__tests__/history-resume-delegation.spec.ts index 0be2d28a3f..1e428faafa 100644 --- a/src/__tests__/history-resume-delegation.spec.ts +++ b/src/__tests__/history-resume-delegation.spec.ts @@ -39,6 +39,12 @@ vi.mock("../core/task-persistence", async (importOriginal) => { saveTaskMessages: vi.fn().mockResolvedValue(undefined), } }) +vi.mock("../core/task-persistence/TaskHistoryLock", () => ({ + taskHistoryLock: { + withLock: vi.fn(async (_globalStoragePath: string, fn: () => Promise) => fn()), + reset: vi.fn(), + }, +})) import { ClineProvider } from "../core/webview/ClineProvider" import { readTaskMessages } from "../core/task-persistence/taskMessages" @@ -923,6 +929,72 @@ describe("History resume delegation - parent metadata transitions", () => { expect(logSpy).toHaveBeenCalledWith(expect.stringContaining("[reopenParentFromDelegation] Aborting")) }) + it("reopenParentFromDelegation revalidates the locked parent snapshot before committing completion metadata", async () => { + const logSpy = vi.fn() + const saveTaskMessagesMock = vi.mocked(saveTaskMessages) + const saveApiMessagesMock = vi.mocked(saveApiMessages) + const createTaskWithHistoryItem = vi.fn() + const parentItem = { + id: "parent-locked-guard", + status: "delegated", + awaitingChildId: "child-locked-guard", + childIds: ["child-locked-guard"], + ts: 100, + task: "Parent locked guard", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + const childItem = { + id: "child-locked-guard", + status: "active", + ts: 101, + task: "Child locked guard", + tokensIn: 0, + tokensOut: 0, + totalCost: 0, + } + const atomicUpdatePair = vi.fn( + async ( + _firstId: string, + _secondId: string, + firstUpdater: (h: HistoryItem) => HistoryItem, + secondUpdater: (h: HistoryItem) => HistoryItem, + ) => { + firstUpdater(childItem as HistoryItem) + secondUpdater({ ...parentItem, status: "active", awaitingChildId: undefined } as HistoryItem) + return [] + }, + ) + + const provider = makeProviderStub({ + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, + getTaskWithId: vi.fn().mockResolvedValue({ historyItem: parentItem }), + emit: vi.fn(), + log: logSpy, + getCurrentTask: vi.fn(() => null), + removeClineFromStack: vi.fn(), + createTaskWithHistoryItem, + taskHistoryStore: { atomicUpdatePair, get: vi.fn() }, + } as any) + vi.mocked(readTaskMessages).mockResolvedValue([]) + vi.mocked(readApiMessages).mockResolvedValue([]) + + await expect( + (ClineProvider.prototype as any).reopenParentFromDelegation.call(provider, { + parentTaskId: "parent-locked-guard", + childTaskId: "child-locked-guard", + completionResultSummary: "should not commit", + }), + ).resolves.toBe(false) + + expect(atomicUpdatePair).toHaveBeenCalled() + expect(createTaskWithHistoryItem).not.toHaveBeenCalled() + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining("[reopenParentFromDelegation] Aborting")) + expect(saveTaskMessagesMock).toHaveBeenCalled() + expect(saveApiMessagesMock).toHaveBeenCalled() + }) + it("serializes delegation transitions and continues after a rejected predecessor", async () => { const provider = makeProviderStub({} as any) as any const calls: string[] = [] diff --git a/src/__tests__/provider-delegation.spec.ts b/src/__tests__/provider-delegation.spec.ts index ddc6a3c4f4..b0c4012e20 100644 --- a/src/__tests__/provider-delegation.spec.ts +++ b/src/__tests__/provider-delegation.spec.ts @@ -3,6 +3,13 @@ import { describe, it, expect, vi } from "vitest" import type { HistoryItem } from "@roo-code/types" import { RooCodeEventName } from "@roo-code/types" +vi.mock("../core/task-persistence/TaskHistoryLock", () => ({ + taskHistoryLock: { + withLock: vi.fn(async (_globalStoragePath: string, fn: () => Promise) => fn()), + reset: vi.fn(), + }, +})) + import { ClineProvider } from "../core/webview/ClineProvider" const parentHistoryItem: HistoryItem = { @@ -53,6 +60,7 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { const taskHistoryStore = makeStoreStub() const provider = { + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, emit: providerEmit, getCurrentTask: vi.fn(() => parentTask), removeClineFromStack, @@ -117,6 +125,7 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { }) const provider = { + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, emit: vi.fn(), getCurrentTask: vi.fn(() => parentTask), removeClineFromStack: vi.fn().mockResolvedValue(undefined), @@ -150,6 +159,7 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { }) const provider = { + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, emit: vi.fn(), getCurrentTask: vi.fn(() => parentTask), removeClineFromStack: vi.fn().mockResolvedValue(undefined), @@ -191,6 +201,7 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { }) const provider = { + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, emit: vi.fn(), getCurrentTask: vi.fn(() => parentTask), removeClineFromStack, @@ -236,6 +247,7 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { }) const provider = { + contextProxy: { globalStorageUri: { fsPath: "/tmp" } }, emit: vi.fn(), getCurrentTask, removeClineFromStack, diff --git a/src/core/task-persistence/TaskHistoryLock.ts b/src/core/task-persistence/TaskHistoryLock.ts index bbc0f7de7c..fcf1462302 100644 --- a/src/core/task-persistence/TaskHistoryLock.ts +++ b/src/core/task-persistence/TaskHistoryLock.ts @@ -76,9 +76,9 @@ export class TaskHistoryLock { update: 10000, realpath: false, retries: { - retries: 10, - factor: 2, - minTimeout: 50, + retries: 36, + factor: 1, + minTimeout: 1000, maxTimeout: 1000, }, onCompromised: (err) => { diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 219918b029..4b6c06594f 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -261,25 +261,27 @@ export class TaskHistoryStore { * Delete multiple tasks' history items in a batch. */ async deleteMany(taskIds: string[]): Promise { - return this.withLock(async () => { - for (const taskId of taskIds) { - this.cache.delete(taskId) + return this.withLock(() => this.deleteManyCore(taskIds)) + } - try { - const filePath = await this.getTaskFilePath(taskId) - await fs.unlink(filePath) - } catch { - // File may already be deleted - } + private async deleteManyCore(taskIds: string[]): Promise { + for (const taskId of taskIds) { + this.cache.delete(taskId) + + try { + const filePath = await this.getTaskFilePath(taskId) + await fs.unlink(filePath) + } catch { + // File may already be deleted } + } - this.scheduleIndexWrite() + this.scheduleIndexWrite() - // Call onWrite callback inside the lock for serialized write-through - if (this.onWrite) { - await this.onWrite(this.getAll()) - } - }) + // Call onWrite callback inside the lock for serialized write-through + if (this.onWrite) { + await this.onWrite(this.getAll()) + } } // ────────────────────────────── Reconciliation ────────────────────────────── @@ -292,50 +294,52 @@ export class TaskHistoryStore { */ async reconcile(): Promise { // Run through the write lock to prevent interleaving with upsert/delete - return this.withLock(async () => { - const tasksDir = await this.getTasksDir() + return this.withLock(() => this.reconcileCore()) + } - let dirEntries: string[] - try { - dirEntries = await fs.readdir(tasksDir) - } catch { - return // tasks dir doesn't exist yet - } + private async reconcileCore(): Promise { + const tasksDir = await this.getTasksDir() - // Filter out the index file and hidden files - const taskDirNames = dirEntries.filter((name) => !name.startsWith("_") && !name.startsWith(".")) - - const onDiskIds = new Set(taskDirNames) - const cacheIds = new Set(this.cache.keys()) - let changed = false - - // Tasks on disk but not in cache: read their history_item.json - for (const taskId of onDiskIds) { - if (!cacheIds.has(taskId)) { - try { - const item = await this.readTaskFile(taskId) - if (item) { - this.cache.set(taskId, item) - changed = true - } - } catch { - // Corrupted or missing file, skip + let dirEntries: string[] + try { + dirEntries = await fs.readdir(tasksDir) + } catch { + return // tasks dir doesn't exist yet + } + + // Filter out the index file and hidden files + const taskDirNames = dirEntries.filter((name) => !name.startsWith("_") && !name.startsWith(".")) + + const onDiskIds = new Set(taskDirNames) + const cacheIds = new Set(this.cache.keys()) + let changed = false + + // Tasks on disk but not in cache: read their history_item.json + for (const taskId of onDiskIds) { + if (!cacheIds.has(taskId)) { + try { + const item = await this.readTaskFile(taskId) + if (item) { + this.cache.set(taskId, item) + changed = true } + } catch { + // Corrupted or missing file, skip } } + } - // Tasks in cache but not on disk: remove from cache - for (const taskId of cacheIds) { - if (!onDiskIds.has(taskId)) { - this.cache.delete(taskId) - changed = true - } + // Tasks in cache but not on disk: remove from cache + for (const taskId of cacheIds) { + if (!onDiskIds.has(taskId)) { + this.cache.delete(taskId) + changed = true } + } - if (changed) { - this.scheduleIndexWrite() - } - }) + if (changed) { + this.scheduleIndexWrite() + } } /** @@ -359,69 +363,71 @@ export class TaskHistoryStore { * A parent awaiting an `active`, `interrupted`, or `delegated` child is left as-is — the child is resumable. */ private async reconcileDelegationState(): Promise { - return this.withLock(async () => { - let repairsInThisPass: number - do { - repairsInThisPass = 0 - // Rebuild the lookup map each pass so repairs from the previous pass - // are visible when evaluating chained delegations. - const byId = new Map(Array.from(this.cache.values()).map((i) => [i.id, i])) - - for (const [, item] of byId) { - if (item.status !== "delegated") { - continue - } + return this.withLock(() => this.reconcileDelegationStateCore()) + } - if (!item.awaitingChildId) { - await this.upsertCore( - { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, - ) - repairsInThisPass++ - continue - } + private async reconcileDelegationStateCore(): Promise { + let repairsInThisPass: number + do { + repairsInThisPass = 0 + // Rebuild the lookup map each pass so repairs from the previous pass + // are visible when evaluating chained delegations. + const byId = new Map(Array.from(this.cache.values()).map((i) => [i.id, i])) + + for (const [, item] of byId) { + if (item.status !== "delegated") { + continue + } - const child = byId.get(item.awaitingChildId) - - if (!child) { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, - ) - repairsInThisPass++ - } else if (child.status === "completed") { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - completedByChildId: child.id, - completionResultSummary: - child.completionResultSummary ?? "Task completed (recovered after interruption)", - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, - ) - repairsInThisPass++ - } - // child.status === "active", "interrupted", or "delegated" → leave as-is this pass + if (!item.awaitingChildId) { + await this.upsertCore( + { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, + ) + repairsInThisPass++ + continue } - } while (repairsInThisPass > 0) - }) + + const child = byId.get(item.awaitingChildId) + + if (!child) { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, + ) + repairsInThisPass++ + } else if (child.status === "completed") { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + completedByChildId: child.id, + completionResultSummary: + child.completionResultSummary ?? "Task completed (recovered after interruption)", + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, + ) + repairsInThisPass++ + } + // child.status === "active", "interrupted", or "delegated" → leave as-is this pass + } + } while (repairsInThisPass > 0) } // ────────────────────────────── Cache invalidation ────────────────────────────── @@ -458,6 +464,22 @@ export class TaskHistoryStore { * file if one doesn't already exist. This is idempotent and safe to re-run. */ async migrateFromGlobalState(taskHistoryEntries: HistoryItem[]): Promise { + return this.withLock(() => this.migrateFromGlobalStateCore(taskHistoryEntries)) + } + + async mutateLocked(fn: () => Promise): Promise { + return this.withLock(fn) + } + + async reconcileLocked(): Promise { + return this.reconcileCore() + } + + async deleteManyLocked(taskIds: string[]): Promise { + return this.deleteManyCore(taskIds) + } + + private async migrateFromGlobalStateCore(taskHistoryEntries: HistoryItem[]): Promise { if (!taskHistoryEntries || taskHistoryEntries.length === 0) { return } @@ -495,7 +517,7 @@ export class TaskHistoryStore { // Repair any delegation inconsistencies introduced by the migrated entries. // reconcileDelegationState() is idempotent so running it again is safe. - await this.reconcileDelegationState() + await this.reconcileDelegationStateCore() } // ────────────────────────────── Private: Index management ────────────────────────────── diff --git a/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts index 1b29d1b267..c7221eb700 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts @@ -1,153 +1,70 @@ +// pnpm --filter roo-cline test core/task-persistence/__tests__/TaskHistoryLock.spec.ts + import * as fs from "fs/promises" import * as os from "os" import * as path from "path" -import { fork, type ChildProcess } from "child_process" -import { TaskHistoryLock } from "../TaskHistoryLock" +const { lockMock } = vi.hoisted(() => ({ + lockMock: vi.fn(), +})) + +vi.mock("proper-lockfile", () => ({ + lock: lockMock, +})) vi.mock("../../../utils/storage", () => ({ getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath), })) -const waitForMessage = (child: ChildProcess, expected: string): Promise => - new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - child.off("message", onMessage) - reject(new Error(`Timed out waiting for child process message: ${expected}`)) - }, 5000) - - const onMessage = (message: unknown) => { - if (message === expected) { - clearTimeout(timeout) - child.off("message", onMessage) - resolve() - } - } - - child.on("message", onMessage) - }) +import { TaskHistoryLock } from "../TaskHistoryLock" +import { GlobalFileNames } from "../../../shared/globalFileNames" + +function cumulativeRetryWindowMs(retries: { + retries: number + factor: number + minTimeout: number + maxTimeout: number +}): number { + let total = 0 + for (let attempt = 0; attempt < retries.retries; attempt++) { + total += Math.min(retries.maxTimeout, retries.minTimeout * retries.factor ** attempt) + } + return total +} describe("TaskHistoryLock", () => { let tmpDir: string beforeEach(async () => { + vi.clearAllMocks() tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-lock-")) + lockMock.mockResolvedValue(vi.fn().mockResolvedValue(undefined)) }) afterEach(async () => { - await fs.rm(tmpDir, { recursive: true, force: true }) - }) - - it("serializes concurrent operations", async () => { - const lock = new TaskHistoryLock() - let activeCount = 0 - let maxActiveCount = 0 - const order: string[] = [] - - const run = (id: string) => - lock.withLock(tmpDir, async () => { - activeCount++ - maxActiveCount = Math.max(maxActiveCount, activeCount) - order.push(`start:${id}`) - await new Promise((resolve) => setTimeout(resolve, 5)) - order.push(`end:${id}`) - activeCount-- - return id - }) - - const results = await Promise.all([run("a"), run("b"), run("c")]) - - expect(results).toEqual(["a", "b", "c"]) - expect(maxActiveCount).toBe(1) - expect(order).toHaveLength(6) - for (const id of ["a", "b", "c"]) { - expect(order).toContain(`start:${id}`) - expect(order).toContain(`end:${id}`) - expect(order.indexOf(`start:${id}`)).toBeLessThan(order.indexOf(`end:${id}`)) - } + await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}) }) - it("continues processing after a previous operation rejects", async () => { - const lock = new TaskHistoryLock() - const order: string[] = [] + it("locks the shared tasks/_history.lock file and releases it after the callback", async () => { + const taskHistoryLock = new TaskHistoryLock() + const release = vi.fn().mockResolvedValue(undefined) + lockMock.mockResolvedValueOnce(release) - const failed = lock.withLock(tmpDir, async () => { - order.push("start:fail") - throw new Error("simulated failure") - }) + await expect(taskHistoryLock.withLock(tmpDir, async () => "done")).resolves.toBe("done") - const succeeded = lock.withLock(tmpDir, async () => { - order.push("start:success") - return "ok" - }) - - await expect(failed).rejects.toThrow("simulated failure") - await expect(succeeded).resolves.toBe("ok") - expect(order).toEqual(["start:fail", "start:success"]) - }) - - it("waits for an independent process holding the same lock file", async () => { - const lock = new TaskHistoryLock() - const lockFilePath = await lock.getLockFilePath(tmpDir) - const childScriptPath = path.join(tmpDir, "hold-history-lock.cjs") - await fs.writeFile( - childScriptPath, - ` -const lockfile = require("proper-lockfile") - -let release - -async function main() { - release = await lockfile.lock(process.argv[2], { - stale: 31000, - update: 10000, - realpath: false, - }) - process.send?.("locked") -} - -process.on("message", async (message) => { - if (message === "release") { - await release?.() - process.send?.("released") - process.exit(0) - } -}) - -main().catch((error) => { - process.send?.({ error: error instanceof Error ? error.message : String(error) }) - process.exit(1) -}) -`, - "utf8", + expect(lockMock).toHaveBeenCalledWith( + path.join(tmpDir, "tasks", GlobalFileNames.historyLock), + expect.any(Object), ) + expect(release).toHaveBeenCalledTimes(1) + }) - const child = fork(childScriptPath, [lockFilePath], { stdio: ["ignore", "ignore", "ignore", "ipc"] }) - try { - await waitForMessage(child, "locked") - - let enteredCriticalSection = false - const blocked = lock.withLock(tmpDir, async () => { - enteredCriticalSection = true - return "acquired" - }) - - await new Promise((resolve) => setTimeout(resolve, 100)) - expect(enteredCriticalSection).toBe(false) + it("keeps retrying long enough for proper-lockfile stale-lock recovery", async () => { + const taskHistoryLock = new TaskHistoryLock() - child.send("release") - await waitForMessage(child, "released") - await expect(blocked).resolves.toBe("acquired") - expect(enteredCriticalSection).toBe(true) - } finally { - if (!child.killed) { - child.kill() - } - } - }) + await taskHistoryLock.withLock(tmpDir, async () => undefined) - it("reset is a no-op for file-based locking", () => { - const lock = new TaskHistoryLock() - expect(() => lock.reset()).not.toThrow() + const options = lockMock.mock.calls[0][1] + expect(cumulativeRetryWindowMs(options.retries)).toBeGreaterThan(options.stale) }) }) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts index 75e0a16253..9c52367572 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts @@ -233,6 +233,37 @@ describe("TaskHistoryStore", () => { }) }) + describe("locked mutation helpers", () => { + it("allows reconcileLocked and deleteManyLocked inside one mutateLocked callback without re-entering the lock", async () => { + await store.initialize() + + await store.upsert(makeHistoryItem({ id: "locked-delete" })) + + const tasksDir = path.join(tmpDir, "tasks") + const orphanDir = path.join(tasksDir, "locked-orphan") + await fs.mkdir(orphanDir, { recursive: true }) + await fs.writeFile( + path.join(orphanDir, GlobalFileNames.historyItem), + JSON.stringify(makeHistoryItem({ id: "locked-orphan" })), + ) + + const result = await Promise.race([ + store + .mutateLocked(async () => { + await store.reconcileLocked() + expect(store.get("locked-orphan")).toBeDefined() + await store.deleteManyLocked(["locked-delete", "locked-orphan"]) + }) + .then(() => "completed"), + new Promise((resolve) => setTimeout(() => resolve("deadlocked"), 100)), + ]) + + expect(result).toBe("completed") + expect(store.get("locked-delete")).toBeUndefined() + expect(store.get("locked-orphan")).toBeUndefined() + }) + }) + describe("reconcile()", () => { it("detects tasks on disk missing from index", async () => { await store.initialize() diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index edb8f4b328..87a2ef46fe 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -377,23 +377,23 @@ export class ClineProvider try { await this.taskHistoryStore.initialize() - // Migration: backfill per-task files from globalState on first run + // Migration: backfill per-task files from globalState on first run. const migrationKey = "taskHistoryMigratedToFiles" - const alreadyMigrated = this.context.globalState.get(migrationKey) + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { + const alreadyMigrated = this.context.globalState.get(migrationKey) + if (alreadyMigrated) { + return + } - if (!alreadyMigrated) { const legacyHistory = this.context.globalState.get("taskHistory") ?? [] - if (legacyHistory.length > 0) { this.log(`[initializeTaskHistoryStore] Migrating ${legacyHistory.length} entries from globalState`) - await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => - this.taskHistoryStore.migrateFromGlobalState(legacyHistory), - ) + await this.taskHistoryStore.migrateFromGlobalState(legacyHistory) } await this.context.globalState.update(migrationKey, true) this.log("[initializeTaskHistoryStore] Migration complete") - } + }) this.taskHistoryStoreInitialized = true } catch (error) { @@ -2017,31 +2017,38 @@ export class ClineProvider // If the task has subtasks (childIds), they will also be deleted recursively async deleteTaskWithId(id: string, cascadeSubtasks: boolean = true) { try { - // get the task directory full path and history item - const { taskDirPath, historyItem } = await this.getTaskWithId(id) - - // Collect all task IDs to delete (parent + all subtasks) - const allIdsToDelete: string[] = [id] - - if (cascadeSubtasks) { - // Recursively collect all child IDs - const collectChildIds = async (taskId: string): Promise => { - try { - const { historyItem: item } = await this.getTaskWithId(taskId) - if (item.childIds && item.childIds.length > 0) { - for (const childId of item.childIds) { + const allIdsToDelete: string[] = [] + + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { + await this.taskHistoryStore.mutateLocked(async () => { + await this.taskHistoryStore.reconcileLocked() + const rootItem = this.taskHistoryStore.get(id) + if (!rootItem) { + throw new Error("Task not found") + } + + allIdsToDelete.push(id) + if (cascadeSubtasks) { + const visited = new Set(allIdsToDelete) + const collectChildIds = (taskId: string): void => { + const item = this.taskHistoryStore.get(taskId) + for (const childId of item?.childIds ?? []) { + if (visited.has(childId)) { + continue + } + visited.add(childId) allIdsToDelete.push(childId) - await collectChildIds(childId) + collectChildIds(childId) } } - } catch (error) { - // Child task may already be deleted or not found, continue - console.log(`[deleteTaskWithId] child task ${taskId} not found, skipping`) + collectChildIds(id) } - } - await collectChildIds(id) - } + await this.taskHistoryStore.deleteManyLocked(allIdsToDelete) + }) + this.recentTasksCache = undefined + await this.postStateToWebview() + }) // Remove from stack if any of the tasks to delete are in the current task stack for (const taskId of allIdsToDelete) { @@ -2052,14 +2059,6 @@ export class ClineProvider } } - await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, async () => { - // Delete all tasks from state in one batch - await this.taskHistoryStore.deleteMany(allIdsToDelete) - this.recentTasksCache = undefined - - await this.postStateToWebview() - }) - // Delete associated shadow repositories or branches and task directories const globalStorageDir = this.contextProxy.globalStorageUri.fsPath const workspaceDir = this.cwd @@ -3839,32 +3838,49 @@ export class ClineProvider // any concurrent write that landed between step 1 and the lock acquisition // is preserved rather than silently overwritten. let updatedHistory!: typeof historyItem - await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => - this.taskHistoryStore.atomicUpdatePair( - childTaskId, - parentTaskId, - (child) => { - assertValidTransition(child.status, "completed") - return { ...child, status: "completed" as const, completionResultSummary } - }, - (parent) => { - if (parent.status !== "active") { - assertValidTransition(parent.status, "active") - } - const childIds = Array.from(new Set([...(parent.childIds ?? []), childTaskId])) - updatedHistory = { - ...parent, - status: "active" as const, - completedByChildId: childTaskId, - completionResultSummary, - awaitingChildId: undefined, - delegatedToId: undefined, - childIds, - } - return updatedHistory - }, - ), - ) + try { + await taskHistoryLock.withLock(this.contextProxy.globalStorageUri.fsPath, () => + this.taskHistoryStore.atomicUpdatePair( + childTaskId, + parentTaskId, + (child) => { + assertValidTransition(child.status, "completed") + return { ...child, status: "completed" as const, completionResultSummary } + }, + (parent) => { + const parentCanCompleteDelegation = + (parent.status === "delegated" || parent.status === "active") && + parent.awaitingChildId === childTaskId + if (!parentCanCompleteDelegation) { + throw new Error( + `[reopenParentFromDelegation] Aborting: parent ${parentTaskId} is no longer delegated to child ${childTaskId} ` + + `(status=${parent.status}, awaitingChildId=${parent.awaitingChildId})`, + ) + } + if (parent.status !== "active") { + assertValidTransition(parent.status, "active") + } + const childIds = Array.from(new Set([...(parent.childIds ?? []), childTaskId])) + updatedHistory = { + ...parent, + status: "active" as const, + completedByChildId: childTaskId, + completionResultSummary, + awaitingChildId: undefined, + delegatedToId: undefined, + childIds, + } + return updatedHistory + }, + ), + ) + } catch (error) { + if (error instanceof Error && error.message.includes("[reopenParentFromDelegation] Aborting")) { + this.log(error.message) + return false + } + throw error + } this.recentTasksCache = undefined // Notify the webview of both updated items so its in-memory history stays current. diff --git a/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts b/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts index f57c0bd6ab..8ec2966936 100644 --- a/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.taskHistory.spec.ts @@ -802,6 +802,8 @@ describe("ClineProvider Task History Synchronization", () => { new ContextProxy(mockContext), ) await provider2.taskHistoryStore.initialized + await new Promise((resolve) => setTimeout(resolve, 10)) + vi.mocked(taskHistoryLock.withLock).mockClear() const provider1Upsert = vi .spyOn(provider.taskHistoryStore, "upsert") @@ -845,6 +847,8 @@ describe("ClineProvider Task History Synchronization", () => { await nextProvider.taskHistoryStore.initialized providers.push(nextProvider) } + await new Promise((resolve) => setTimeout(resolve, 10)) + vi.mocked(taskHistoryLock.withLock).mockClear() try { providers.forEach((currentProvider) => { @@ -874,15 +878,14 @@ describe("ClineProvider Task History Synchronization", () => { }) it("routes normal deleteTaskWithId deleteMany mutations through the shared lock", async () => { - const deleteManySpy = vi.spyOn(provider.taskHistoryStore, "deleteMany").mockResolvedValue(undefined) + const deleteManySpy = vi.spyOn(provider.taskHistoryStore, "deleteManyLocked").mockResolvedValue(undefined) const postStateSpy = vi.spyOn(provider, "postStateToWebview").mockResolvedValue(undefined) - vi.spyOn(provider, "getTaskWithId").mockResolvedValue({ - historyItem: createHistoryItem({ id: "delete-normal", task: "Delete normal" }), - taskDirPath: "/test/task/path", - apiConversationHistoryFilePath: "/test/task/path/api_conversation_history.json", - uiMessagesFilePath: "/test/task/path/ui_messages.json", - apiConversationHistory: [], - }) + vi.spyOn(provider.taskHistoryStore, "reconcile").mockResolvedValue(undefined) + vi.spyOn(provider.taskHistoryStore, "get").mockImplementation((taskId: string) => + taskId === "delete-normal" + ? createHistoryItem({ id: "delete-normal", task: "Delete normal" }) + : undefined, + ) await provider.deleteTaskWithId("delete-normal", false) @@ -894,6 +897,67 @@ describe("ClineProvider Task History Synchronization", () => { expect(postStateSpy).toHaveBeenCalled() }) + it("collects cascade delete descendants from the locked persisted snapshot", async () => { + let insideSharedLock = false + vi.mocked(taskHistoryLock.withLock).mockImplementationOnce(async (_globalStoragePath, fn) => { + insideSharedLock = true + return fn() + }) + + const deleteManySpy = vi.spyOn(provider.taskHistoryStore, "deleteManyLocked").mockResolvedValue(undefined) + vi.spyOn(provider, "postStateToWebview").mockResolvedValue(undefined) + vi.spyOn(provider.taskHistoryStore, "reconcile").mockResolvedValue(undefined) + vi.spyOn(provider.taskHistoryStore, "get").mockImplementation((taskId: string) => { + if (!insideSharedLock) { + return undefined + } + if (taskId === "delete-parent") { + return createHistoryItem({ id: "delete-parent", task: "Delete parent", childIds: ["late-child"] }) + } + if (taskId === "late-child") { + return createHistoryItem({ id: "late-child", task: "Late child", childIds: [] }) + } + return undefined + }) + + await provider.deleteTaskWithId("delete-parent", true) + + expect(deleteManySpy).toHaveBeenCalledWith(["delete-parent", "late-child"]) + }) + + it("rechecks migration state inside the shared lock before migrating legacy globalState history", async () => { + let locked = false + const legacyHistory = [createHistoryItem({ id: "legacy-task", task: "Legacy task" })] + const get = vi.fn((key: string) => { + if (key === "taskHistoryMigratedToFiles") { + return locked + } + if (key === "taskHistory") { + return locked ? [] : legacyHistory + } + return undefined + }) + const update = vi.fn().mockResolvedValue(undefined) + const migrateFromGlobalState = vi.fn().mockResolvedValue(undefined) + vi.mocked(taskHistoryLock.withLock).mockImplementationOnce(async (_globalStoragePath, fn) => { + locked = true + return fn() + }) + + await (ClineProvider.prototype as any).initializeTaskHistoryStore.call({ + taskHistoryStore: { + initialize: vi.fn().mockResolvedValue(undefined), + migrateFromGlobalState, + }, + context: { globalState: { get, update } }, + contextProxy: { globalStorageUri: { fsPath: mockContext.globalStorageUri.fsPath } }, + log: vi.fn(), + }) + + expect(migrateFromGlobalState).not.toHaveBeenCalled() + expect(update).not.toHaveBeenCalledWith("taskHistoryMigratedToFiles", true) + }) + it("routes fallback deleteTaskFromState mutations through the shared lock", async () => { const deleteSpy = vi.spyOn(provider.taskHistoryStore, "delete").mockResolvedValue(undefined) vi.spyOn(provider, "postStateToWebview").mockResolvedValue(undefined) From 0d004f7d3eb29cb7244e5f7e6f7c6f55263c204c Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 18 Jul 2026 03:21:39 +0800 Subject: [PATCH 3/4] fix(task-history): serialize store mutations with shared lock --- src/core/task-persistence/TaskHistoryStore.ts | 28 ++++++---- .../__tests__/TaskHistoryStore.spec.ts | 52 ++++++++++++++----- 2 files changed, 58 insertions(+), 22 deletions(-) diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 4b6c06594f..da2d6fad0e 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -7,6 +7,7 @@ import type { HistoryItem } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" import { safeWriteJson } from "../../utils/safeWriteJson" import { getStorageBasePath } from "../../utils/storage" +import { taskHistoryLock } from "./TaskHistoryLock" /** Valid status values for a task's HistoryItem. */ export type HistoryItemStatus = NonNullable @@ -184,7 +185,7 @@ export class TaskHistoryStore { * updates the in-memory Map, and schedules a debounced index write. */ async upsert(item: HistoryItem): Promise { - return this.withLock(() => this.upsertCore(item)) + return this.withSharedStorageLock(() => this.upsertCore(item)) } /** @@ -237,7 +238,7 @@ export class TaskHistoryStore { * Delete a single task's history item. */ async delete(taskId: string): Promise { - return this.withLock(async () => { + return this.withSharedStorageLock(async () => { this.cache.delete(taskId) // Remove per-task file (best-effort) @@ -261,7 +262,7 @@ export class TaskHistoryStore { * Delete multiple tasks' history items in a batch. */ async deleteMany(taskIds: string[]): Promise { - return this.withLock(() => this.deleteManyCore(taskIds)) + return this.withSharedStorageLock(() => this.deleteManyCore(taskIds)) } private async deleteManyCore(taskIds: string[]): Promise { @@ -293,8 +294,8 @@ export class TaskHistoryStore { * - Tasks in cache but missing from disk: remove */ async reconcile(): Promise { - // Run through the write lock to prevent interleaving with upsert/delete - return this.withLock(() => this.reconcileCore()) + // Run through the shared storage lock and write lock to prevent interleaving with upsert/delete. + return this.withSharedStorageLock(() => this.reconcileCore()) } private async reconcileCore(): Promise { @@ -363,7 +364,7 @@ export class TaskHistoryStore { * A parent awaiting an `active`, `interrupted`, or `delegated` child is left as-is — the child is resumable. */ private async reconcileDelegationState(): Promise { - return this.withLock(() => this.reconcileDelegationStateCore()) + return this.withSharedStorageLock(() => this.reconcileDelegationStateCore()) } private async reconcileDelegationStateCore(): Promise { @@ -464,11 +465,11 @@ export class TaskHistoryStore { * file if one doesn't already exist. This is idempotent and safe to re-run. */ async migrateFromGlobalState(taskHistoryEntries: HistoryItem[]): Promise { - return this.withLock(() => this.migrateFromGlobalStateCore(taskHistoryEntries)) + return this.withSharedStorageLock(() => this.migrateFromGlobalStateCore(taskHistoryEntries)) } async mutateLocked(fn: () => Promise): Promise { - return this.withLock(fn) + return this.withSharedStorageLock(fn) } async reconcileLocked(): Promise { @@ -703,7 +704,7 @@ export class TaskHistoryStore { * @throws If the task ID is not present in the cache. */ public atomicReadAndUpdate(taskId: string, updater: (current: HistoryItem) => HistoryItem): Promise { - return this.withLock(async () => { + return this.withSharedStorageLock(async () => { const current = this.cache.get(taskId) if (!current) { throw new Error(`[TaskHistoryStore] atomicReadAndUpdate: task ${taskId} not found in cache`) @@ -734,7 +735,7 @@ export class TaskHistoryStore { firstUpdater: (current: HistoryItem) => HistoryItem, secondUpdater: (current: HistoryItem) => HistoryItem, ): Promise { - return this.withLock(async () => { + return this.withSharedStorageLock(async () => { const first = this.cache.get(firstId) if (!first) throw new Error(`[TaskHistoryStore] atomicUpdatePair: ${firstId} not found`) const second = this.cache.get(secondId) @@ -791,6 +792,13 @@ export class TaskHistoryStore { // ────────────────────────────── Private: Write lock ────────────────────────────── + /** + * Acquires the shared task-history storage lock before entering the instance-local queue. + */ + private withSharedStorageLock(fn: () => Promise): Promise { + return taskHistoryLock.withLock(this.globalStoragePath, () => this.withLock(fn)) + } + /** * Serializes all read-modify-write operations within a single extension * host process to prevent concurrent interleaving. diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts index 9c52367572..646434a388 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts @@ -6,6 +6,16 @@ import * as os from "os" import type { HistoryItem } from "@roo-code/types" +const { taskHistoryLockWithLockMock } = vi.hoisted(() => ({ + taskHistoryLockWithLockMock: vi.fn(async (_globalStoragePath: string, fn: () => Promise) => fn()), +})) + +vi.mock("../TaskHistoryLock", () => ({ + taskHistoryLock: { + withLock: taskHistoryLockWithLockMock, + }, +})) + import { TaskHistoryStore, assertValidTransition } from "../TaskHistoryStore" import { GlobalFileNames } from "../../../shared/globalFileNames" @@ -42,6 +52,10 @@ describe("TaskHistoryStore", () => { let store: TaskHistoryStore beforeEach(async () => { + taskHistoryLockWithLockMock.mockClear() + taskHistoryLockWithLockMock.mockImplementation(async (_globalStoragePath: string, fn: () => Promise) => + fn(), + ) tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-test-")) store = new TaskHistoryStore(tmpDir) }) @@ -231,6 +245,17 @@ describe("TaskHistoryStore", () => { expect(store.getAll()).toHaveLength(1) expect(store.get("batch-2")).toBeDefined() }) + + it("acquires the shared storage lock before entering the instance queue", async () => { + await store.initialize() + await store.upsert(makeHistoryItem({ id: "shared-lock-delete" })) + taskHistoryLockWithLockMock.mockClear() + + await store.deleteMany(["shared-lock-delete"]) + + expect(taskHistoryLockWithLockMock).toHaveBeenCalledTimes(1) + expect(taskHistoryLockWithLockMock).toHaveBeenCalledWith(tmpDir, expect.any(Function)) + }) }) describe("locked mutation helpers", () => { @@ -247,21 +272,24 @@ describe("TaskHistoryStore", () => { JSON.stringify(makeHistoryItem({ id: "locked-orphan" })), ) - const result = await Promise.race([ - store - .mutateLocked(async () => { - await store.reconcileLocked() - expect(store.get("locked-orphan")).toBeDefined() - await store.deleteManyLocked(["locked-delete", "locked-orphan"]) - }) - .then(() => "completed"), - new Promise((resolve) => setTimeout(() => resolve("deadlocked"), 100)), - ]) - - expect(result).toBe("completed") + await store.mutateLocked(async () => { + await store.reconcileLocked() + expect(store.get("locked-orphan")).toBeDefined() + await store.deleteManyLocked(["locked-delete", "locked-orphan"]) + }) + expect(store.get("locked-delete")).toBeUndefined() expect(store.get("locked-orphan")).toBeUndefined() }) + it("acquires the shared storage lock before entering the instance queue", async () => { + await store.initialize() + taskHistoryLockWithLockMock.mockClear() + + await store.mutateLocked(async () => undefined) + + expect(taskHistoryLockWithLockMock).toHaveBeenCalledTimes(1) + expect(taskHistoryLockWithLockMock).toHaveBeenCalledWith(tmpDir, expect.any(Function)) + }) }) describe("reconcile()", () => { From 381c1837cf6e807ae5e94b09728f714faeeb3840 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 18 Jul 2026 03:35:48 +0800 Subject: [PATCH 4/4] fix(task-history): avoid fs open in lock path setup --- src/core/task-persistence/TaskHistoryLock.ts | 11 +---------- .../__tests__/ClineProvider.sticky-profile.spec.ts | 2 ++ 2 files changed, 3 insertions(+), 10 deletions(-) diff --git a/src/core/task-persistence/TaskHistoryLock.ts b/src/core/task-persistence/TaskHistoryLock.ts index fcf1462302..6d330969e3 100644 --- a/src/core/task-persistence/TaskHistoryLock.ts +++ b/src/core/task-persistence/TaskHistoryLock.ts @@ -55,16 +55,7 @@ export class TaskHistoryLock { const basePath = await getStorageBasePath(globalStoragePath) const tasksDir = path.join(basePath, "tasks") await fs.mkdir(tasksDir, { recursive: true }) - const lockFilePath = path.join(tasksDir, GlobalFileNames.historyLock) - - try { - await fs.open(lockFilePath, "a").then((handle) => handle.close()) - } catch (error) { - console.error(`[TaskHistoryLock] Failed to create lock file at ${lockFilePath}:`, error) - throw error - } - - return lockFilePath + return path.join(tasksDir, GlobalFileNames.historyLock) } private async runWithFileLock(lockFilePath: string, fn: () => Promise): Promise { diff --git a/src/core/webview/__tests__/ClineProvider.sticky-profile.spec.ts b/src/core/webview/__tests__/ClineProvider.sticky-profile.spec.ts index c982cf53c0..9e3587ceb2 100644 --- a/src/core/webview/__tests__/ClineProvider.sticky-profile.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.sticky-profile.spec.ts @@ -4,6 +4,7 @@ import * as vscode from "vscode" import { TelemetryService } from "@roo-code/telemetry" import { ClineProvider } from "../ClineProvider" import { ContextProxy } from "../../config/ContextProxy" +import { taskHistoryLock } from "../../task-persistence/TaskHistoryLock" import type { HistoryItem } from "@roo-code/types" vi.mock("vscode", () => ({ @@ -215,6 +216,7 @@ describe("ClineProvider - Sticky Provider Profile", () => { beforeEach(async () => { vi.clearAllMocks() + vi.spyOn(taskHistoryLock, "withLock").mockImplementation(async (_globalStoragePath, fn) => fn()) taskIdCounter = 0 originalRooCliRuntimeEnv = process.env.ROO_CLI_RUNTIME delete process.env.ROO_CLI_RUNTIME