diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index e4707ee0a9..2a9d7d7002 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -33,15 +33,6 @@ export function assertValidTransition(from: HistoryItemStatus | undefined, to: H } } -/** - * Index file format for fast startup reads. - */ -interface HistoryIndex { - version: number - updatedAt: number - entries: HistoryItem[] -} - /** * Durable intent for the one repair that spans an active delegated child and * its parent. Task files remain authoritative; this file only records the @@ -75,12 +66,14 @@ interface DelegationRepairIntent { * * Each task's HistoryItem is stored as an individual JSON file in its * existing task directory (`globalStorage/tasks//history_item.json`). - * A single index file (`globalStorage/tasks/_index.json`) is maintained - * as a cache for fast list reads at startup. + * There is no shared index file. Reads scan the task directories. * - * 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` with a `merge` callback: each write reads the + * current file under the advisory lock and merges incoming fields, so + * a concurrent writer's changes are preserved rather than silently + * dropped. Within a single extension host process, an in-process write + * lock serializes mutations. */ /** * Options for TaskHistoryStore constructor. @@ -100,7 +93,6 @@ export class TaskHistoryStore { private cache: Map = new Map() private taskFileMtimes: Map = new Map() private writeLock: Promise = Promise.resolve() - private indexWriteTimer: ReturnType | null = null private fsWatcher: fsSync.FSWatcher | null = null private reconcileTimer: ReturnType | null = null private disposed = false @@ -112,9 +104,6 @@ export class TaskHistoryStore { public readonly initialized: Promise private resolveInitialized!: () => void - /** Debounce window for index writes in milliseconds. */ - private static readonly INDEX_WRITE_DEBOUNCE_MS = 2000 - /** Periodic reconciliation interval in milliseconds. */ private static readonly RECONCILE_INTERVAL_MS = 5 * 60 * 1000 @@ -136,30 +125,27 @@ export class TaskHistoryStore { const tasksDir = await this.getTasksDir() await fs.mkdir(tasksDir, { recursive: true }) - // 1. Load existing index into the cache - await this.loadIndex() - - // 2. Reconcile cache against actual task directories on disk + // 1. Scan task directories to populate the cache await this.reconcile({ forceRefresh: true }) // Capture which active tasks were present in persisted state before replay can // change any statuses. Reconciliation must not treat a replay-repaired parent // as an orphaned active child in the same startup pass. const persistedActiveIds = this.getPersistedActiveIds() - // 3. Complete any two-record repair interrupted after its intent was durable. + // 2. Complete any two-record repair interrupted after its intent was durable. try { await this.replayDelegationRepairIntent() } catch (error) { console.error("[TaskHistoryStore] Failed to replay delegation repair intent:", error) } - // 4. Repair delegation inconsistencies left by a previous crash + // 3. Repair delegation inconsistencies left by a previous crash await this.reconcileDelegationState(persistedActiveIds) - // 5. Start fs.watch for cross-instance reactivity + // 4. Start fs.watch for cross-instance reactivity this.startWatcher() - // 6. Start periodic reconciliation as a defensive fallback + // 5. Start periodic reconciliation as a defensive fallback this.startPeriodicReconciliation() } finally { // Mark initialization as complete so callers awaiting `initialized` can proceed @@ -173,11 +159,6 @@ export class TaskHistoryStore { dispose(): void { this.disposed = true - if (this.indexWriteTimer) { - clearTimeout(this.indexWriteTimer) - this.indexWriteTimer = null - } - if (this.reconcileTimer) { clearTimeout(this.reconcileTimer) this.reconcileTimer = null @@ -187,11 +168,6 @@ export class TaskHistoryStore { this.fsWatcher.close() this.fsWatcher = null } - - // Synchronously flush the index (best-effort) - this.flushIndex().catch((err) => { - console.error("[TaskHistoryStore] Error flushing index on dispose:", err) - }) } // ────────────────────────────── Reads ────────────────────────────── @@ -257,13 +233,18 @@ export class TaskHistoryStore { // Merge: preserve existing metadata unless explicitly overwritten const merged = existing ? { ...existing, ...item } : item - // Write per-task file (source of truth) - await this.writeTaskFile(merged) + // Compute the actual changed fields relative to the cached state. + // Only these are applied to the disk version, so fields updated by + // another process are preserved rather than reverted from a stale cache. + const delta = existing + ? Object.fromEntries( + Object.entries(item).filter(([k, v]) => !deepEqual(v, (existing as Record)[k])), + ) + : undefined + await this.writeTaskFile(merged, delta ? ({ id: item.id, ...delta } as HistoryItem) : undefined) // Update in-memory cache this.cache.set(merged.id, merged) - // Schedule debounced index write - this.scheduleIndexWrite() const all = this.getAll() @@ -291,8 +272,6 @@ export class TaskHistoryStore { // File may already be deleted } - this.scheduleIndexWrite() - // Call onWrite callback inside the lock for serialized write-through if (this.onWrite) { await this.onWrite(this.getAll()) @@ -317,8 +296,6 @@ export class TaskHistoryStore { } } - this.scheduleIndexWrite() - // Call onWrite callback inside the lock for serialized write-through if (this.onWrite) { await this.onWrite(this.getAll()) @@ -346,20 +323,18 @@ export class TaskHistoryStore { return // tasks dir doesn't exist yet } - // Filter out the index file and hidden files + // Filter out hidden and reserved names const taskDirNames = dirEntries.filter((name) => !name.startsWith("_") && !name.startsWith(".")) const onDiskIds = new Set(taskDirNames) const cacheIds = new Set(this.cache.keys()) - let changed = false + const liveIds = new Set() - // Task files are authoritative during startup. Later watcher and periodic - // reconciliations use mtime change detection to avoid rewriting the index when - // nothing changed on disk. for (const taskId of onDiskIds) { try { const taskFilePath = await this.getTaskFilePath(taskId) const { mtimeMs } = await fs.stat(taskFilePath) + liveIds.add(taskId) if ( !options.forceRefresh && this.cache.has(taskId) && @@ -374,26 +349,20 @@ export class TaskHistoryStore { this.taskFileMtimes.set(taskId, mtimeMs) if (!deepEqual(previous, item)) { this.cache.set(taskId, item) - changed = true } } } catch { - // Corrupted or missing file, skip + // history_item.json missing or corrupt — not live } } - // Tasks in cache but not on disk: remove from cache + // Evict tasks whose history_item.json no longer exists for (const taskId of cacheIds) { - if (!onDiskIds.has(taskId)) { + if (!liveIds.has(taskId)) { this.cache.delete(taskId) this.taskFileMtimes.delete(taskId) - changed = true } } - - if (changed) { - this.scheduleIndexWrite() - } }) } @@ -584,12 +553,6 @@ export class TaskHistoryStore { await this.onWrite(this.getAll()) } await this.removeDelegationRepairIntent() - // Task files are authoritative and the intent is the recovery journal. - // Clean up the journal before scheduling the derived index: a crash after - // cleanup but before the index write is safe because startup rebuilds the - // index from task files, while the reverse ordering could make the index - // appear durable before recovery metadata is settled. - this.scheduleIndexWrite() }) } @@ -645,9 +608,6 @@ export class TaskHistoryStore { await this.onWrite(this.getAll()) } await this.removeDelegationRepairIntent() - // The index is derived state; keep the intent until authoritative task-file - // writes and write-through have completed, then schedule the index update. - this.scheduleIndexWrite() } private matchesDelegationRepairParentPreconditions(intent: DelegationRepairIntent, parent: HistoryItem): boolean { @@ -845,96 +805,36 @@ export class TaskHistoryStore { } } - // Write the index - await this.writeIndex() - // Repair any delegation inconsistencies introduced by the migrated entries. // Run the lock-free core because migration already holds the store lock. await this.reconcileDelegationStateCore(this.getPersistedActiveIds()) }) } - // ────────────────────────────── Private: Index management ────────────────────────────── - - /** - * Load the `_index.json` file into the in-memory cache. - */ - private async loadIndex(): Promise { - const indexPath = await this.getIndexPath() - - try { - const raw = await fs.readFile(indexPath, "utf8") - const index: HistoryIndex = JSON.parse(raw) - - if (index.version === 1 && Array.isArray(index.entries)) { - for (const entry of index.entries) { - if (entry.id) { - this.cache.set(entry.id, entry) - } - } - } - } catch { - // Index doesn't exist or is corrupted; cache stays empty. - // Reconciliation will rebuild it from per-task files. - } - } - - /** - * Write the full index to disk. - */ - private async writeIndex(): Promise { - const indexPath = await this.getIndexPath() - const index: HistoryIndex = { - version: 1, - updatedAt: Date.now(), - entries: this.getAll(), - } - - await safeWriteJson(indexPath, index) - } - - /** - * Schedule a debounced index write. - */ - private scheduleIndexWrite(): void { - if (this.disposed) { - return - } - - if (this.indexWriteTimer) { - clearTimeout(this.indexWriteTimer) - } - - this.indexWriteTimer = setTimeout(async () => { - this.indexWriteTimer = null - try { - await this.writeIndex() - } catch (err) { - console.error("[TaskHistoryStore] Failed to write index:", err) - } - }, TaskHistoryStore.INDEX_WRITE_DEBOUNCE_MS) - } - - /** - * Force an immediate index write (called on dispose/shutdown). - */ - async flushIndex(): Promise { - if (this.indexWriteTimer) { - clearTimeout(this.indexWriteTimer) - this.indexWriteTimer = null - } - - await this.writeIndex() - } - // ────────────────────────────── Private: Per-task file I/O ────────────────────────────── /** * Write a HistoryItem to its per-task `history_item.json` file. + * + * When `delta` is provided, the merge callback applies only the + * delta to the current disk state, so fields written by another + * process are preserved. Without a delta the full item is written + * as-is (used by administrative repair paths that are authoritative). */ - private async writeTaskFile(item: HistoryItem): Promise { + private async writeTaskFile(item: HistoryItem, delta?: Partial): Promise { const filePath = await this.getTaskFilePath(item.id) - await safeWriteJson(filePath, item) + if (delta) { + await safeWriteJson(filePath, item, { + merge: (existing, incoming) => { + if (!existing || typeof existing !== "object" || !("id" in existing)) { + return incoming + } + return { ...existing, ...delta } + }, + }) + } else { + await safeWriteJson(filePath, item) + } } /** @@ -1105,16 +1005,23 @@ export class TaskHistoryStore { const mergedFirst = { ...first, ...updatedFirst } const mergedSecond = { ...second, ...updatedSecond } + // Compute actual diffs against cached state, mirroring upsertCore. + const deltaFirst = Object.fromEntries( + Object.entries(updatedFirst).filter(([k, v]) => !deepEqual(v, (first as Record)[k])), + ) + const deltaSecond = Object.fromEntries( + Object.entries(updatedSecond).filter(([k, v]) => !deepEqual(v, (second as Record)[k])), + ) + // Write both files before touching the cache so readers never observe a // half-updated in-memory state between the two await points. - await this.writeTaskFile(mergedFirst) - await this.writeTaskFile(mergedSecond) + await this.writeTaskFile(mergedFirst, { id: firstId, ...deltaFirst } as HistoryItem) + await this.writeTaskFile(mergedSecond, { id: secondId, ...deltaSecond } as HistoryItem) // Both disk writes succeeded — now update the cache atomically. this.cache.set(firstId, mergedFirst) this.cache.set(secondId, mergedSecond) - this.scheduleIndexWrite() const all = this.getAll() if (this.onWrite) { await this.onWrite(all) @@ -1155,12 +1062,4 @@ export class TaskHistoryStore { const tasksDir = await this.getTasksDir() return path.join(tasksDir, taskId, GlobalFileNames.historyItem) } - - /** - * Get the path to the `_index.json` file. - */ - private async getIndexPath(): Promise { - const tasksDir = await this.getTasksDir() - return path.join(tasksDir, GlobalFileNames.historyIndex) - } } diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts index e5166c478c..bb21f5adb5 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts @@ -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 { @@ -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,62 @@ 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. + expect(storeA.get("shared-task")!.totalCost).toBe(9.99) + }) + + /** + * 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) + }) }) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index e788b5d96a..e37fd1a25e 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/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 replayDelegationRepairIntent: () => Promise } - 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() }) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts index 3188e9c505..8d23623ead 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts +++ b/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((resolve) => { signalMigrationWriteStarted = resolve }) - const storeInternals = store as unknown as { writeIndex: () => Promise } - 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) }) }) diff --git a/src/core/webview/__tests__/webviewMessageHandler.importRooHistory.spec.ts b/src/core/webview/__tests__/webviewMessageHandler.importRooHistory.spec.ts index df85ff1df4..1f23e353c6 100644 --- a/src/core/webview/__tests__/webviewMessageHandler.importRooHistory.spec.ts +++ b/src/core/webview/__tests__/webviewMessageHandler.importRooHistory.spec.ts @@ -52,7 +52,6 @@ describe("webviewMessageHandler - importRooHistory", () => { taskHistoryStore: { invalidateAll: ReturnType reconcile: ReturnType - flushIndex: ReturnType } postMessageToWebview: ReturnType postStateToWebview: ReturnType @@ -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, { diff --git a/src/core/webview/webviewMessageHandler.ts b/src/core/webview/webviewMessageHandler.ts index f0fc33501f..2e2a4c8f58 100644 --- a/src/core/webview/webviewMessageHandler.ts +++ b/src/core/webview/webviewMessageHandler.ts @@ -997,7 +997,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", diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 87ff0dfec5..040bfba450 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -779,11 +779,6 @@ "count": 4 } }, - "core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts": { - "@typescript-eslint/no-explicit-any": { - "count": 1 - } - }, "core/task-persistence/__tests__/TaskHistoryStore.spec.ts": { "@typescript-eslint/no-explicit-any": { "count": 4 diff --git a/src/shared/globalFileNames.ts b/src/shared/globalFileNames.ts index 7bfe18f4bc..9f15a06319 100644 --- a/src/shared/globalFileNames.ts +++ b/src/shared/globalFileNames.ts @@ -5,6 +5,5 @@ export const GlobalFileNames = { customModes: "custom_modes.yaml", taskMetadata: "task_metadata.json", historyItem: "history_item.json", - historyIndex: "_index.json", delegationRepairIntent: "_delegation_repair_intent.json", } diff --git a/src/utils/__tests__/safeWriteJson.test.ts b/src/utils/__tests__/safeWriteJson.test.ts index e060de4a31..bc1dcfca8a 100644 --- a/src/utils/__tests__/safeWriteJson.test.ts +++ b/src/utils/__tests__/safeWriteJson.test.ts @@ -468,4 +468,43 @@ describe("safeWriteJson", () => { consoleErrorSpy.mockRestore() }) + + // Merge option tests + test("should merge incoming data with existing file content when merge callback is provided", async () => { + const initial = { a: 1, b: 2 } + await safeWriteJson(currentTestFilePath, initial) + + const incoming = { b: 3, c: 4 } + await safeWriteJson(currentTestFilePath, incoming, { + merge: (existing, data) => ({ + ...(existing as Record), + ...(data as Record), + }), + }) + + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual({ a: 1, b: 3, c: 4 }) + }) + + test("should pass null to merge callback when file does not exist", async () => { + const newFilePath = path.join(tempDir, "nonexistent.json") + const mergeFn = vi.fn((existing, incoming) => incoming) + + await safeWriteJson(newFilePath, { value: 42 }, { merge: mergeFn }) + + expect(mergeFn).toHaveBeenCalledWith(null, { value: 42 }) + const content = await readFileContent(newFilePath) + expect(content).toEqual({ value: 42 }) + }) + + test("should write incoming data directly when no merge callback is provided", async () => { + const initial = { a: 1, b: 2 } + await safeWriteJson(currentTestFilePath, initial) + + const replacement = { c: 3 } + await safeWriteJson(currentTestFilePath, replacement) + + const content = await readFileContent(currentTestFilePath) + expect(content).toEqual({ c: 3 }) + }) }) diff --git a/src/utils/safeWriteJson.ts b/src/utils/safeWriteJson.ts index c32dd92ce5..277929e0b3 100644 --- a/src/utils/safeWriteJson.ts +++ b/src/utils/safeWriteJson.ts @@ -15,6 +15,16 @@ export interface SafeWriteJsonOptions { * @default false */ prettyPrint?: boolean + + /** + * When provided, the current file is read under the advisory lock + * and passed to this function along with the incoming data. The + * return value replaces `data` for the write. This turns a blind + * overwrite into an atomic read-modify-write, preventing cross-process + * lost updates. `existing` is null when the file does not exist or + * cannot be parsed. + */ + merge?: (existing: unknown, incoming: unknown) => unknown } /** @@ -83,6 +93,19 @@ async function safeWriteJson(filePath: string, data: any, options?: SafeWriteJso let actualTempBackupFilePath: string | null = null try { + // If a merge callback was provided, read the current file under the lock + // and let the caller merge before we write. Must be inside try/finally + // so a throwing merge still releases the lock. + if (options?.merge) { + let existing: unknown = null + try { + existing = JSON.parse(await fs.readFile(absoluteFilePath, "utf8")) + } catch { + // No readable file yet, so the merge receives null. + } + data = options.merge(existing, data) + } + // Step 1: Write data to a new temporary file. actualTempNewFilePath = path.join( path.dirname(absoluteFilePath),