fix(task-history): prevent concurrent index clobbering

src/core/task-persistence/TaskHistoryLock.ts

@@ -0,0 +1,104 @@
+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 index rebuilds.
+ *
+ * Multiple extension hosts (VS Code windows, JetBrains multi-agent sessions)
+ * may share the same task-history storage. Each process has its own
+ * `TaskHistoryStore` and in-memory cache, so an in-process mutex cannot
+ * protect `_index.json` read-merge-write. This lock serializes index rebuilds
+ * via an exclusive advisory lock on `tasks/_history.lock`.
+ */
+export class TaskHistoryLock {
+	private queue: Promise<unknown> = 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 processes targeting the same history
+	 * store contend on the same file.
+	 */
+	async withLock<T>(globalStoragePath: string, fn: () => Promise<T>): Promise<T> {
+		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<string> {
+		const basePath = await getStorageBasePath(globalStoragePath)
+		const tasksDir = path.join(basePath, "tasks")
+		await fs.mkdir(tasksDir, { recursive: true })
+		return path.join(tasksDir, GlobalFileNames.historyLock)
+	}
+
+	private async runWithFileLock<T>(lockFilePath: string, fn: () => Promise<T>): Promise<T> {
+		let releaseLock: (() => Promise<void>) | undefined
+
+		try {
+			// Ensure the lock target exists; proper-lockfile needs a path it can stat
+			// when realpath is disabled for not-yet-created targets.
+			try {
+				await fs.writeFile(lockFilePath, "", { flag: "wx" })
+			} catch (error: unknown) {
+				const code = (error as NodeJS.ErrnoException)?.code
+				if (code !== "EEXIST") {
+					throw error
+				}
+			}
+
+			releaseLock = await lockfile.lock(lockFilePath, {
+				stale: 31000,
+				update: 10000,
+				realpath: false,
+				retries: {
+					// Keep retrying longer than the stale window so a crashed holder
+					// can be recovered instead of failing index flushes permanently.
+					retries: 36,
+					factor: 1,
+					minTimeout: 1000,
+					maxTimeout: 1000,
+				},
+				onCompromised: (err) => {
+					console.error(`[TaskHistoryLock] Lock at ${lockFilePath} was compromised:`, err)
+					throw err
+				},
+			})
+
+			return await fn()
+		} finally {
+			if (releaseLock) {
+				await releaseLock()
+			}
+		}
+	}
+}
+
+/** Singleton shared by all TaskHistoryStore instances in this process. */
+export const taskHistoryLock = new TaskHistoryLock()

src/core/task-persistence/TaskHistoryStore.ts

@@ -9,6 +9,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<HistoryItem["status"]>
@@ -78,9 +79,12 @@ interface DelegationRepairIntent {
  * A single index file (`globalStorage/tasks/_index.json`) is maintained
  * as a cache for fast list reads at startup.
  *
- * Cross-process safety comes from `safeWriteJson`'s `proper-lockfile`
- * on per-task file writes. Within a single extension host process,
- * an in-process write lock serializes mutations.
+ * Cross-process safety for per-task files comes from `safeWriteJson`'s
+ * `proper-lockfile`. The shared `_index.json` is treated as a rebuildable
+ * cache: writers rebuild it from on-disk `history_item.json` files under a
+ * cross-process `tasks/_history.lock`, so a stale in-memory snapshot cannot
+ * clobber entries published by another extension host. Within a single
+ * process, an in-process write lock serializes mutations.
  */
 /**
  * Options for TaskHistoryStore constructor.
@@ -880,17 +884,57 @@ export class TaskHistoryStore {
 	}
 
 	/**
-	 * Write the full index to disk.
+	 * Rebuild `_index.json` from authoritative per-task files under a
+	 * cross-process lock.
+	 *
+	 * Must not dump `this.cache` directly: each extension host only has a
+	 * partial view, and a full-cache snapshot write is a lost-update hazard
+	 * when two hosts flush inside the watcher debounce window (see #1231).
+	 *
+	 * Safe to call while the in-process `withLock` is already held (e.g.
+	 * migration) — only the shared file lock is acquired here.
 	 */
 	private async writeIndex(): Promise<void> {
-		const indexPath = await this.getIndexPath()
-		const index: HistoryIndex = {
-			version: 1,
-			updatedAt: Date.now(),
-			entries: this.getAll(),
+		await taskHistoryLock.withLock(this.globalStoragePath, async () => {
+			const indexPath = await this.getIndexPath()
+			const entries = await this.collectIndexEntriesFromDisk()
+			const index: HistoryIndex = {
+				version: 1,
+				updatedAt: Date.now(),
+				entries,
+			}
+
+			await safeWriteJson(indexPath, index)
+		})
+	}
+
+	/**
+	 * Scan task directories and load each `history_item.json` for the index.
+	 * Per-task files are the source of truth; missing/corrupt files are skipped.
+	 */
+	private async collectIndexEntriesFromDisk(): Promise<HistoryItem[]> {
+		const tasksDir = await this.getTasksDir()
+
+		let dirEntries: string[]
+		try {
+			dirEntries = await fs.readdir(tasksDir)
+		} catch {
+			return []
+		}
+
+		const entries: HistoryItem[] = []
+		for (const name of dirEntries) {
+			if (name.startsWith("_") || name.startsWith(".")) {
+				continue
+			}
+
+			const item = await this.readTaskFile(name)
+			if (item) {
+				entries.push(item)
+			}
 		}
 
-		await safeWriteJson(indexPath, index)
+		return entries.sort((a, b) => b.ts - a.ts)
 	}
 
 	/**

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

@@ -0,0 +1,107 @@
+// 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"
+
+const { lockMock } = vi.hoisted(() => ({
+	lockMock: vi.fn(),
+}))
+
+vi.mock("proper-lockfile", () => ({
+	lock: lockMock,
+}))
+
+vi.mock("../../../utils/storage", () => ({
+	getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath),
+}))
+
+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 }).catch(() => {})
+	})
+
+	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)
+
+		await expect(taskHistoryLock.withLock(tmpDir, async () => "done")).resolves.toBe("done")
+
+		expect(lockMock).toHaveBeenCalledWith(
+			path.join(tmpDir, "tasks", GlobalFileNames.historyLock),
+			expect.any(Object),
+		)
+		expect(release).toHaveBeenCalledTimes(1)
+	})
+
+	it("keeps retrying long enough for proper-lockfile stale-lock recovery", async () => {
+		const taskHistoryLock = new TaskHistoryLock()
+
+		await taskHistoryLock.withLock(tmpDir, async () => undefined)
+
+		const options = lockMock.mock.calls[0][1] as {
+			stale: number
+			retries: { retries: number; factor: number; minTimeout: number; maxTimeout: number }
+		}
+		expect(cumulativeRetryWindowMs(options.retries)).toBeGreaterThan(options.stale)
+	})
+
+	it("serializes concurrent withLock callers in-process", async () => {
+		const taskHistoryLock = new TaskHistoryLock()
+		const order: string[] = []
+		let releaseFirst!: () => void
+		const firstHeld = new Promise<void>((resolve) => {
+			releaseFirst = resolve
+		})
+
+		lockMock.mockImplementation(async () => {
+			return async () => undefined
+		})
+
+		const first = taskHistoryLock.withLock(tmpDir, async () => {
+			order.push("first-start")
+			await firstHeld
+			order.push("first-end")
+			return 1
+		})
+		const second = taskHistoryLock.withLock(tmpDir, async () => {
+			order.push("second")
+			return 2
+		})
+
+		// Allow the first callback to start before releasing it.
+		await vi.waitFor(() => {
+			expect(order).toContain("first-start")
+		})
+		expect(order).not.toContain("second")
+
+		releaseFirst()
+		await expect(Promise.all([first, second])).resolves.toEqual([1, 2])
+		expect(order).toEqual(["first-start", "first-end", "second"])
+	})
+})

src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts

@@ -164,4 +164,52 @@ describe("TaskHistoryStore cross-instance safety", () => {
 		expect(storeA.getAll().length).toBe(10)
 		expect(storeB.getAll().length).toBe(10)
 	})
+
+	/**
+	 * Regression for #1231: two hosts each hold only their own task in cache and
+	 * flush `_index.json` without watcher reconciliation. A cache-snapshot writer
+	 * would drop the other host's entry; disk-authoritative rebuild must keep both.
+	 */
+	it("stale in-memory snapshots cannot drop peer entries from _index.json on flush", async () => {
+		await storeA.initialize()
+		await storeB.initialize()
+
+		// Model both processes flushing inside the watcher debounce window:
+		// no reconcile between upserts and index flushes.
+		disableBackgroundReconciliation(storeA)
+		disableBackgroundReconciliation(storeB)
+
+		await storeA.upsert(makeHistoryItem({ id: "task-a", task: "from A", ts: 1000 }))
+		await storeB.upsert(makeHistoryItem({ id: "task-b", task: "from B", ts: 2000 }))
+
+		// Each cache is intentionally partial — the pre-fix failure mode.
+		expect(storeA.get("task-a")).toBeDefined()
+		expect(storeA.get("task-b")).toBeUndefined()
+		expect(storeB.get("task-b")).toBeDefined()
+		expect(storeB.get("task-a")).toBeUndefined()
+
+		await storeA.flushIndex()
+		await storeB.flushIndex()
+
+		const tasksDir = path.join(tmpDir, "tasks")
+		const taskDirs = (await fs.readdir(tasksDir)).filter((name) => !name.startsWith("_") && !name.startsWith("."))
+		expect(taskDirs.sort()).toEqual(["task-a", "task-b"])
+
+		const indexRaw = await fs.readFile(path.join(tasksDir, GlobalFileNames.historyIndex), "utf8")
+		const index = JSON.parse(indexRaw) as { entries: HistoryItem[] }
+		const indexIds = index.entries.map((entry) => entry.id).sort()
+		expect(indexIds).toEqual(["task-a", "task-b"])
+	})
 })
+
+/** Stop fs.watch / periodic reconcile so flushes exercise the stale-cache path only. */
+function disableBackgroundReconciliation(store: TaskHistoryStore): void {
+	if (store["fsWatcher"]) {
+		store["fsWatcher"].close()
+		store["fsWatcher"] = null
+	}
+	if (store["reconcileTimer"]) {
+		clearTimeout(store["reconcileTimer"])
+		store["reconcileTimer"] = null
+	}
+}