fix(core): serialize task history updates across parallel tabs

Author: easonliang28

src/core/task-persistence/TaskHistoryLock.ts

@@ -0,0 +1,40 @@
+/**
+ * Cross-instance lock for task history mutations.
+ *
+ * When multiple `ClineProvider` instances exist (e.g., parallel tabs or windows),
+ * each has its own `TaskHistoryStore` with an independent in-process write lock.
+ * This singleton lock serializes all `updateTaskHistory()` calls across instances
+ * to prevent lost entries due to concurrent writes.
+ *
+ * The lock is a simple Promise chain — each operation waits for the previous one
+ * to complete before starting. This guarantees that only one mutation sequence
+ * runs at a time, eliminating race conditions between parallel tabs.
+ */
+
+export class TaskHistoryLock {
+	private queue: Promise<unknown> = Promise.resolve()
+
+	/**
+	 * Acquires the lock and executes `fn` sequentially.
+	 * Subsequent calls will wait for this one to finish.
+	 */
+	async withLock<T>(fn: () => Promise<T>): Promise<T> {
+		const result = this.queue.then(fn, fn)
+		this.queue = result.then(
+			() => undefined,
+			() => undefined,
+		)
+		return result
+	}
+
+	/**
+	 * Resets the lock queue to a clean state.
+	 * Useful for testing or when the provider is disposed.
+	 */
+	reset(): void {
+		this.queue = Promise.resolve()
+	}
+}
+
+// Singleton instance shared across all ClineProvider instances
+export const taskHistoryLock = new TaskHistoryLock()

src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts

@@ -0,0 +1,76 @@
+import { TaskHistoryLock } from "../TaskHistoryLock"
+
+describe("TaskHistoryLock", () => {
+	it("serializes concurrent operations", async () => {
+		const lock = new TaskHistoryLock()
+		let activeCount = 0
+		let maxActiveCount = 0
+		const order: string[] = []
+
+		const run = (id: string) =>
+			lock.withLock(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).toEqual(["start:a", "end:a", "start:b", "end:b", "start:c", "end:c"])
+	})
+
+	it("continues processing after a previous operation rejects", async () => {
+		const lock = new TaskHistoryLock()
+		const order: string[] = []
+
+		const failed = lock.withLock(async () => {
+			order.push("start:fail")
+			throw new Error("simulated failure")
+		})
+
+		const succeeded = lock.withLock(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("reset clears the queue for subsequent operations", async () => {
+		const lock = new TaskHistoryLock()
+		let releaseFirstOperation!: () => void
+		const order: string[] = []
+
+		const blocked = lock.withLock(
+			() =>
+				new Promise<string>((resolve) => {
+					order.push("start:blocked")
+					releaseFirstOperation = () => {
+						order.push("end:blocked")
+						resolve("blocked")
+					}
+				}),
+		)
+
+		lock.reset()
+
+		const afterReset = await lock.withLock(async () => {
+			order.push("after-reset")
+			return "after-reset"
+		})
+
+		expect(afterReset).toBe("after-reset")
+		expect(order).toEqual(["start:blocked", "after-reset"])
+
+		releaseFirstOperation()
+		await expect(blocked).resolves.toBe("blocked")
+	})
+})

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"
@@ -2092,10 +2093,12 @@ export class ClineProvider
 	}
 
 	async deleteTaskFromState(id: string) {
-		await this.taskHistoryStore.delete(id)
-		this.recentTasksCache = undefined
+		await taskHistoryLock.withLock(async () => {
+			await this.taskHistoryStore.delete(id)
+			this.recentTasksCache = undefined
 
-		await this.postStateToWebview()
+			await this.postStateToWebview()
+		})
 	}
 
 	async refreshWorkspace() {
@@ -2736,17 +2739,20 @@ export class ClineProvider
 	async updateTaskHistory(item: HistoryItem, options: { broadcast?: boolean } = {}): Promise<HistoryItem[]> {
 		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(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
+		})
 	}
 
 	/**

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
@@ -240,6 +241,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 +262,7 @@ describe("ClineProvider Task History Synchronization", () => {
 
 	beforeEach(async () => {
 		vi.clearAllMocks()
+		taskHistoryLock.reset()
 
 		if (!TelemetryService.hasInstance()) {
 			TelemetryService.createInstance([])
@@ -779,5 +782,147 @@ 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("serializes concurrent updateTaskHistory from two parallel instances", async () => {
+			const provider2 = new ClineProvider(
+				mockContext,
+				mockOutputChannel,
+				"sidebar",
+				new ContextProxy(mockContext),
+			)
+			await provider2.taskHistoryStore.initialized
+
+			let inCriticalSection = 0
+			let maxConcurrentMutations = 0
+			const entered: string[] = []
+
+			const makeSerializedUpsert = (instanceName: string) =>
+				vi.fn(async (item: HistoryItem) => {
+					inCriticalSection++
+					maxConcurrentMutations = Math.max(maxConcurrentMutations, inCriticalSection)
+					entered.push(`${instanceName}:${item.id}`)
+					await new Promise((resolve) => setTimeout(resolve, 5))
+					inCriticalSection--
+					return [item]
+				})
+
+			const provider1Upsert = makeSerializedUpsert("provider1")
+			const provider2Upsert = makeSerializedUpsert("provider2")
+			vi.spyOn(provider.taskHistoryStore, "upsert").mockImplementation(provider1Upsert)
+			vi.spyOn(provider2.taskHistoryStore, "upsert").mockImplementation(provider2Upsert)
+
+			try {
+				const provider1Item = createHistoryItem({ id: "parallel-provider-1", task: "Provider 1" })
+				const provider2Item = createHistoryItem({ id: "parallel-provider-2", task: "Provider 2" })
+
+				await Promise.all([
+					provider.updateTaskHistory(provider1Item, { broadcast: false }),
+					provider2.updateTaskHistory(provider2Item, { broadcast: false }),
+				])
+
+				expect(provider1Upsert).toHaveBeenCalledTimes(1)
+				expect(provider2Upsert).toHaveBeenCalledTimes(1)
+				expect(entered).toHaveLength(2)
+				expect(maxConcurrentMutations).toBe(1)
+			} finally {
+				await provider2.dispose()
+			}
+		})
+
+		it("serializes 5+ concurrent updateTaskHistory calls from different tabs", 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)
+			}
+
+			let inCriticalSection = 0
+			let maxConcurrentMutations = 0
+			const entered: string[] = []
+
+			try {
+				providers.forEach((currentProvider, providerIndex) => {
+					vi.spyOn(currentProvider.taskHistoryStore, "upsert").mockImplementation(
+						async (item: HistoryItem) => {
+							inCriticalSection++
+							maxConcurrentMutations = Math.max(maxConcurrentMutations, inCriticalSection)
+							entered.push(`provider-${providerIndex}:${item.id}`)
+							await new Promise((resolve) => setTimeout(resolve, 5))
+							inCriticalSection--
+							return [item]
+						},
+					)
+				})
+
+				await Promise.all(
+					providers.map((currentProvider, index) =>
+						currentProvider.updateTaskHistory(
+							createHistoryItem({ id: `parallel-tab-${index}`, task: `Parallel Tab ${index}` }),
+							{ broadcast: false },
+						),
+					),
+				)
+
+				expect(entered).toHaveLength(5)
+				expect(maxConcurrentMutations).toBe(1)
+			} finally {
+				for (const currentProvider of providers.slice(1)) {
+					await currentProvider.dispose()
+				}
+			}
+		})
+
+		it("serializes concurrent updateTaskHistory and deleteTaskFromState across tabs", async () => {
+			const provider2 = new ClineProvider(
+				mockContext,
+				mockOutputChannel,
+				"sidebar",
+				new ContextProxy(mockContext),
+			)
+			await provider2.taskHistoryStore.initialized
+
+			let inCriticalSection = 0
+			let maxConcurrentMutations = 0
+			const entered: string[] = []
+
+			vi.spyOn(provider.taskHistoryStore, "upsert").mockImplementation(async (item: HistoryItem) => {
+				inCriticalSection++
+				maxConcurrentMutations = Math.max(maxConcurrentMutations, inCriticalSection)
+				entered.push(`update:${item.id}`)
+				await new Promise((resolve) => setTimeout(resolve, 5))
+				inCriticalSection--
+				return [item]
+			})
+			vi.spyOn(provider2.taskHistoryStore, "delete").mockImplementation(async (id: string) => {
+				inCriticalSection++
+				maxConcurrentMutations = Math.max(maxConcurrentMutations, inCriticalSection)
+				entered.push(`delete:${id}`)
+				await new Promise((resolve) => setTimeout(resolve, 5))
+				inCriticalSection--
+			})
+			vi.spyOn(provider2, "postStateToWebview").mockResolvedValue(undefined)
+
+			try {
+				await Promise.all([
+					provider.updateTaskHistory(createHistoryItem({ id: "cross-update", task: "Cross update" }), {
+						broadcast: false,
+					}),
+					provider2.deleteTaskFromState("cross-delete"),
+				])
+
+				expect(entered).toEqual(expect.arrayContaining(["update:cross-update", "delete:cross-delete"]))
+				expect(entered).toHaveLength(2)
+				expect(maxConcurrentMutations).toBe(1)
+			} finally {
+				await provider2.dispose()
+			}
+		})
 	})
 })