From aee7d64bc4a9283dee4aa61351cd44de20deaee8 Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 16 Aug 2026 21:15:22 +0000 Subject: [PATCH] fix(task-history): prevent concurrent index clobbering --- src/core/task-persistence/TaskHistoryLock.ts | 104 +++++ src/core/task-persistence/TaskHistoryStore.ts | 64 ++- .../__tests__/TaskHistoryLock.spec.ts | 107 +++++ .../TaskHistoryStore.crossInstance.spec.ts | 48 +++ .../TaskHistoryStore.process.spec.ts | 385 ++++++++++++++++++ .../TaskHistoryStore.reconciliation.spec.ts | 1 + .../__tests__/TaskHistoryStore.spec.ts | 24 ++ .../fixtures/taskHistoryProcessProtocol.ts | 40 ++ .../fixtures/taskHistoryProcessWorker.ts | 211 ++++++++++ .../__tests__/fixtures/tsconfig.json | 10 + src/shared/globalFileNames.ts | 2 + 11 files changed, 986 insertions(+), 10 deletions(-) create mode 100644 src/core/task-persistence/TaskHistoryLock.ts create mode 100644 src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts create mode 100644 src/core/task-persistence/__tests__/TaskHistoryStore.process.spec.ts create mode 100644 src/core/task-persistence/__tests__/fixtures/taskHistoryProcessProtocol.ts create mode 100644 src/core/task-persistence/__tests__/fixtures/taskHistoryProcessWorker.ts create mode 100644 src/core/task-persistence/__tests__/fixtures/tsconfig.json diff --git a/src/core/task-persistence/TaskHistoryLock.ts b/src/core/task-persistence/TaskHistoryLock.ts new file mode 100644 index 0000000000..89a7822f5e --- /dev/null +++ b/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 = 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(globalStoragePath: string, fn: () => Promise): Promise { + 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 { + 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(lockFilePath: string, fn: () => Promise): Promise { + let releaseLock: (() => Promise) | 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() diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index e4707ee0a9..937b595dd0 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/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 @@ -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 { - 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 { + 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) } /** diff --git a/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts new file mode 100644 index 0000000000..2ed881ba0d --- /dev/null +++ b/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((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"]) + }) +}) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts index e5166c478c..29264052a4 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts +++ b/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 + } +} diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.process.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.process.spec.ts new file mode 100644 index 0000000000..075bcb267a --- /dev/null +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.process.spec.ts @@ -0,0 +1,385 @@ +// pnpm --dir src exec vitest run core/task-persistence/__tests__/TaskHistoryStore.process.spec.ts + +import { spawn, type ChildProcess } from "child_process" +import { createRequire } from "module" +import * as fs from "fs/promises" +import * as os from "os" +import * as path from "path" +import { fileURLToPath, pathToFileURL } from "url" + +import type { HistoryItem } from "@roo-code/types" + +import { GlobalFileNames } from "../../../shared/globalFileNames" +import type { ParentToWorkerMessage, WorkerId, WorkerToParentMessage } from "./fixtures/taskHistoryProcessProtocol" + +const DIAGNOSTIC_TIMEOUT_MS = 8_000 +const MAX_CAPTURED_OUTPUT_BYTES = 32 * 1024 +const require = createRequire(import.meta.url) +const tsxLoaderUrl = pathToFileURL(require.resolve("tsx")).href +const workerPath = fileURLToPath(new URL("./fixtures/taskHistoryProcessWorker.ts", import.meta.url)) +const workerTsconfigPath = fileURLToPath(new URL("./fixtures/tsconfig.json", import.meta.url)) +const repositoryRoot = fileURLToPath(new URL("../../../../", import.meta.url)) + +interface HistoryIndex { + version: number + entries: HistoryItem[] +} + +class ProcessWorker { + private readonly child: ChildProcess + private readonly events: WorkerToParentMessage[] = [] + private readonly waiters = new Set<{ + predicate: (event: WorkerToParentMessage) => boolean + resolve: (event: WorkerToParentMessage) => void + reject: (error: Error) => void + timer: ReturnType + }>() + private stdout = "" + private stderr = "" + private terminalError: Error | undefined + private exited = false + private expectedExit = false + + constructor( + readonly workerId: WorkerId, + private readonly onEvent?: (event: WorkerToParentMessage) => void, + ) { + this.child = spawn(process.execPath, ["--import", tsxLoaderUrl, workerPath], { + cwd: repositoryRoot, + env: { ...process.env, TSX_TSCONFIG_PATH: workerTsconfigPath }, + stdio: ["ignore", "pipe", "pipe", "ipc"], + }) + + this.child.stdout?.on("data", (chunk: Buffer | string) => { + this.stdout = appendBounded(this.stdout, chunk.toString()) + }) + this.child.stderr?.on("data", (chunk: Buffer | string) => { + this.stderr = appendBounded(this.stderr, chunk.toString()) + }) + this.child.on("message", (value: unknown) => this.handleEvent(value)) + this.child.on("error", (error) => this.fail(new Error(`Worker ${workerId} process error: ${error.message}`))) + this.child.on("exit", (code, signal) => { + this.exited = true + if (!this.expectedExit || code !== 0 || this.waiters.size > 0) { + this.fail( + new Error( + `Worker ${workerId} exited prematurely (code=${String(code)}, signal=${String(signal)})${this.diagnostics()}`, + ), + ) + } + }) + } + + send(message: ParentToWorkerMessage): void { + if (this.terminalError) throw this.terminalError + if (!this.child.connected) + throw new Error(`Worker ${this.workerId} IPC channel is disconnected${this.diagnostics()}`) + this.child.send(message, (error) => { + if (error) + this.fail(new Error(`Failed to send to worker ${this.workerId}: ${error.message}${this.diagnostics()}`)) + }) + } + + async initialize(storageRoot: string, pauseFirstLockCallback = false): Promise { + this.send({ type: "initialize", workerId: this.workerId, storageRoot, pauseFirstLockCallback }) + await this.waitFor((event) => event.type === "initialized", "initialize") + } + + async stage(requestId: string, item: HistoryItem): Promise> { + this.send({ type: "stage", requestId, item }) + return this.waitForEventType("staged", requestId) + } + + flush(requestId: string): void { + this.send({ type: "flush", requestId }) + } + + async probe(requestId: string): Promise> { + this.send({ type: "probe", requestId }) + return this.waitForEventType("probe-result", requestId) + } + + releaseLock(requestId: string): void { + this.send({ type: "release-lock", requestId }) + } + + waitForEventType( + type: TType, + requestId: string, + ): Promise> { + return this.waitFor( + (event): event is Extract => + event.type === type && "requestId" in event && event.requestId === requestId, + `${type} (${requestId})`, + ) + } + + async close(): Promise { + if (this.exited) return + const requestId = `shutdown-${this.workerId}` + if (this.child.connected) { + this.send({ type: "shutdown", requestId }) + await this.waitForEventType("shutdown-complete", requestId) + } + this.expectedExit = true + await this.waitForExit() + } + + kill(): void { + if (!this.exited) this.child.kill() + } + + private waitFor( + predicate: (event: WorkerToParentMessage) => event is TEvent, + description: string, + ): Promise + private waitFor( + predicate: (event: WorkerToParentMessage) => boolean, + description: string, + ): Promise + private waitFor( + predicate: (event: WorkerToParentMessage) => boolean, + description: string, + ): Promise { + if (this.terminalError) return Promise.reject(this.terminalError) + const existing = this.events.find(predicate) + if (existing) return Promise.resolve(existing) + + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.waiters.delete(waiter) + reject( + new Error(`Timed out waiting for worker ${this.workerId} to ${description}${this.diagnostics()}`), + ) + }, DIAGNOSTIC_TIMEOUT_MS) + const waiter = { predicate, resolve, reject, timer } + this.waiters.add(waiter) + }) + } + + private waitForExit(): Promise { + if (this.exited) return Promise.resolve() + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`Timed out waiting for worker ${this.workerId} to exit${this.diagnostics()}`)) + }, DIAGNOSTIC_TIMEOUT_MS) + this.child.once("exit", () => { + clearTimeout(timer) + resolve() + }) + }) + } + + private handleEvent(value: unknown): void { + if (!isWorkerEvent(value) || value.workerId !== this.workerId) { + this.fail( + new Error(`Worker ${this.workerId} sent malformed IPC: ${JSON.stringify(value)}${this.diagnostics()}`), + ) + return + } + if (value.type === "worker-error") { + this.fail( + new Error( + `Worker ${this.workerId} failed handling ${value.requestType}: ${value.message}\n${value.stack ?? ""}${this.diagnostics()}`, + ), + ) + return + } + + this.events.push(value) + this.onEvent?.(value) + for (const waiter of this.waiters) { + if (waiter.predicate(value)) { + clearTimeout(waiter.timer) + this.waiters.delete(waiter) + waiter.resolve(value) + } + } + } + + private fail(error: Error): void { + if (this.terminalError) return + this.terminalError = error + for (const waiter of this.waiters) { + clearTimeout(waiter.timer) + waiter.reject(error) + } + this.waiters.clear() + } + + private diagnostics(): string { + const eventSummary = this.events.map((event) => event.type).join(", ") + return `\nEvents: [${eventSummary}]\nstdout:\n${this.stdout}\nstderr:\n${this.stderr}` + } +} + +describe("TaskHistoryStore separate-process integration", () => { + let storageRoot: string + let workers: ProcessWorker[] + + beforeEach(async () => { + storageRoot = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-process-")) + workers = [] + }) + + afterEach(async () => { + try { + await Promise.all(workers.map((worker) => worker.close())) + } finally { + workers.forEach((worker) => worker.kill()) + await fs.rm(storageRoot, { recursive: true, force: true }) + } + }) + + it("rebuilds the index from authoritative task files when both process caches are stale and partial", async () => { + const workerA = addWorker(workers, new ProcessWorker("A")) + const workerB = addWorker(workers, new ProcessWorker("B")) + + // Both initialization barriers complete before either task exists. The worker + // disables all background cache/index activity before constructing the store. + await Promise.all([workerA.initialize(storageRoot), workerB.initialize(storageRoot)]) + + const stagedA = await workerA.stage("stage-a", makeHistoryItem("task-a", 1_000)) + const stagedB = await workerB.stage("stage-b", makeHistoryItem("task-b", 2_000)) + expect(stagedA.cacheIds).toEqual(["task-a"]) + expect(stagedB.cacheIds).toEqual(["task-b"]) + + workerA.flush("flush-a") + await workerA.waitForEventType("flush-completed", "flush-a") + expect(await readIndexIds(storageRoot)).toEqual(["task-a", "task-b"]) + + workerB.flush("flush-b") + await workerB.waitForEventType("flush-completed", "flush-b") + + expect(await readTaskFileIds(storageRoot, ["task-a", "task-b"])).toEqual(["task-a", "task-b"]) + expect(await readIndexIds(storageRoot)).toEqual(["task-a", "task-b"]) + }) + + it("serializes real advisory-lock contention across process IDs without blocking the waiting event loop", async () => { + const ordering: string[] = [] + const recordOrdering = (event: WorkerToParentMessage): void => { + if ( + event.type === "lock-acquired" || + (event.workerId === "B" && event.type === "lock-attempted") || + event.type === "lock-callback-completed" + ) { + ordering.push(`${event.workerId}:${event.type}`) + } + } + const workerA = addWorker(workers, new ProcessWorker("A", recordOrdering)) + const workerB = addWorker(workers, new ProcessWorker("B", recordOrdering)) + + await Promise.all([workerA.initialize(storageRoot, true), workerB.initialize(storageRoot)]) + await workerA.stage("stage-a", makeHistoryItem("task-a", 1_000)) + await workerB.stage("stage-b", makeHistoryItem("task-b", 2_000)) + + workerA.flush("flush-a") + const acquiredA = await workerA.waitForEventType("lock-acquired", "flush-a") + await workerA.waitForEventType("lock-paused", "flush-a") + + workerB.flush("flush-b") + const attemptedB = await workerB.waitForEventType("lock-attempted", "flush-b") + expect(acquiredA.pid).not.toBe(attemptedB.pid) + + // This response is positive, event-driven evidence that B handled another IPC + // command while its flush remained unresolved outside the critical callback. + const probeB = await workerB.probe("probe-b-waiting") + expect(probeB).toMatchObject({ flushPending: true, insideLockCallback: false }) + ordering.push("B:not-entered-responsive") + expect(ordering).toEqual(["A:lock-acquired", "B:lock-attempted", "B:not-entered-responsive"]) + + workerA.releaseLock("release-a") + await workerA.waitForEventType("lock-callback-completed", "flush-a") + await workerA.waitForEventType("flush-completed", "flush-a") + await workerB.waitForEventType("lock-acquired", "flush-b") + await workerB.waitForEventType("lock-callback-completed", "flush-b") + await workerB.waitForEventType("flush-completed", "flush-b") + + // IPC order is guaranteed per child channel, but not between A's and B's + // independent channels after the lock is released. Assert each process's + // causal sequence without relying on cross-channel delivery timing. + expect(ordering.filter((entry) => entry.startsWith("A:"))).toEqual([ + "A:lock-acquired", + "A:lock-callback-completed", + ]) + expect(ordering.filter((entry) => entry.startsWith("B:"))).toEqual([ + "B:lock-attempted", + "B:not-entered-responsive", + "B:lock-acquired", + "B:lock-callback-completed", + ]) + expect(await readIndexIds(storageRoot)).toEqual(["task-a", "task-b"]) + }) +}) + +function addWorker(workers: ProcessWorker[], worker: ProcessWorker): ProcessWorker { + workers.push(worker) + return worker +} + +function makeHistoryItem(id: string, ts: number): HistoryItem { + return { + id, + number: ts / 1_000, + ts, + task: `Task ${id}`, + tokensIn: 100, + tokensOut: 50, + totalCost: 0.01, + workspace: path.join("workspace", id), + } +} + +async function readIndexIds(storageRoot: string): Promise { + const indexPath = path.join(storageRoot, "tasks", GlobalFileNames.historyIndex) + const parsed = JSON.parse(await fs.readFile(indexPath, "utf8")) as unknown + if (!isHistoryIndex(parsed)) throw new Error(`Malformed task history index at ${indexPath}`) + return parsed.entries.map((entry) => entry.id).sort() +} + +async function readTaskFileIds(storageRoot: string, taskIds: string[]): Promise { + const ids = await Promise.all( + taskIds.map(async (taskId) => { + const taskPath = path.join(storageRoot, "tasks", taskId, GlobalFileNames.historyItem) + const parsed = JSON.parse(await fs.readFile(taskPath, "utf8")) as unknown + if (!isHistoryItem(parsed)) throw new Error(`Malformed task history item at ${taskPath}`) + return parsed.id + }), + ) + return ids.sort() +} + +function isHistoryIndex(value: unknown): value is HistoryIndex { + return ( + !!value && + typeof value === "object" && + "version" in value && + value.version === 1 && + "entries" in value && + Array.isArray(value.entries) && + value.entries.every(isHistoryItem) + ) +} + +function isHistoryItem(value: unknown): value is HistoryItem { + return !!value && typeof value === "object" && "id" in value && typeof value.id === "string" +} + +function isWorkerEvent(value: unknown): value is WorkerToParentMessage { + return ( + !!value && + typeof value === "object" && + "type" in value && + typeof value.type === "string" && + "workerId" in value && + (value.workerId === "A" || value.workerId === "B") && + "pid" in value && + typeof value.pid === "number" + ) +} + +function appendBounded(current: string, addition: string): string { + const combined = current + addition + if (Buffer.byteLength(combined) <= MAX_CAPTURED_OUTPUT_BYTES) return combined + return `[output truncated to last ${MAX_CAPTURED_OUTPUT_BYTES} bytes]\n${combined.slice(-MAX_CAPTURED_OUTPUT_BYTES)}` +} diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index e788b5d96a..61aac1c8eb 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -836,6 +836,7 @@ describe("TaskHistoryStore migrateFromGlobalState reconciliation", () => { afterEach(async () => { store.dispose() + await store.flushIndex() await fs.rm(tmpDir, { recursive: true, force: true }) }) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts index 3188e9c505..415918330b 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts @@ -430,6 +430,30 @@ describe("TaskHistoryStore", () => { expect(index.entries).toHaveLength(1) expect(index.entries[0].id).toBe("flush-task") }) + + it("rebuilds the index from on-disk task files rather than a stale cache snapshot", async () => { + await store.initialize() + + const cached = makeHistoryItem({ id: "cached-only", task: "in cache", ts: 1000 }) + const onDiskOnly = makeHistoryItem({ id: "disk-only", task: "on disk", ts: 2000 }) + + await store.upsert(cached) + + // Peer process wrote a task file the local cache never saw. + const diskOnlyDir = path.join(tmpDir, "tasks", onDiskOnly.id) + await fs.mkdir(diskOnlyDir, { recursive: true }) + await fs.writeFile(path.join(diskOnlyDir, GlobalFileNames.historyItem), JSON.stringify(onDiskOnly), "utf8") + + // Local cache still only knows about its own upsert. + expect(store.get("disk-only")).toBeUndefined() + + await store.flushIndex() + + const indexPath = path.join(tmpDir, "tasks", GlobalFileNames.historyIndex) + const index = JSON.parse(await fs.readFile(indexPath, "utf8")) as { entries: HistoryItem[] } + const ids = index.entries.map((entry) => entry.id).sort() + expect(ids).toEqual(["cached-only", "disk-only"]) + }) }) describe("dispose()", () => { diff --git a/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessProtocol.ts b/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessProtocol.ts new file mode 100644 index 0000000000..6446cc0c23 --- /dev/null +++ b/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessProtocol.ts @@ -0,0 +1,40 @@ +import type { HistoryItem } from "@roo-code/types" + +export type WorkerId = "A" | "B" + +export type ParentToWorkerMessage = + | { + type: "initialize" + workerId: WorkerId + storageRoot: string + pauseFirstLockCallback: boolean + } + | { type: "stage"; requestId: string; item: HistoryItem } + | { type: "flush"; requestId: string } + | { type: "probe"; requestId: string } + | { type: "release-lock"; requestId: string } + | { type: "shutdown"; requestId: string } + +interface WorkerEventBase { + workerId: WorkerId + pid: number +} + +export type WorkerToParentMessage = + | (WorkerEventBase & { type: "initialized"; cacheIds: string[] }) + | (WorkerEventBase & { type: "staged"; requestId: string; cacheIds: string[] }) + | (WorkerEventBase & { type: "flush-started"; requestId: string }) + | (WorkerEventBase & { type: "lock-attempted"; requestId: string }) + | (WorkerEventBase & { type: "lock-acquired"; requestId: string }) + | (WorkerEventBase & { type: "lock-paused"; requestId: string }) + | (WorkerEventBase & { type: "lock-callback-completed"; requestId: string }) + | (WorkerEventBase & { type: "flush-completed"; requestId: string }) + | (WorkerEventBase & { + type: "probe-result" + requestId: string + flushPending: boolean + insideLockCallback: boolean + }) + | (WorkerEventBase & { type: "lock-released-by-parent"; requestId: string }) + | (WorkerEventBase & { type: "shutdown-complete"; requestId: string }) + | (WorkerEventBase & { type: "worker-error"; requestType: string; message: string; stack?: string }) diff --git a/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessWorker.ts b/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessWorker.ts new file mode 100644 index 0000000000..22e74f594d --- /dev/null +++ b/src/core/task-persistence/__tests__/fixtures/taskHistoryProcessWorker.ts @@ -0,0 +1,211 @@ +import type { HistoryItem } from "@roo-code/types" + +import { TaskHistoryStore } from "../../TaskHistoryStore" +import { taskHistoryLock } from "../../TaskHistoryLock" +import type { ParentToWorkerMessage, WorkerId, WorkerToParentMessage } from "./taskHistoryProcessProtocol" + +let workerId: WorkerId | undefined +let store: TaskHistoryStore | undefined +let pauseFirstLockCallback = false +let releasePausedLock: (() => void) | undefined +let activeFlush: Promise | undefined +let activeFlushRequestId: string | undefined +let flushPending = false +let insideLockCallback = false + +// These tests control every reconciliation and flush through IPC barriers. Patch the +// child-only prototype before constructing/initializing the store so no watcher, +// periodic timer, or debounced write can make either process's cache less stale. +TaskHistoryStore.prototype["startWatcher"] = () => undefined +TaskHistoryStore.prototype["startPeriodicReconciliation"] = () => undefined +TaskHistoryStore.prototype["scheduleIndexWrite"] = () => undefined + +const originalWithLock = taskHistoryLock.withLock.bind(taskHistoryLock) +taskHistoryLock.withLock = async function (globalStoragePath: string, callback: () => Promise): Promise { + const requestId = requireActiveFlushRequestId() + send({ type: "lock-attempted", ...identity(), requestId }) + + return originalWithLock(globalStoragePath, async () => { + insideLockCallback = true + send({ type: "lock-acquired", ...identity(), requestId }) + + try { + if (pauseFirstLockCallback) { + pauseFirstLockCallback = false + send({ type: "lock-paused", ...identity(), requestId }) + await new Promise((resolve) => { + releasePausedLock = resolve + }) + releasePausedLock = undefined + } + + const result = await callback() + send({ type: "lock-callback-completed", ...identity(), requestId }) + return result + } finally { + insideLockCallback = false + } + }) +} + +process.on("message", (value: unknown) => { + void handleMessage(value).catch((error: unknown) => { + const requestType = getMessageType(value) + const normalized = error instanceof Error ? error : new Error(String(error)) + if (workerId) { + send({ + type: "worker-error", + ...identity(), + requestType, + message: normalized.message, + stack: normalized.stack, + }) + } else { + console.error(`[task-history-process-worker] ${requestType}:`, normalized) + process.exitCode = 1 + } + }) +}) + +async function handleMessage(value: unknown): Promise { + const message = parseParentMessage(value) + + switch (message.type) { + case "initialize": { + if (store) throw new Error("Worker was initialized more than once") + workerId = message.workerId + pauseFirstLockCallback = message.pauseFirstLockCallback + store = new TaskHistoryStore(message.storageRoot) + await store.initialize() + send({ type: "initialized", ...identity(), cacheIds: cacheIds(store) }) + return + } + case "stage": { + const activeStore = requireStore() + await activeStore.upsert(message.item) + send({ type: "staged", ...identity(), requestId: message.requestId, cacheIds: cacheIds(activeStore) }) + return + } + case "flush": { + if (activeFlush) throw new Error("A flush is already active") + const activeStore = requireStore() + activeFlushRequestId = message.requestId + flushPending = true + send({ type: "flush-started", ...identity(), requestId: message.requestId }) + activeFlush = activeStore + .flushIndex() + .then(() => { + send({ type: "flush-completed", ...identity(), requestId: message.requestId }) + }) + .finally(() => { + flushPending = false + activeFlushRequestId = undefined + activeFlush = undefined + }) + await activeFlush + return + } + case "probe": { + send({ + type: "probe-result", + ...identity(), + requestId: message.requestId, + flushPending, + insideLockCallback, + }) + return + } + case "release-lock": { + if (!releasePausedLock) throw new Error("No paused lock callback is awaiting release") + releasePausedLock() + send({ type: "lock-released-by-parent", ...identity(), requestId: message.requestId }) + return + } + case "shutdown": { + releasePausedLock?.() + await activeFlush?.catch(() => undefined) + send({ type: "shutdown-complete", ...identity(), requestId: message.requestId }, () => process.disconnect()) + return + } + } +} + +function parseParentMessage(value: unknown): ParentToWorkerMessage { + if (!value || typeof value !== "object" || !("type" in value) || typeof value.type !== "string") { + throw new Error("Received malformed parent IPC message") + } + + const message = value as Record + switch (message.type) { + case "initialize": + if ( + (message.workerId === "A" || message.workerId === "B") && + typeof message.storageRoot === "string" && + typeof message.pauseFirstLockCallback === "boolean" + ) { + return { + type: "initialize", + workerId: message.workerId, + storageRoot: message.storageRoot, + pauseFirstLockCallback: message.pauseFirstLockCallback, + } + } + break + case "stage": + if (typeof message.requestId === "string" && isHistoryItem(message.item)) { + return { type: "stage", requestId: message.requestId, item: message.item } + } + break + case "flush": + case "probe": + case "release-lock": + case "shutdown": + if (typeof message.requestId === "string") { + return { type: message.type, requestId: message.requestId } + } + break + } + + throw new Error(`Received invalid ${String(message.type)} IPC message`) +} + +function isHistoryItem(value: unknown): value is HistoryItem { + return !!value && typeof value === "object" && "id" in value && typeof value.id === "string" +} + +function getMessageType(value: unknown): string { + return value && typeof value === "object" && "type" in value && typeof value.type === "string" + ? value.type + : "unknown" +} + +function requireStore(): TaskHistoryStore { + if (!store) throw new Error("Worker has not been initialized") + return store +} + +function requireActiveFlushRequestId(): string { + if (!activeFlushRequestId) throw new Error("Task-history lock was invoked outside a parent-requested flush") + return activeFlushRequestId +} + +function identity(): { workerId: WorkerId; pid: number } { + if (!workerId) throw new Error("Worker identity is not initialized") + return { workerId, pid: process.pid } +} + +function cacheIds(activeStore: TaskHistoryStore): string[] { + return activeStore + .getAll() + .map((item) => item.id) + .sort() +} + +function send(message: WorkerToParentMessage, callback?: (error: Error | null) => void): void { + if (!process.send) throw new Error("Worker IPC channel is unavailable") + if (callback) { + process.send(message, callback) + } else { + process.send(message) + } +} diff --git a/src/core/task-persistence/__tests__/fixtures/tsconfig.json b/src/core/task-persistence/__tests__/fixtures/tsconfig.json new file mode 100644 index 0000000000..ecd2ec9ed3 --- /dev/null +++ b/src/core/task-persistence/__tests__/fixtures/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "../../../../tsconfig.json", + "compilerOptions": { + "baseUrl": ".", + "paths": { + "vscode": ["../../../../__mocks__/vscode.js"] + } + }, + "include": ["./*.ts"] +} diff --git a/src/shared/globalFileNames.ts b/src/shared/globalFileNames.ts index 7bfe18f4bc..c2214ad53d 100644 --- a/src/shared/globalFileNames.ts +++ b/src/shared/globalFileNames.ts @@ -6,5 +6,7 @@ export const GlobalFileNames = { taskMetadata: "task_metadata.json", historyItem: "history_item.json", historyIndex: "_index.json", + /** Advisory lock file serializing cross-process `_index.json` rebuilds. */ + historyLock: "_history.lock", delegationRepairIntent: "_delegation_repair_intent.json", }