fix(task-history): atomic per-task merge and drop shared index file (#1231)

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

@@ -6,7 +6,7 @@ import * as os from "os"
 
 import type { HistoryItem } from "@roo-code/types"
 
-import { TaskHistoryStore } from "../TaskHistoryStore"
+import { TaskHistoryStore, DeltaRejectedError } from "../TaskHistoryStore"
 import { GlobalFileNames } from "../../../shared/globalFileNames"
 
 vi.mock("../../../utils/storage", () => ({
@@ -15,12 +15,30 @@ vi.mock("../../../utils/storage", () => ({
 	}),
 }))
 
-// Mock safeWriteJson to use plain fs writes in tests (avoids proper-lockfile issues)
+// Mock safeWriteJson to use plain fs writes but honor the merge callback.
 vi.mock("../../../utils/safeWriteJson", () => ({
-	safeWriteJson: vi.fn().mockImplementation(async (filePath: string, data: any) => {
-		await fs.mkdir(path.dirname(filePath), { recursive: true })
-		await fs.writeFile(filePath, JSON.stringify(data, null, "\t"), "utf8")
-	}),
+	safeWriteJson: vi
+		.fn()
+		.mockImplementation(
+			async (
+				filePath: string,
+				data: unknown,
+				options?: { merge?: (existing: unknown, incoming: unknown) => unknown },
+			) => {
+				await fs.mkdir(path.dirname(filePath), { recursive: true })
+				if (options?.merge) {
+					let existing: unknown = null
+					try {
+						const raw = await fs.readFile(filePath, "utf8")
+						existing = JSON.parse(raw)
+					} catch {
+						// File does not exist or is corrupt
+					}
+					data = options.merge(existing, data)
+				}
+				await fs.writeFile(filePath, JSON.stringify(data, null, "\t"), "utf8")
+			},
+		),
 }))
 
 function makeHistoryItem(overrides: Partial<HistoryItem> = {}): HistoryItem {
@@ -124,6 +142,27 @@ describe("TaskHistoryStore cross-instance safety", () => {
 		expect(storeB.get("shared-task")).toBeUndefined()
 	})
 
+	it("delete by instance A is detected even when the task directory remains", async () => {
+		await storeA.initialize()
+		await storeB.initialize()
+
+		const item = makeHistoryItem({ id: "file-only-delete" })
+		await storeA.upsert(item)
+		await storeB.reconcile()
+
+		expect(storeB.get("file-only-delete")).toBeDefined()
+
+		// delete() unlinks history_item.json but leaves the task directory.
+		await storeA.delete("file-only-delete")
+
+		// Directory still exists (other files like ui_messages.json may remain).
+		const taskDir = path.join(tmpDir, "tasks", "file-only-delete")
+		await expect(fs.access(taskDir)).resolves.toBeUndefined()
+
+		await storeB.reconcile()
+		expect(storeB.get("file-only-delete")).toBeUndefined()
+	})
+
 	it("per-task file updates by one instance are visible to another after invalidation", async () => {
 		await storeA.initialize()
 		await storeB.initialize()
@@ -164,4 +203,107 @@ describe("TaskHistoryStore cross-instance safety", () => {
 		expect(storeA.getAll().length).toBe(10)
 		expect(storeB.getAll().length).toBe(10)
 	})
+
+	/**
+	 * Host B completes a task on disk while host A's cache still has it
+	 * active. Host A's next save updates only totalCost (a full-object
+	 * upsert — the realistic production shape). The diff-delta merge
+	 * preserves B's status because status did not change in A's cache.
+	 */
+	it("per-task diff-delta preserves a peer's status change on full-object upsert", async () => {
+		await storeA.initialize()
+
+		// Base item with an explicit status — mirrors real production items.
+		const base = makeHistoryItem({ id: "shared-task", status: "active", totalCost: 0.01, ts: 1000 })
+		await storeA.upsert(base)
+
+		// Host B completes the task on disk; A's cache still has "active".
+		const filePath = path.join(tmpDir, "tasks", "shared-task", GlobalFileNames.historyItem)
+		const onDisk = JSON.parse(await fs.readFile(filePath, "utf8"))
+		onDisk.status = "completed"
+		onDisk.completionResultSummary = "done by host B"
+		await fs.writeFile(filePath, JSON.stringify(onDisk), "utf8")
+
+		// Host A does a full-object upsert (the realistic path — spread the
+		// cached item and change one field). The cached item has status: "active".
+		await storeA.upsert({ ...storeA.get("shared-task")!, totalCost: 9.99 })
+
+		const final = JSON.parse(await fs.readFile(filePath, "utf8")) as HistoryItem
+		expect(final.totalCost).toBe(9.99)
+		// Status is preserved from disk because A's delta does not include
+		// status — it was unchanged relative to A's cache.
+		expect(final.status).toBe("completed")
+		expect(final.completionResultSummary).toBe("done by host B")
+
+		// Cache reflects the caller's totalCost change and the peer's status.
+		expect(storeA.get("shared-task")!.totalCost).toBe(9.99)
+		expect(storeA.get("shared-task")!.status).toBe("completed")
+		expect(storeA.get("shared-task")!.completionResultSummary).toBe("done by host B")
+	})
+
+	/**
+	 * Regression: a stale host whose cache says "active" tries to write
+	 * status: "delegated" after a peer already wrote "completed" to disk.
+	 * The merge must reject the entire delta (including companion fields)
+	 * to prevent an internally-inconsistent record.
+	 */
+	it("merge rejects an invalid status transition against disk and throws DeltaRejectedError", async () => {
+		await storeA.initialize()
+
+		const base = makeHistoryItem({ id: "guarded-task", status: "active", totalCost: 0.01, ts: 1000 })
+		await storeA.upsert(base)
+
+		// Peer writes terminal "completed" directly to disk.
+		const filePath = path.join(tmpDir, "tasks", "guarded-task", GlobalFileNames.historyItem)
+		const onDisk = JSON.parse(await fs.readFile(filePath, "utf8"))
+		onDisk.status = "completed"
+		onDisk.completionResultSummary = "done by peer"
+		await fs.writeFile(filePath, JSON.stringify(onDisk), "utf8")
+
+		// Host A's cache still has "active". It tries to delegate (active → delegated
+		// passes the cache check, but completed → delegated is invalid on disk).
+		const staleItem = storeA.get("guarded-task")!
+		await expect(
+			storeA.upsert({
+				...staleItem,
+				status: "delegated",
+				awaitingChildId: "child-99",
+				delegatedToId: "child-99",
+			}),
+		).rejects.toThrow(DeltaRejectedError)
+
+		const final = JSON.parse(await fs.readFile(filePath, "utf8")) as HistoryItem
+		// Terminal status must survive — disk is untouched.
+		expect(final.status).toBe("completed")
+		expect(final.completionResultSummary).toBe("done by peer")
+		// Companion fields from the rejected delta must NOT be applied.
+		expect(final.awaitingChildId).toBeUndefined()
+		expect(final.delegatedToId).toBeUndefined()
+
+		// Cache must reflect the disk state, not the stale delta.
+		expect(storeA.get("guarded-task")!.status).toBe("completed")
+	})
+
+	/**
+	 * When both hosts change the same field, the last writer wins.
+	 * This is expected — true conflict resolution requires application
+	 * semantics that a generic merge cannot provide.
+	 */
+	it("same-field changes from both hosts are last-writer-wins", async () => {
+		await storeA.initialize()
+		await storeB.initialize()
+
+		const base = makeHistoryItem({ id: "shared-task", status: "active", totalCost: 0.01, ts: 1000 })
+		await storeA.upsert(base)
+		await storeB.reconcile()
+
+		// Both hosts change totalCost.
+		await storeA.upsert({ ...storeA.get("shared-task")!, totalCost: 1.0 })
+		await storeB.upsert({ ...storeB.get("shared-task")!, totalCost: 2.0 })
+
+		const filePath = path.join(tmpDir, "tasks", "shared-task", GlobalFileNames.historyItem)
+		const final = JSON.parse(await fs.readFile(filePath, "utf8")) as HistoryItem
+		// B wrote last, so B's value wins.
+		expect(final.totalCost).toBe(2.0)
+	})
 })

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

@@ -465,7 +465,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => {
 		await expect(fs.access(intentPath)).rejects.toThrow()
 	})
 
-	it("does not schedule the derived index before repair-intent cleanup succeeds", async () => {
+	it("removes the repair-intent file after successful replay", async () => {
 		const child = makeItem({ id: "child-deferred-index", status: "active", parentTaskId: "parent-deferred-index" })
 		const parent = makeItem({
 			id: "parent-deferred-index",
@@ -478,23 +478,12 @@ describe("TaskHistoryStore reconcileDelegationState", () => {
 		await fs.writeFile(intentPath, JSON.stringify(makeRepairIntent(parent, child)))
 		await store.reconcile({ forceRefresh: true })
 
-		const events: string[] = []
 		const storeInternals = store as unknown as {
-			scheduleIndexWrite: () => void
-			removeDelegationRepairIntent: () => Promise<void>
 			replayDelegationRepairIntent: () => Promise<void>
 		}
-		vi.spyOn(storeInternals, "removeDelegationRepairIntent").mockImplementation(async () => {
-			events.push("cleanup")
-			await fs.unlink(intentPath)
-		})
-		vi.spyOn(storeInternals, "scheduleIndexWrite").mockImplementation(() => {
-			events.push("schedule")
-		})
 
 		await storeInternals.replayDelegationRepairIntent()
 
-		expect(events).toEqual(["cleanup", "schedule"])
 		await expect(fs.access(intentPath)).rejects.toThrow()
 	})
 

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

@@ -57,29 +57,19 @@ describe("TaskHistoryStore", () => {
 			expect(store.getAll()).toEqual([])
 		})
 
-		it("initializes from existing index file", async () => {
+		it("initializes from existing per-task files", async () => {
 			const tasksDir = path.join(tmpDir, "tasks")
 			await fs.mkdir(tasksDir, { recursive: true })
 
 			const item1 = makeHistoryItem({ id: "task-1", ts: 1000 })
 			const item2 = makeHistoryItem({ id: "task-2", ts: 2000 })
 
-			// Create task directories so reconciliation doesn't remove them
 			await fs.mkdir(path.join(tasksDir, "task-1"), { recursive: true })
 			await fs.mkdir(path.join(tasksDir, "task-2"), { recursive: true })
 
-			// Write per-task files
 			await fs.writeFile(path.join(tasksDir, "task-1", GlobalFileNames.historyItem), JSON.stringify(item1))
 			await fs.writeFile(path.join(tasksDir, "task-2", GlobalFileNames.historyItem), JSON.stringify(item2))
 
-			// Write index
-			const index = {
-				version: 1,
-				updatedAt: Date.now(),
-				entries: [item1, item2],
-			}
-			await fs.writeFile(path.join(tasksDir, GlobalFileNames.historyIndex), JSON.stringify(index))
-
 			await store.initialize()
 
 			expect(store.getAll()).toHaveLength(2)
@@ -374,7 +364,7 @@ describe("TaskHistoryStore", () => {
 			expect(store.get("idem-task")).toBeDefined()
 		})
 
-		it("serializes migration cache and index updates behind the store lock", async () => {
+		it("serializes migration cache updates behind the store lock", async () => {
 			const tasksDir = path.join(tmpDir, "tasks")
 			const migrated = makeHistoryItem({ id: "migration-locked" })
 			const concurrent = makeHistoryItem({ id: "migration-concurrent" })
@@ -389,12 +379,17 @@ describe("TaskHistoryStore", () => {
 			const migrationWriteStarted = new Promise<void>((resolve) => {
 				signalMigrationWriteStarted = resolve
 			})
-			const storeInternals = store as unknown as { writeIndex: () => Promise<void> }
-			const originalWriteIndex = storeInternals.writeIndex.bind(store)
-			vi.spyOn(storeInternals, "writeIndex").mockImplementation(async () => {
-				signalMigrationWriteStarted()
-				await migrationWriteCanFinish
-				return originalWriteIndex()
+
+			const { safeWriteJson: mockSafeWriteJson } = await import("../../../utils/safeWriteJson")
+			const originalImpl = vi.mocked(mockSafeWriteJson).getMockImplementation()!
+			let firstCall = true
+			vi.mocked(mockSafeWriteJson).mockImplementation(async (...args) => {
+				if (firstCall) {
+					firstCall = false
+					signalMigrationWriteStarted()
+					await migrationWriteCanFinish
+				}
+				return originalImpl(...args)
 			})
 
 			const migration = store.migrateFromGlobalState([migrated])
@@ -407,45 +402,6 @@ describe("TaskHistoryStore", () => {
 
 			expect(store.get(migrated.id)).toEqual(migrated)
 			expect(store.get(concurrent.id)).toEqual(concurrent)
-			await store.flushIndex()
-			const index = JSON.parse(await fs.readFile(path.join(tasksDir, GlobalFileNames.historyIndex), "utf8")) as {
-				entries: HistoryItem[]
-			}
-			expect(index.entries.map((entry) => entry.id)).toEqual(expect.arrayContaining([migrated.id, concurrent.id]))
-		})
-	})
-
-	describe("flushIndex()", () => {
-		it("writes index to disk on flush", async () => {
-			await store.initialize()
-
-			await store.upsert(makeHistoryItem({ id: "flush-task" }))
-			await store.flushIndex()
-
-			const indexPath = path.join(tmpDir, "tasks", GlobalFileNames.historyIndex)
-			const raw = await fs.readFile(indexPath, "utf8")
-			const index = JSON.parse(raw)
-
-			expect(index.version).toBe(1)
-			expect(index.entries).toHaveLength(1)
-			expect(index.entries[0].id).toBe("flush-task")
-		})
-	})
-
-	describe("dispose()", () => {
-		it("flushes index on dispose", async () => {
-			await store.initialize()
-
-			await store.upsert(makeHistoryItem({ id: "dispose-task" }))
-			store.dispose()
-
-			// Give the flush a moment to complete
-			await new Promise((resolve) => setTimeout(resolve, 100))
-
-			const indexPath = path.join(tmpDir, "tasks", GlobalFileNames.historyIndex)
-			const raw = await fs.readFile(indexPath, "utf8")
-			const index = JSON.parse(raw)
-			expect(index.entries).toHaveLength(1)
 		})
 	})
 
@@ -748,8 +704,9 @@ describe("TaskHistoryStore", () => {
 			const parentDisk = JSON.parse(await fs.readFile(parentFile, "utf8"))
 			expect(parentDisk.status).toBe("delegated")
 
-			// Cache was NOT updated (cache set is deferred until after both writes succeed)
-			expect(store.get("child-partial")?.status).toBe("active")
+			// First record's cache IS updated (it was committed to disk).
+			// Second record's cache is unchanged (write never completed).
+			expect(store.get("child-partial")?.status).toBe("completed")
 			expect(store.get("parent-partial")?.status).toBe("delegated")
 		})
 

src/core/webview/__tests__/webviewMessageHandler.importRooHistory.spec.ts

@@ -52,7 +52,6 @@ describe("webviewMessageHandler - importRooHistory", () => {
 		taskHistoryStore: {
 			invalidateAll: ReturnType<typeof vi.fn>
 			reconcile: ReturnType<typeof vi.fn>
-			flushIndex: ReturnType<typeof vi.fn>
 		}
 		postMessageToWebview: ReturnType<typeof vi.fn>
 		postStateToWebview: ReturnType<typeof vi.fn>
@@ -71,7 +70,6 @@ describe("webviewMessageHandler - importRooHistory", () => {
 			taskHistoryStore: {
 				invalidateAll: vi.fn(),
 				reconcile: vi.fn().mockResolvedValue(undefined),
-				flushIndex: vi.fn().mockResolvedValue(undefined),
 			},
 			postMessageToWebview: vi.fn().mockResolvedValue(undefined),
 			postStateToWebview: vi.fn().mockResolvedValue(undefined),
@@ -106,7 +104,7 @@ describe("webviewMessageHandler - importRooHistory", () => {
 		expect(importRooTaskHistoryMock).toHaveBeenCalledWith("/mock/storage", expect.any(Function))
 		expect(mockProvider.taskHistoryStore.invalidateAll).toHaveBeenCalledTimes(1)
 		expect(mockProvider.taskHistoryStore.reconcile).toHaveBeenCalledTimes(1)
-		expect(mockProvider.taskHistoryStore.flushIndex).toHaveBeenCalledTimes(1)
+
 		expect(mockProvider.postStateToWebview).toHaveBeenCalledTimes(1)
 		expect(mockProvider.postMessageToWebview).toHaveBeenNthCalledWith(1, {
 			type: "rooHistoryImportProgress",
@@ -189,7 +187,7 @@ describe("webviewMessageHandler - importRooHistory", () => {
 		expect(importRooTaskHistoryMock).toHaveBeenCalledWith("/mock/storage", expect.any(Function))
 		expect(mockProvider.taskHistoryStore.invalidateAll).not.toHaveBeenCalled()
 		expect(mockProvider.taskHistoryStore.reconcile).not.toHaveBeenCalled()
-		expect(mockProvider.taskHistoryStore.flushIndex).not.toHaveBeenCalled()
+
 		expect(mockProvider.postStateToWebview).not.toHaveBeenCalled()
 		expect(mockProvider.postMessageToWebview).toHaveBeenNthCalledWith(2, {
 			type: "rooHistoryImportProgress",
@@ -222,7 +220,7 @@ describe("webviewMessageHandler - importRooHistory", () => {
 		// after a partial-copy failure still reconciles the store.
 		expect(mockProvider.taskHistoryStore.invalidateAll).toHaveBeenCalledTimes(1)
 		expect(mockProvider.taskHistoryStore.reconcile).toHaveBeenCalledTimes(1)
-		expect(mockProvider.taskHistoryStore.flushIndex).toHaveBeenCalledTimes(1)
+
 		expect(mockProvider.postStateToWebview).toHaveBeenCalledTimes(1)
 		expect(vscode.window.showWarningMessage).toHaveBeenCalledWith(
 			"common:warnings.rooHistoryImport.alreadyImported",
@@ -237,7 +235,7 @@ describe("webviewMessageHandler - importRooHistory", () => {
 
 		expect(mockProvider.taskHistoryStore.invalidateAll).not.toHaveBeenCalled()
 		expect(mockProvider.taskHistoryStore.reconcile).not.toHaveBeenCalled()
-		expect(mockProvider.taskHistoryStore.flushIndex).not.toHaveBeenCalled()
+
 		expect(mockProvider.postStateToWebview).not.toHaveBeenCalled()
 		expect(mockProvider.log).toHaveBeenCalledWith("[importRooHistory] failed: permission denied")
 		expect(mockProvider.postMessageToWebview).toHaveBeenNthCalledWith(2, {

src/core/webview/webviewMessageHandler.ts

@@ -1002,7 +1002,6 @@ export const webviewMessageHandler = async (
 				// so a retry after a partial-copy failure still reconciles the store.
 				await provider.taskHistoryStore.invalidateAll()
 				await provider.taskHistoryStore.reconcile()
-				await provider.taskHistoryStore.flushIndex()
 				await provider.postStateToWebview()
 				await provider.postMessageToWebview({
 					type: "rooHistoryImportProgress",