fix(task-history): throw DeltaRejectedError on invalid merge and harden edge cases

Author: test22345

src/core/task-persistence/TaskHistoryStore.ts

@@ -13,6 +13,17 @@ import { getStorageBasePath } from "../../utils/storage"
 /** Valid status values for a task's HistoryItem. */
 export type HistoryItemStatus = NonNullable<HistoryItem["status"]>
 
+export class DeltaRejectedError extends Error {
+	constructor(
+		public readonly taskId: string,
+		public readonly diskStatus: HistoryItemStatus,
+		public readonly attemptedStatus: HistoryItemStatus,
+	) {
+		super(`Delta rejected for task ${taskId}: disk status ${diskStatus} rejects transition to ${attemptedStatus}`)
+		this.name = "DeltaRejectedError"
+	}
+}
+
 const VALID_TRANSITIONS: Record<HistoryItemStatus, HistoryItemStatus[]> = {
 	active: ["delegated", "completed", "interrupted"],
 	delegated: ["active"],
@@ -48,10 +59,7 @@ function mergeWithDisk(delta: Partial<HistoryItem>): (existing: unknown, incomin
 			if (delta.status !== diskStatus) {
 				const validTargets = VALID_TRANSITIONS[diskStatus]
 				if (!validTargets?.includes(delta.status as HistoryItemStatus)) {
-					console.warn(
-						`[TaskHistoryStore] Dropped stale delta for task ${disk.id}: disk status ${diskStatus} rejects transition to ${delta.status}`,
-					)
-					return disk
+					throw new DeltaRejectedError(disk.id, diskStatus, delta.status as HistoryItemStatus)
 				}
 			}
 		}
@@ -258,20 +266,35 @@ export class TaskHistoryStore {
 			if (item.status !== normalizedExisting) {
 				try {
 					assertValidTransition(existing.status, item.status)
-				} catch {
+				} catch (cacheError) {
 					// Cache may be stale from a peer write. Re-read disk
 					// under the store lock before rejecting the transition.
 					const diskItem = await this.readTaskFile(item.id)
-					assertValidTransition(diskItem?.status, item.status)
+					if (!diskItem) {
+						throw cacheError
+					}
+					assertValidTransition(diskItem.status, item.status)
 				}
 			}
 		}
 
 		// Merge: preserve existing metadata unless explicitly overwritten
 		const merged = existing ? { ...existing, ...item } : item
 
-		const delta = existing ? this.buildDelta(item.id, existing, item) : undefined
-		const written = await this.writeTaskFile(merged, delta)
+		const delta = existing ? this.buildDelta(item.id, existing, item) : { ...item }
+		let written: HistoryItem
+		try {
+			written = await this.writeTaskFile(merged, delta)
+		} catch (error) {
+			if (error instanceof DeltaRejectedError) {
+				const diskItem = await this.readTaskFile(item.id)
+				if (diskItem) {
+					this.cache.set(item.id, diskItem)
+				}
+				throw error
+			}
+			throw error
+		}
 
 		// Update in-memory cache with what was actually persisted
 		this.cache.set(written.id, written)
@@ -388,8 +411,10 @@ export class TaskHistoryStore {
 					// write is in progress — keep the task live.
 					try {
 						const lockPath = (await this.getTaskFilePath(taskId)) + ".lock"
-						await fs.access(lockPath)
-						liveIds.add(taskId)
+						const lockStat = await fs.stat(lockPath)
+						if (Date.now() - lockStat.mtimeMs < 31_000) {
+							liveIds.add(taskId)
+						}
 					} catch {
 						// No lock file — file is genuinely absent
 					}
@@ -1063,12 +1088,17 @@ export class TaskHistoryStore {
 			const mergedSecond = { ...second, ...updatedSecond }
 
 			const writtenFirst = await this.writeTaskFile(mergedFirst, this.buildDelta(firstId, first, updatedFirst))
-			const writtenSecond = await this.writeTaskFile(
-				mergedSecond,
-				this.buildDelta(secondId, second, updatedSecond),
-			)
+			let writtenSecond: HistoryItem
+			try {
+				writtenSecond = await this.writeTaskFile(mergedSecond, this.buildDelta(secondId, second, updatedSecond))
+			} catch (error) {
+				// First record is committed on disk. Update cache so it
+				// reflects disk state before propagating the error.
+				this.cache.set(firstId, writtenFirst)
+				throw error
+			}
 
-			// Both disk writes succeeded — now update the cache atomically.
+			// Both disk writes succeeded — now update the cache.
 			this.cache.set(firstId, writtenFirst)
 			this.cache.set(secondId, writtenSecond)
 

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", () => ({
@@ -245,7 +245,7 @@ describe("TaskHistoryStore cross-instance safety", () => {
 	 * 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 drops the entire delta", async () => {
+	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 })
@@ -261,15 +261,17 @@ describe("TaskHistoryStore cross-instance safety", () => {
 		// 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 storeA.upsert({
-			...staleItem,
-			status: "delegated",
-			awaitingChildId: "child-99",
-			delegatedToId: "child-99",
-		})
+		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.
+		// 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.

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

@@ -704,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")
 		})