From 5cd2acfe150de0791b97c4b6785b6e3c073f6a7a Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 9 Aug 2026 15:03:45 +0000 Subject: [PATCH 1/5] fix(task): recover stale delegated children after restart --- src/core/task-persistence/TaskHistoryStore.ts | 35 ++++++- .../TaskHistoryStore.reconciliation.spec.ts | 93 +++++++++++++++++-- 2 files changed, 116 insertions(+), 12 deletions(-) diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index c6c3c6910f..16cef6a2f6 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -355,11 +355,23 @@ export class TaskHistoryStore { * - Parent `delegated` with no `awaitingChildId` → parent → `active` (invalid state) * - Parent `delegated`, child not found → parent → `active` (orphaned delegation) * - Parent `delegated`, child `completed` → parent → `active` (interrupted handoff) + * - Parent `delegated`, child `active` → child → `interrupted`, parent → `active` * - * A parent awaiting an `active`, `interrupted`, or `delegated` child is left as-is — the child is resumable. + * A parent awaiting an `interrupted` or `delegated` child is left as-is — the child is + * resumable. An `active` child is treated as orphaned during startup recovery because + * no live task session exists to own it. */ private async reconcileDelegationState(): Promise { return this.withLock(async () => { + // Only statuses loaded from persistence represent sessions that could have + // been orphaned by a crash. A delegated parent repaired to active earlier in + // this pass remains resumable and must not be mistaken for a second orphaned + // child in a delegation chain. + const persistedActiveIds = new Set( + Array.from(this.cache.values()) + .filter((item) => item.status === "active") + .map((item) => item.id), + ) let repairsInThisPass: number do { repairsInThisPass = 0 @@ -400,6 +412,25 @@ export class TaskHistoryStore { `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, ) repairsInThisPass++ + } else if (child.status === "active" && persistedActiveIds.has(child.id)) { + // An active child persisted across startup cannot have a live task session + // behind it. Mark it interrupted before releasing the parent's delegation + // link so the normal resume/re-delegate flow can take over. This is an + // administrative recovery, not a runtime delegation transition. + await this.upsertCore({ ...child, status: "interrupted" }, { skipTransitionCheck: true }) + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, + ) + repairsInThisPass++ } else if (child.status === "completed") { await this.upsertCore( { @@ -418,7 +449,7 @@ export class TaskHistoryStore { ) repairsInThisPass++ } - // child.status === "active", "interrupted", or "delegated" → leave as-is this pass + // child.status === "interrupted" or "delegated" → leave as-is this pass } } while (repairsInThisPass > 0) }) diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index 2888c0f7b2..1f797f4d02 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -185,16 +185,59 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(repaired?.completionResultSummary).toBe("Task completed (recovered after interruption)") }) - it("leaves delegated parent alone when child is still active", async () => { - const child = makeItem({ id: "child-4", status: "active" }) - const parent = makeItem({ id: "parent-4", status: "delegated", awaitingChildId: "child-4" }) + it("repairs a delegated parent with an active orphaned child", async () => { + const child = makeItem({ + id: "child-4", + status: "active", + parentTaskId: "parent-4", + rootTaskId: "parent-4", + childIds: ["grandchild-4"], + }) + const parent = makeItem({ + id: "parent-4", + status: "delegated", + awaitingChildId: "child-4", + delegatedToId: "child-4", + childIds: ["child-4"], + }) await seedItems([parent, child]) await store.initialize() - const unchanged = store.get("parent-4") - expect(unchanged?.status).toBe("delegated") - expect(unchanged?.awaitingChildId).toBe("child-4") + const repairedParent = store.get("parent-4") + const repairedChild = store.get("child-4") + expect(repairedChild).toMatchObject({ + id: "child-4", + status: "interrupted", + parentTaskId: "parent-4", + rootTaskId: "parent-4", + childIds: ["grandchild-4"], + }) + expect(repairedParent).toMatchObject({ + id: "parent-4", + status: "active", + childIds: ["child-4"], + }) + expect(repairedParent?.awaitingChildId).toBeUndefined() + expect(repairedParent?.delegatedToId).toBeUndefined() + + const tasksDir = path.join(tmpDir, "tasks") + const persistedChild = JSON.parse( + await fs.readFile(path.join(tasksDir, "child-4", "history_item.json"), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tasksDir, "parent-4", "history_item.json"), "utf8"), + ) as HistoryItem + expect(persistedChild).toMatchObject({ + id: "child-4", + status: "interrupted", + parentTaskId: "parent-4", + rootTaskId: "parent-4", + childIds: ["grandchild-4"], + }) + expect(persistedParent).toMatchObject({ id: "parent-4", status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() }) it("repairs invalid delegation: delegated parent with no awaitingChildId → active (clears delegatedToId and awaitingChildId)", async () => { @@ -239,9 +282,9 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(store.get("parent-b")?.status).toBe("active") }) - it("handles chained delegation (A→B→C): repairs B first, then A sees B as active and is left delegated", async () => { + it("handles chained delegation (A→B→C) until all orphaned links converge", async () => { // C doesn't exist (orphaned). B is delegated waiting for C → repaired to active. - // A is delegated waiting for B → left delegated (B is now active, resumable by user). + // A then sees B as an orphaned active child and is repaired as well. const parentA = makeItem({ id: "parent-a-chain", status: "delegated", awaitingChildId: "parent-b-chain" }) const parentB = makeItem({ id: "parent-b-chain", @@ -254,9 +297,39 @@ describe("TaskHistoryStore reconcileDelegationState", () => { // B is repaired: its child (C) was missing expect(store.get("parent-b-chain")?.status).toBe("active") - // A stays delegated: its child (B) is now active, which is a valid state + // A stays delegated: B was repaired from delegated to active and remains + // resumable rather than being mistaken for an active orphan from disk. expect(store.get("parent-a-chain")?.status).toBe("delegated") expect(store.get("parent-a-chain")?.awaitingChildId).toBe("parent-b-chain") + expect(store.get("parent-b-chain")?.status).toBe("active") + expect(store.get("parent-b-chain")?.awaitingChildId).toBeUndefined() + }) + + it("is idempotent when recovering an active child", async () => { + const child = makeItem({ id: "child-active-idempotent", status: "active" }) + const parent = makeItem({ + id: "parent-active-idempotent", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + + await store.initialize() + const afterFirstParent = { ...store.get(parent.id) } + const afterFirstChild = { ...store.get(child.id) } + + store.dispose() + const store2 = new TaskHistoryStore(tmpDir) + await store2.initialize() + const afterSecondParent = { ...store2.get(parent.id) } + const afterSecondChild = { ...store2.get(child.id) } + store2.dispose() + + expect(afterFirstParent).toMatchObject({ status: "active" }) + expect(afterSecondParent).toEqual(afterFirstParent) + expect(afterFirstChild).toMatchObject({ status: "interrupted" }) + expect(afterSecondChild).toEqual(afterFirstChild) }) it("is idempotent: running initialize twice produces the same result", async () => { @@ -407,7 +480,7 @@ describe("TaskHistoryStore upsert transition guard", () => { it("rejects delegated → completed transition", async () => { // Must include a live active child so reconciliation doesn't repair the parent to active - const child = makeItem({ id: "child-guard-2", status: "active" }) + const child = makeItem({ id: "child-guard-2", status: "interrupted" }) const item = makeItem({ id: "task-guard-2", status: "delegated", awaitingChildId: "child-guard-2" }) await seedItems([child, item]) store.dispose() From b676bd3eecb902027fb498818fda80f918ef726e Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 9 Aug 2026 22:29:53 +0000 Subject: [PATCH 2/5] fix(task): make delegated recovery durable --- src/core/task-persistence/TaskHistoryStore.ts | 307 ++++++++++++++++-- .../TaskHistoryStore.reconciliation.spec.ts | 193 ++++++++++- src/shared/globalFileNames.ts | 1 + .../__tests__/providerModelConfig.spec.ts | 6 +- .../settings/utils/providerModelConfig.ts | 8 +- 5 files changed, 474 insertions(+), 41 deletions(-) diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 16cef6a2f6..01ec41e5d4 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -40,6 +40,34 @@ interface HistoryIndex { 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 + * guarded target transition that must be completed after a crash. + */ +interface DelegationRepairIntent { + version: 1 + operationId: string + parentTaskId: string + childTaskId: string + expected: { + parent: { + status: "delegated" + awaitingChildId: string + delegatedToId?: string + } + child: { + status: "active" + parentTaskId?: string + rootTaskId?: string + } + } + target: { + childStatus: "interrupted" + parentStatus: "active" + } +} + /** * TaskHistoryStore encapsulates all task history persistence logic. * @@ -111,13 +139,16 @@ export class TaskHistoryStore { // 2. Reconcile cache against actual task directories on disk await this.reconcile() - // 3. Repair delegation inconsistencies left by a previous crash + // 3. Complete any two-record repair interrupted after its intent was durable. + await this.replayDelegationRepairIntent() + + // 4. Repair delegation inconsistencies left by a previous crash await this.reconcileDelegationState() - // 4. Start fs.watch for cross-instance reactivity + // 5. Start fs.watch for cross-instance reactivity this.startWatcher() - // 5. Start periodic reconciliation as a defensive fallback + // 6. Start periodic reconciliation as a defensive fallback this.startPeriodicReconciliation() } finally { // Mark initialization as complete so callers awaiting `initialized` can proceed @@ -309,18 +340,17 @@ export class TaskHistoryStore { const cacheIds = new Set(this.cache.keys()) let changed = false - // Tasks on disk but not in cache: read their history_item.json + // Task files are authoritative. Always refresh entries from disk so a stale + // index cannot overwrite a repair or another instance's newer task state. for (const taskId of onDiskIds) { - if (!cacheIds.has(taskId)) { - try { - const item = await this.readTaskFile(taskId) - if (item) { - this.cache.set(taskId, item) - changed = true - } - } catch { - // Corrupted or missing file, skip + try { + const item = await this.readTaskFile(taskId) + if (item) { + this.cache.set(taskId, item) + changed = true } + } catch { + // Corrupted or missing file, skip } } @@ -369,7 +399,7 @@ export class TaskHistoryStore { // child in a delegation chain. const persistedActiveIds = new Set( Array.from(this.cache.values()) - .filter((item) => item.status === "active") + .filter((item) => (item.status ?? "active") === "active") .map((item) => item.id), ) let repairsInThisPass: number @@ -412,21 +442,12 @@ export class TaskHistoryStore { `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, ) repairsInThisPass++ - } else if (child.status === "active" && persistedActiveIds.has(child.id)) { + } else if ((child.status ?? "active") === "active" && persistedActiveIds.has(child.id)) { // An active child persisted across startup cannot have a live task session // behind it. Mark it interrupted before releasing the parent's delegation // link so the normal resume/re-delegate flow can take over. This is an // administrative recovery, not a runtime delegation transition. - await this.upsertCore({ ...child, status: "interrupted" }, { skipTransitionCheck: true }) - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - }, - { skipTransitionCheck: true }, - ) + await this.repairActiveDelegation(item, child) console.warn( `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, ) @@ -455,6 +476,242 @@ export class TaskHistoryStore { }) } + /** + * Replay the durable active-child repair intent, if one was left by a crash. + * The expected fields are guards: an intent may update only the missing side + * when the other side is already at its target, or when both records still + * describe the original delegated handoff. + */ + private async replayDelegationRepairIntent(): Promise { + return this.withLock(async () => { + const intent = await this.readDelegationRepairIntent() + if (!intent) { + return + } + + const child = this.cache.get(intent.childTaskId) + const parent = this.cache.get(intent.parentTaskId) + if (!child || !parent) { + await this.quarantineDelegationRepairIntent( + intent, + `missing ${!child ? "child" : "parent"} task record`, + ) + return + } + + const childAtTarget = child.status === intent.target.childStatus + const parentAtTarget = + parent.status === intent.target.parentStatus && + parent.awaitingChildId === undefined && + parent.delegatedToId === undefined + const childMatchesExpected = this.matchesDelegationRepairChildPreconditions(intent, child) + const parentMatchesExpected = this.matchesDelegationRepairParentPreconditions(intent, parent) + + if ((!childAtTarget && !childMatchesExpected) || (!parentAtTarget && !parentMatchesExpected)) { + await this.quarantineDelegationRepairIntent(intent, "task state no longer matches its guards") + return + } + + const repairedChild = childAtTarget ? child : { ...child, status: intent.target.childStatus } + const repairedParent = parentAtTarget + ? parent + : { + ...parent, + status: intent.target.parentStatus, + awaitingChildId: undefined, + delegatedToId: undefined, + } + + if (!childAtTarget) await this.writeTaskFile(repairedChild) + if (!parentAtTarget) await this.writeTaskFile(repairedParent) + + this.cache.set(repairedChild.id, repairedChild) + this.cache.set(repairedParent.id, repairedParent) + this.scheduleIndexWrite() + + // The journal is retained until the write-through callback succeeds. + if (this.onWrite) { + await this.onWrite(this.getAll()) + } + await this.removeDelegationRepairIntent() + }) + } + + /** + * Start and complete a guarded active-child repair while already holding the + * store lock. The intent is durable before either task file is touched. + */ + private async repairActiveDelegation(parent: HistoryItem, child: HistoryItem): Promise { + const intent: DelegationRepairIntent = { + version: 1, + operationId: `delegation-repair-${Date.now()}-${Math.random().toString(36).slice(2)}`, + parentTaskId: parent.id, + childTaskId: child.id, + expected: { + parent: { + status: "delegated", + awaitingChildId: child.id, + delegatedToId: parent.delegatedToId, + }, + child: { + status: "active", + parentTaskId: child.parentTaskId, + rootTaskId: child.rootTaskId, + }, + }, + target: { childStatus: "interrupted", parentStatus: "active" }, + } + + await this.writeDelegationRepairIntent(intent) + await this.applyDelegationRepairIntent(intent, child, parent) + } + + private async applyDelegationRepairIntent( + intent: DelegationRepairIntent, + child: HistoryItem, + parent: HistoryItem, + ): Promise { + const repairedChild = { ...child, status: intent.target.childStatus } + const repairedParent = { + ...parent, + status: intent.target.parentStatus, + awaitingChildId: undefined, + delegatedToId: undefined, + } + + await this.writeTaskFile(repairedChild) + await this.writeTaskFile(repairedParent) + + this.cache.set(repairedChild.id, repairedChild) + this.cache.set(repairedParent.id, repairedParent) + this.scheduleIndexWrite() + + if (this.onWrite) { + await this.onWrite(this.getAll()) + } + await this.removeDelegationRepairIntent() + } + + private matchesDelegationRepairParentPreconditions(intent: DelegationRepairIntent, parent: HistoryItem): boolean { + return ( + parent.status === intent.expected.parent.status && + parent.awaitingChildId === intent.expected.parent.awaitingChildId && + parent.delegatedToId === intent.expected.parent.delegatedToId + ) + } + + private matchesDelegationRepairChildPreconditions(intent: DelegationRepairIntent, child: HistoryItem): boolean { + return ( + (child.status ?? "active") === intent.expected.child.status && + child.parentTaskId === intent.expected.child.parentTaskId && + child.rootTaskId === intent.expected.child.rootTaskId + ) + } + + private async readDelegationRepairIntent(): Promise { + const intentPath = await this.getDelegationRepairIntentPath() + let parsed: unknown + try { + parsed = JSON.parse(await fs.readFile(intentPath, "utf8")) as unknown + } catch { + try { + await fs.access(intentPath) + } catch { + return null + } + await this.quarantineDelegationRepairIntent(null, "malformed JSON") + return null + } + + if (!this.isDelegationRepairIntent(parsed)) { + await this.quarantineDelegationRepairIntent(null, "malformed intent") + return null + } + return parsed + } + + private isDelegationRepairIntent(value: unknown): value is DelegationRepairIntent { + if (!value || typeof value !== "object") { + return false + } + const candidate = value as Record + const expected = candidate.expected + const expectedRecord = expected && typeof expected === "object" ? (expected as Record) : null + const expectedParent = + expectedRecord?.parent && typeof expectedRecord.parent === "object" + ? (expectedRecord.parent as Record) + : null + const expectedChild = + expectedRecord?.child && typeof expectedRecord.child === "object" + ? (expectedRecord.child as Record) + : null + const target = candidate.target + return ( + candidate.version === 1 && + typeof candidate.operationId === "string" && + this.isSafeTaskId(candidate.parentTaskId) && + this.isSafeTaskId(candidate.childTaskId) && + candidate.parentTaskId !== candidate.childTaskId && + !!expectedParent && + expectedParent.status === "delegated" && + typeof expectedParent.awaitingChildId === "string" && + expectedParent.awaitingChildId === candidate.childTaskId && + (expectedParent.delegatedToId === undefined || typeof expectedParent.delegatedToId === "string") && + !!expectedChild && + expectedChild.status === "active" && + (expectedChild.parentTaskId === undefined || typeof expectedChild.parentTaskId === "string") && + (expectedChild.rootTaskId === undefined || typeof expectedChild.rootTaskId === "string") && + !!target && + typeof target === "object" && + (target as Record).childStatus === "interrupted" && + (target as Record).parentStatus === "active" + ) + } + + private async writeDelegationRepairIntent(intent: DelegationRepairIntent): Promise { + await safeWriteJson(await this.getDelegationRepairIntentPath(), intent) + } + + private async removeDelegationRepairIntent(): Promise { + try { + await fs.unlink(await this.getDelegationRepairIntentPath()) + } catch (error) { + console.warn("[TaskHistoryStore] Failed to remove completed delegation repair intent:", error) + } + } + + private async quarantineDelegationRepairIntent( + intent: DelegationRepairIntent | null, + reason: string, + ): Promise { + const intentPath = await this.getDelegationRepairIntentPath() + const quarantinePath = `${intentPath}.quarantine-${Date.now()}-${Math.random().toString(36).slice(2)}` + try { + await fs.rename(intentPath, quarantinePath) + } catch (error) { + console.warn("[TaskHistoryStore] Failed to quarantine delegation repair intent:", error) + } + console.warn( + `[TaskHistoryStore] Ignored ${intent ? `stale delegation repair intent ${intent.operationId}` : "malformed delegation repair intent"}: ${reason}`, + ) + } + + private isSafeTaskId(value: unknown): value is string { + return ( + typeof value === "string" && + value.length > 0 && + value !== "." && + value !== ".." && + !value.includes("/") && + !value.includes("\\") + ) + } + + private async getDelegationRepairIntentPath(): Promise { + const tasksDir = await this.getTasksDir() + return path.join(tasksDir, GlobalFileNames.delegationRepairIntent) + } + // ────────────────────────────── Cache invalidation ────────────────────────────── /** diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index 1f797f4d02..d08ef999f9 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -6,18 +6,23 @@ import * as os from "os" import type { HistoryItem } from "@roo-code/types" +import { GlobalFileNames } from "../../../shared/globalFileNames" import { TaskHistoryStore, assertValidTransition } from "../TaskHistoryStore" vi.mock("../../../utils/storage", () => ({ getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath), })) -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") - }), -})) +const writeJson = async (filePath: string, data: unknown): Promise => { + await fs.mkdir(path.dirname(filePath), { recursive: true }) + await fs.writeFile(filePath, JSON.stringify(data, null, "\t"), "utf8") +} + +const safeWriteJsonMock = vi.hoisted(() => vi.fn()) + +vi.mock("../../../utils/safeWriteJson", () => ({ safeWriteJson: safeWriteJsonMock })) + +safeWriteJsonMock.mockImplementation(writeJson) function makeItem(overrides: Partial = {}): HistoryItem { return { @@ -32,6 +37,28 @@ function makeItem(overrides: Partial = {}): HistoryItem { } } +function makeRepairIntent(parent: HistoryItem, child: HistoryItem): object { + return { + version: 1, + operationId: "delegation-repair-test", + parentTaskId: parent.id, + childTaskId: child.id, + expected: { + parent: { + status: "delegated", + awaitingChildId: child.id, + delegatedToId: parent.delegatedToId, + }, + child: { + status: "active", + parentTaskId: child.parentTaskId, + rootTaskId: child.rootTaskId, + }, + }, + target: { childStatus: "interrupted", parentStatus: "active" }, + } +} + // ───────────────────────────────────────────────────────────────────────────── // assertValidTransition — pure function tests // ───────────────────────────────────────────────────────────────────────────── @@ -134,6 +161,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { }) afterEach(async () => { + safeWriteJsonMock.mockImplementation(writeJson) store.dispose() await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}) }) @@ -240,6 +268,153 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(persistedParent.delegatedToId).toBeUndefined() }) + it("repairs a delegated child with an omitted status as implicit active", async () => { + const child = makeItem({ + id: "child-implicit-active", + parentTaskId: "parent-implicit-active", + rootTaskId: "parent-implicit-active", + }) + const parent = makeItem({ + id: "parent-implicit-active", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + + await store.initialize() + + expect(store.get(child.id)).toMatchObject({ id: child.id, status: "interrupted" }) + expect(store.get(parent.id)).toMatchObject({ id: parent.id, status: "active" }) + expect(store.get(parent.id)?.awaitingChildId).toBeUndefined() + expect(store.get(parent.id)?.delegatedToId).toBeUndefined() + + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedChild.status).toBe("interrupted") + }) + + it("replays an intent after a child-only write and removes it after completion", async () => { + const child = makeItem({ id: "child-replay", status: "active", parentTaskId: "parent-replay" }) + const parent = makeItem({ + id: "parent-replay", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + await fs.writeFile(intentPath, JSON.stringify(makeRepairIntent(parent, child))) + await fs.writeFile( + path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), + JSON.stringify({ ...child, status: "interrupted" }), + ) + + await store.initialize() + + expect(store.get(child.id)?.status).toBe("interrupted") + expect(store.get(parent.id)?.status).toBe("active") + await expect(fs.access(intentPath)).rejects.toThrow() + }) + + it("replays an intent after a failure before the child write", async () => { + const child = makeItem({ + id: "child-fault-before-child", + status: "active", + parentTaskId: "parent-fault-before-child", + }) + const parent = makeItem({ + id: "parent-fault-before-child", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + safeWriteJsonMock.mockImplementation(async (filePath, data) => { + if (filePath.includes(child.id) && filePath.endsWith(GlobalFileNames.historyItem)) + throw new Error("fault before child write") + await writeJson(filePath, data) + }) + await expect(store.initialize()).rejects.toThrow("fault before child write") + store.dispose() + safeWriteJsonMock.mockImplementation(writeJson) + const replayedStore = new TaskHistoryStore(tmpDir) + await replayedStore.initialize() + expect(replayedStore.get(child.id)?.status).toBe("interrupted") + expect(replayedStore.get(parent.id)?.status).toBe("active") + replayedStore.dispose() + }) + + it("replays an intent after a failure before the parent write", async () => { + const child = makeItem({ + id: "child-fault-before-parent", + status: "active", + parentTaskId: "parent-fault-before-parent", + }) + const parent = makeItem({ + id: "parent-fault-before-parent", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + safeWriteJsonMock.mockImplementation(async (filePath, data) => { + if (filePath.includes(parent.id) && filePath.endsWith(GlobalFileNames.historyItem)) + throw new Error("fault before parent write") + await writeJson(filePath, data) + }) + await expect(store.initialize()).rejects.toThrow("fault before parent write") + store.dispose() + safeWriteJsonMock.mockImplementation(writeJson) + const replayedStore = new TaskHistoryStore(tmpDir) + await replayedStore.initialize() + expect(replayedStore.get(child.id)?.status).toBe("interrupted") + expect(replayedStore.get(parent.id)?.status).toBe("active") + replayedStore.dispose() + }) + + it("retains an intent when the callback fails after both writes", async () => { + const child = makeItem({ id: "child-fault-cleanup", status: "active", parentTaskId: "parent-fault-cleanup" }) + const parent = makeItem({ + id: "parent-fault-cleanup", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + store.dispose() + store = new TaskHistoryStore(tmpDir, { onWrite: vi.fn().mockRejectedValue(new Error("fault before cleanup")) }) + await expect(store.initialize()).rejects.toThrow("fault before cleanup") + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + expect(await fs.readFile(intentPath, "utf8")).toContain(child.id) + store.dispose() + safeWriteJsonMock.mockImplementation(writeJson) + const replayedStore = new TaskHistoryStore(tmpDir) + await replayedStore.initialize() + expect(replayedStore.get(child.id)?.status).toBe("interrupted") + expect(replayedStore.get(parent.id)?.status).toBe("active") + await expect(fs.access(intentPath)).rejects.toThrow() + replayedStore.dispose() + }) + + it("quarantines malformed and stale intents without blocking unrelated startup", async () => { + const unrelated = makeItem({ id: "unrelated-startup", status: "active" }) + await seedItems([unrelated]) + const tasksDir = path.join(tmpDir, "tasks") + const intentPath = path.join(tasksDir, GlobalFileNames.delegationRepairIntent) + await fs.writeFile(intentPath, JSON.stringify({ malformed: true })) + + await store.initialize() + + expect(store.get(unrelated.id)?.status).toBe("active") + expect( + (await fs.readdir(tasksDir)).some((name) => + name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), + ), + ).toBe(true) + }) + it("repairs invalid delegation: delegated parent with no awaitingChildId → active (clears delegatedToId and awaitingChildId)", async () => { // awaitingChildId is falsy but explicitly set (empty string), delegatedToId is stale const parent = makeItem({ @@ -247,7 +422,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { status: "delegated", delegatedToId: "stale-child", awaitingChildId: "", - } as any) + }) await seedItems([parent]) await store.initialize() @@ -543,8 +718,8 @@ describe("TaskHistoryStore upsert transition guard", () => { // Legacy items pre-dating the status field have status: undefined, which normalizes // to "active". Writing status: "active" must not throw as an invalid self-loop. const item = makeItem({ id: "task-guard-legacy" }) - delete (item as any).status - await seedItems([item]) + const { status: _status, ...legacyItem } = item + await seedItems([legacyItem]) store.dispose() store = new TaskHistoryStore(tmpDir) await store.initialize() diff --git a/src/shared/globalFileNames.ts b/src/shared/globalFileNames.ts index 0b54ff6809..7bfe18f4bc 100644 --- a/src/shared/globalFileNames.ts +++ b/src/shared/globalFileNames.ts @@ -6,4 +6,5 @@ export const GlobalFileNames = { taskMetadata: "task_metadata.json", historyItem: "history_item.json", historyIndex: "_index.json", + delegationRepairIntent: "_delegation_repair_intent.json", } diff --git a/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts b/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts index 153f30b0e2..3eb8dc08b8 100644 --- a/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts +++ b/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts @@ -64,7 +64,7 @@ describe("providerModelConfig", () => { }) it("returns fallback config for unknown provider", () => { - const config = getProviderServiceConfig("unknown-provider" as any) + const config = getProviderServiceConfig("unknown-provider") expect(config.serviceName).toBe("unknown-provider") expect(config.serviceUrl).toBe("") }) @@ -88,7 +88,7 @@ describe("providerModelConfig", () => { }) it("returns empty string for unknown provider", () => { - const defaultId = getDefaultModelIdForProvider("unknown" as any) + const defaultId = getDefaultModelIdForProvider("unknown") expect(defaultId).toBe("") }) @@ -152,7 +152,7 @@ describe("providerModelConfig", () => { }) it("returns undefined for a provider with no model config entry", () => { - expect(getProviderModelConfig("unknown-provider" as any)).toBeUndefined() + expect(getProviderModelConfig("unknown-provider")).toBeUndefined() }) it("returns the static field config for a non-zai provider", () => { diff --git a/webview-ui/src/components/settings/utils/providerModelConfig.ts b/webview-ui/src/components/settings/utils/providerModelConfig.ts index eccbf7ba1d..4396ddc68b 100644 --- a/webview-ui/src/components/settings/utils/providerModelConfig.ts +++ b/webview-ui/src/components/settings/utils/providerModelConfig.ts @@ -90,11 +90,11 @@ export const PROVIDER_DEFAULT_MODEL_IDS: Partial> = [providerIdentifiers.baseten]: basetenDefaultModelId, } -export const getProviderServiceConfig = (provider: ProviderName): ProviderServiceConfig => { - return PROVIDER_SERVICE_CONFIG[provider] ?? { serviceName: provider, serviceUrl: "" } +export const getProviderServiceConfig = (provider: string): ProviderServiceConfig => { + return PROVIDER_SERVICE_CONFIG[provider as ProviderName] ?? { serviceName: provider, serviceUrl: "" } } -export const getDefaultModelIdForProvider = (provider: ProviderName, apiConfiguration?: ProviderSettings): string => { +export const getDefaultModelIdForProvider = (provider: string, apiConfiguration?: ProviderSettings): string => { // Handle Z.ai's China/International entrypoint distinction if (provider === providerIdentifiers.zai && apiConfiguration) { return apiConfiguration.zaiApiLine === "china_coding" @@ -102,7 +102,7 @@ export const getDefaultModelIdForProvider = (provider: ProviderName, apiConfigur : internationalZAiDefaultModelId } - return PROVIDER_DEFAULT_MODEL_IDS[provider] ?? "" + return PROVIDER_DEFAULT_MODEL_IDS[provider as ProviderName] ?? "" } export type ProviderModelConfig = { From f29b3c189579541ed71f89b6e997dbbf7a7f487b Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Sun, 9 Aug 2026 22:30:32 +0000 Subject: [PATCH 3/5] test(e2e): verify persistence across VS Code restart --- .../fixtures/restart-persistence.json | 18 +++ apps/vscode-e2e/src/restart/phaseProtocol.ts | 107 ++++++++++++++++ .../src/restart/scenarioWorkspace.ts | 58 +++++++++ .../src/restart/vscodeCoordinator.ts | 90 +++++++++++++ apps/vscode-e2e/src/runTest.ts | 58 +++++++-- apps/vscode-e2e/src/suite/index.ts | 2 +- .../src/suite/restart-persistence.test.ts | 121 ++++++++++++++++++ src/eslint-suppressions.json | 5 - 8 files changed, 442 insertions(+), 17 deletions(-) create mode 100644 apps/vscode-e2e/fixtures/restart-persistence.json create mode 100644 apps/vscode-e2e/src/restart/phaseProtocol.ts create mode 100644 apps/vscode-e2e/src/restart/scenarioWorkspace.ts create mode 100644 apps/vscode-e2e/src/restart/vscodeCoordinator.ts create mode 100644 apps/vscode-e2e/src/suite/restart-persistence.test.ts diff --git a/apps/vscode-e2e/fixtures/restart-persistence.json b/apps/vscode-e2e/fixtures/restart-persistence.json new file mode 100644 index 0000000000..240b91da20 --- /dev/null +++ b/apps/vscode-e2e/fixtures/restart-persistence.json @@ -0,0 +1,18 @@ +{ + "fixtures": [ + { + "match": { + "userMessage": "RESTART_PERSISTENCE_SMOKE" + }, + "response": { + "toolCalls": [ + { + "name": "attempt_completion", + "arguments": "{\"result\":\"RESTART_PERSISTENCE_MARKER\"}", + "id": "call_restart_persistence_done" + } + ] + } + } + ] +} diff --git a/apps/vscode-e2e/src/restart/phaseProtocol.ts b/apps/vscode-e2e/src/restart/phaseProtocol.ts new file mode 100644 index 0000000000..06ddc5f508 --- /dev/null +++ b/apps/vscode-e2e/src/restart/phaseProtocol.ts @@ -0,0 +1,107 @@ +import * as path from "path" +import * as fs from "fs/promises" + +import { createWriteStream } from "fs" + +export const PHASE_RESULT_VERSION = 1 as const + +export type RestartPhase = "create" | "verify" +export type PhaseStatus = "passed" | "failed" + +export type PhaseError = { + message: string + stack?: string +} + +export type PhaseResult = { + version: typeof PHASE_RESULT_VERSION + phase: RestartPhase + status: PhaseStatus + values?: Record + error?: PhaseError +} + +const phaseResultNames: Record = { + create: "01-create.json", + verify: "02-verify.json", +} + +export function getPhaseResultPath(resultsDir: string, phase: RestartPhase): string { + const resolvedResultsDir = path.resolve(resultsDir) + const resultPath = path.resolve(resolvedResultsDir, phaseResultNames[phase]) + if (path.dirname(resultPath) !== resolvedResultsDir) { + throw new Error(`Phase result path escaped the results directory: ${phase}`) + } + return resultPath +} + +export function serializePhaseError(error: unknown): PhaseError { + if (error instanceof Error) { + return { + message: error.message.slice(0, 2_000), + ...(error.stack && { stack: error.stack.slice(0, 4_000) }), + } + } + + return { message: String(error).slice(0, 2_000) } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null +} + +export function validatePhaseResult(value: unknown): asserts value is PhaseResult { + if (!isRecord(value) || value.version !== PHASE_RESULT_VERSION) { + throw new Error("Invalid phase result version") + } + if (value.phase !== "create" && value.phase !== "verify") { + throw new Error("Invalid phase result phase") + } + if (value.status !== "passed" && value.status !== "failed") { + throw new Error("Invalid phase result status") + } + if (value.values !== undefined) { + if (!isRecord(value.values) || Object.values(value.values).some((entry) => typeof entry !== "string")) { + throw new Error("Phase result values must be string-valued") + } + } + if (value.error !== undefined) { + if (!isRecord(value.error) || typeof value.error.message !== "string") { + throw new Error("Invalid phase result error") + } + if (value.error.stack !== undefined && typeof value.error.stack !== "string") { + throw new Error("Invalid phase result error stack") + } + } +} + +export async function writePhaseResult(resultsDir: string, result: PhaseResult): Promise { + validatePhaseResult(result) + const targetPath = getPhaseResultPath(resultsDir, result.phase) + const temporaryPath = `${targetPath}.${process.pid}.${Date.now()}.tmp` + try { + await writeJsonAtomically(temporaryPath, result) + await fs.rename(temporaryPath, targetPath) + } finally { + await fs.rm(temporaryPath, { force: true }) + } +} + +async function writeJsonAtomically(filePath: string, value: PhaseResult): Promise { + await fs.mkdir(path.dirname(filePath), { recursive: true }) + await new Promise((resolve, reject) => { + const stream = createWriteStream(filePath, { encoding: "utf8" }) + stream.once("error", reject) + stream.once("finish", resolve) + stream.end(JSON.stringify(value)) + }) +} + +export async function readPhaseResult(resultsDir: string, phase: RestartPhase): Promise { + const result = JSON.parse(await fs.readFile(getPhaseResultPath(resultsDir, phase), "utf8")) as unknown + validatePhaseResult(result) + if (result.phase !== phase) { + throw new Error(`Phase result does not match requested phase: ${phase}`) + } + return result +} diff --git a/apps/vscode-e2e/src/restart/scenarioWorkspace.ts b/apps/vscode-e2e/src/restart/scenarioWorkspace.ts new file mode 100644 index 0000000000..9ce49deaf0 --- /dev/null +++ b/apps/vscode-e2e/src/restart/scenarioWorkspace.ts @@ -0,0 +1,58 @@ +import * as os from "os" +import * as path from "path" +import * as fs from "fs/promises" + +const SCENARIO_ROOT_PREFIX = "roo-vscode-e2e-restart-" + +export type ScenarioWorkspace = { + root: string + workspace: string + userData: string + extensions: string + results: string +} + +function childPath(root: string, name: string): string { + const resolvedRoot = path.resolve(root) + const resolvedChild = path.resolve(resolvedRoot, name) + if (path.dirname(resolvedChild) !== resolvedRoot) { + throw new Error(`Scenario path escaped its root: ${name}`) + } + return resolvedChild +} + +export async function createScenarioWorkspace(): Promise { + const root = await fs.mkdtemp(path.join(os.tmpdir(), SCENARIO_ROOT_PREFIX)) + const scenarioWorkspace: ScenarioWorkspace = { + root, + workspace: childPath(root, "workspace"), + userData: childPath(root, "user-data"), + extensions: childPath(root, "extensions"), + results: childPath(root, "results"), + } + + await Promise.all( + [ + scenarioPath(scenarioWorkspace, "workspace"), + scenarioPath(scenarioWorkspace, "userData"), + scenarioPath(scenarioWorkspace, "extensions"), + scenarioPath(scenarioWorkspace, "results"), + ].map((directory) => fs.mkdir(directory, { recursive: true })), + ) + + return scenarioWorkspace +} + +function scenarioPath(scenarioWorkspace: ScenarioWorkspace, key: "workspace" | "userData" | "extensions" | "results") { + return scenarioWorkspace[key] +} + +export async function removeScenarioWorkspace(scenarioWorkspace: ScenarioWorkspace): Promise { + const root = path.resolve(scenarioWorkspace.root) + const tempRoot = path.resolve(os.tmpdir()) + if (path.dirname(root) !== tempRoot || !path.basename(root).startsWith(SCENARIO_ROOT_PREFIX)) { + throw new Error(`Refusing to remove an unowned scenario root: ${scenarioWorkspace.root}`) + } + + await fs.rm(root, { recursive: true, force: true }) +} diff --git a/apps/vscode-e2e/src/restart/vscodeCoordinator.ts b/apps/vscode-e2e/src/restart/vscodeCoordinator.ts new file mode 100644 index 0000000000..9400aaaaf4 --- /dev/null +++ b/apps/vscode-e2e/src/restart/vscodeCoordinator.ts @@ -0,0 +1,90 @@ +import { spawn } from "child_process" +import * as path from "path" + +import { readPhaseResult, type RestartPhase, type PhaseResult } from "./phaseProtocol" +import type { ScenarioWorkspace } from "./scenarioWorkspace" + +export type RestartCoordinatorOptions = { + vscodeExecutablePath: string + extensionDevelopmentPath: string + extensionTestsPath: string + scenario: string + workspace: ScenarioWorkspace + environment: NodeJS.ProcessEnv + expectedExitPolicies: readonly ExpectedExitPolicy[] +} + +export type ExpectedExitPolicy = { + phase: RestartPhase + termination: "graceful-quit" + code: number + signal: NodeJS.Signals | null +} + +function phaseEnvironment(options: RestartCoordinatorOptions, phase: RestartPhase): NodeJS.ProcessEnv { + return { + ...options.environment, + E2E_PHASE: phase, + E2E_SCENARIO: options.scenario, + E2E_RESULTS_DIR: options.workspace.results, + } +} + +function phaseArguments(options: RestartCoordinatorOptions): string[] { + return [ + options.workspace.workspace, + `--user-data-dir=${options.workspace.userData}`, + `--extensions-dir=${options.workspace.extensions}`, + "--no-sandbox", + "--disable-gpu-sandbox", + "--disable-updates", + "--skip-welcome", + "--skip-release-notes", + "--disable-workspace-trust", + `--extensionTestsPath=${options.extensionTestsPath}`, + `--extensionDevelopmentPath=${options.extensionDevelopmentPath}`, + ] +} + +async function runPhase(options: RestartCoordinatorOptions, phase: RestartPhase): Promise { + const args = phaseArguments(options) + console.log(`[restart:${phase}] spawning VS Code: ${path.basename(options.vscodeExecutablePath)}`) + + const child = spawn(options.vscodeExecutablePath, args, { + env: phaseEnvironment(options, phase), + shell: process.platform === "win32", + stdio: ["ignore", "pipe", "pipe"], + }) + + child.stdout?.on("data", (chunk: Buffer) => process.stdout.write(`[restart:${phase}] ${chunk}`)) + child.stderr?.on("data", (chunk: Buffer) => process.stderr.write(`[restart:${phase}] ${chunk}`)) + + const exit = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => { + child.once("error", reject) + child.once("close", (code, signal) => resolve({ code, signal })) + }) + + const result = await readPhaseResult(options.workspace.results, phase) + if (result.status !== "passed") { + throw new Error(`[restart:${phase}] phase result was not passed: ${result.error?.message ?? "unknown failure"}`) + } + + const expectedExit = options.expectedExitPolicies.find((policy) => policy.phase === phase) + const isExpectedNonzeroExit = + exit.code !== 0 && + expectedExit !== undefined && + expectedExit.termination === "graceful-quit" && + exit.code === expectedExit.code && + exit.signal === expectedExit.signal + const isSuccessfulExit = exit.code === 0 && exit.signal === null + if (!isSuccessfulExit && !isExpectedNonzeroExit) { + throw new Error(`[restart:${phase}] VS Code exited with ${exit.code ?? exit.signal}`) + } + + return result +} + +export async function runRestartScenario(options: RestartCoordinatorOptions): Promise { + await runPhase(options, "create") + await runPhase(options, "verify") +} diff --git a/apps/vscode-e2e/src/runTest.ts b/apps/vscode-e2e/src/runTest.ts index 482e73e945..eaf7310207 100644 --- a/apps/vscode-e2e/src/runTest.ts +++ b/apps/vscode-e2e/src/runTest.ts @@ -3,7 +3,7 @@ import * as os from "os" import * as fs from "fs/promises" import { readFileSync } from "fs" -import { runTests } from "@vscode/test-electron" +import { downloadAndUnzipVSCode, runTests } from "@vscode/test-electron" import { LLMock } from "@copilotkit/aimock" import { addApplyDiffResultFixtures } from "./fixtures/apply-diff" @@ -21,6 +21,8 @@ import { addSearchFilesResultFixtures } from "./fixtures/search-files" import { addSubtaskFixtures } from "./fixtures/subtasks" import { addUseMcpToolResultFixtures } from "./fixtures/use-mcp-tool" import { addWriteToFileResultFixtures } from "./fixtures/write-to-file" +import { createScenarioWorkspace, removeScenarioWorkspace } from "./restart/scenarioWorkspace" +import { runRestartScenario } from "./restart/vscodeCoordinator" function getCliFlagValue(flag: string) { return process.argv.find((arg, index) => process.argv[index - 1] === flag) @@ -43,6 +45,10 @@ function isBedrockTargetedRun(testFile?: string, testGrep?: string) { return testGrep?.toLowerCase().includes("bedrock") ?? false } +function isRestartPersistenceTargetedRun(testFile?: string): boolean { + return testFile?.toLowerCase().includes("restart-persistence") ?? false +} + async function main() { const isRecord = process.env.AIMOCK_RECORD === "true" const testGrep = getCliFlagValue("--grep") || process.env.TEST_GREP @@ -50,6 +56,7 @@ async function main() { const isDeepSeekTest = isDeepSeekTargetedRun(testFile, testGrep) const isGeminiTest = testFile?.toLowerCase().includes("gemini.test") ?? false const isBedrockTest = isBedrockTargetedRun(testFile, testGrep) + const isRestartPersistenceTest = isRestartPersistenceTargetedRun(testFile) if (isRecord && isDeepSeekTest && !process.env.DEEPSEEK_API_KEY) { throw new Error("AIMOCK_RECORD=true requires DEEPSEEK_API_KEY to record DeepSeek fixtures") @@ -83,11 +90,14 @@ async function main() { const extensionTestsPath = path.resolve(__dirname, "./suite/index") let testWorkspace: string | undefined + let scenarioWorkspace: Awaited> | undefined try { - // Create a temporary workspace folder for tests before installing fixtures that - // need workspace-specific paths. - testWorkspace = await fs.mkdtemp(path.join(os.tmpdir(), "roo-test-workspace-")) + // Create a temporary workspace folder for regular tests. Restart scenarios own + // all of their paths under the dedicated scenario root below. + if (!isRestartPersistenceTest) { + testWorkspace = await fs.mkdtemp(path.join(os.tmpdir(), "roo-test-workspace-")) + } if (useMock) { const fixturesDir = path.resolve(__dirname, "../fixtures") @@ -173,13 +183,36 @@ async function main() { const pkg = JSON.parse(readFileSync(path.resolve(__dirname, "../package.json"), "utf-8")) const vscodeVersion = process.env.VSCODE_VERSION || pkg.devDependencies["@types/vscode"] - await runTests({ - extensionDevelopmentPath, - extensionTestsPath, - launchArgs: [testWorkspace], - extensionTestsEnv, - version: vscodeVersion, - }) + if (isRestartPersistenceTest) { + scenarioWorkspace = await createScenarioWorkspace() + const vscodeExecutablePath = await downloadAndUnzipVSCode({ + version: vscodeVersion, + extensionDevelopmentPath, + }) + await runRestartScenario({ + vscodeExecutablePath, + extensionDevelopmentPath, + extensionTestsPath, + scenario: "restart-persistence", + workspace: scenarioWorkspace, + environment: extensionTestsEnv, + expectedExitPolicies: [ + { phase: "create", termination: "graceful-quit", code: 1, signal: null }, + { phase: "verify", termination: "graceful-quit", code: 1, signal: null }, + ], + }) + } else { + if (!testWorkspace) { + throw new Error("Regular E2E runs require a temporary test workspace") + } + await runTests({ + extensionDevelopmentPath, + extensionTestsPath, + launchArgs: [testWorkspace], + extensionTestsEnv, + version: vscodeVersion, + }) + } } catch (error) { console.error("Failed to run tests", error) process.exitCode = 1 @@ -187,6 +220,9 @@ async function main() { if (testWorkspace) { await fs.rm(testWorkspace, { recursive: true, force: true }) } + if (scenarioWorkspace) { + await removeScenarioWorkspace(scenarioWorkspace) + } await mock?.stop() } } diff --git a/apps/vscode-e2e/src/suite/index.ts b/apps/vscode-e2e/src/suite/index.ts index 63d29ec28c..e93d73bd37 100644 --- a/apps/vscode-e2e/src/suite/index.ts +++ b/apps/vscode-e2e/src/suite/index.ts @@ -72,7 +72,7 @@ export async function run() { testFiles = await glob(`**/${specificFile}`, { cwd }) console.log(`Running specific test file: ${specificFile}`) } else { - testFiles = await glob("**/**.test.js", { cwd }) + testFiles = await glob("**/**.test.js", { cwd, ignore: "**/suite/restart-persistence.test.js" }) } if (testFiles.length === 0) { diff --git a/apps/vscode-e2e/src/suite/restart-persistence.test.ts b/apps/vscode-e2e/src/suite/restart-persistence.test.ts new file mode 100644 index 0000000000..a944a5a389 --- /dev/null +++ b/apps/vscode-e2e/src/suite/restart-persistence.test.ts @@ -0,0 +1,121 @@ +import * as assert from "assert" +import * as vscode from "vscode" + +import { RooCodeEventName, type RooCodeAPI } from "@roo-code/types" + +import { + PHASE_RESULT_VERSION, + readPhaseResult, + serializePhaseError, + type PhaseResult, + writePhaseResult, +} from "../restart/phaseProtocol" +import { waitFor, waitUntilCompleted } from "./utils" + +const SCENARIO = "restart-persistence" +const MARKER = "RESTART_PERSISTENCE_MARKER" + +function getResultsDir(): string { + const resultsDir = process.env.E2E_RESULTS_DIR + if (!resultsDir) throw new Error("E2E_RESULTS_DIR is required") + return resultsDir +} + +async function quitGracefully(): Promise { + await vscode.commands.executeCommand("workbench.action.quit") +} + +async function runCreate(api: RooCodeAPI): Promise { + let taskId: string | undefined + let sawMarker = false + const messageHandler = ({ message }: { message: { type: string; text?: string; partial?: boolean } }) => { + if (message.type === "say" && message.partial === false && message.text?.includes(MARKER)) { + sawMarker = true + } + } + api.on(RooCodeEventName.Message, messageHandler) + + try { + taskId = await api.startNewTask({ + configuration: { mode: "ask", alwaysAllowModeSwitch: true, autoApprovalEnabled: true }, + text: `${SCENARIO}: RESTART_PERSISTENCE_SMOKE`, + }) + await waitUntilCompleted({ api, taskId }) + assert.strictEqual(sawMarker, true, `Completion should include ${MARKER}`) + const historyItem = await api.getTaskHistoryItem(taskId) + assert.ok(historyItem, "Completed task should have a history item") + assert.ok(historyItem.task.includes("RESTART_PERSISTENCE_SMOKE"), "History title should include the marker") + const conversationLength = await api.getTaskApiConversationHistoryLength(taskId) + assert.ok(conversationLength > 0, "Completed task should persist API conversation history") + + const result: PhaseResult = { + version: PHASE_RESULT_VERSION, + phase: "create", + status: "passed", + values: { taskId, conversationLength: String(conversationLength) }, + } + await writePhaseResult(getResultsDir(), result) + await quitGracefully() + } catch (error) { + await writePhaseResult(getResultsDir(), { + version: PHASE_RESULT_VERSION, + phase: "create", + status: "failed", + error: serializePhaseError(error), + }) + throw error + } finally { + api.off(RooCodeEventName.Message, messageHandler) + if (taskId && api.getCurrentTaskStack().includes(taskId)) await api.cancelCurrentTask() + } +} + +async function runVerify(api: RooCodeAPI): Promise { + try { + const createResult = await readPhaseResult(getResultsDir(), "create") + assert.strictEqual(createResult.status, "passed") + const taskId = createResult.values?.taskId + assert.ok(taskId, "Create phase should record a task ID") + + await waitFor(() => api.isReady()) + assert.strictEqual(await api.isTaskInHistory(taskId), true, "Task should be present after restart") + const historyItem = await api.getTaskHistoryItem(taskId) + assert.ok(historyItem, "Task history item should be available after restart") + assert.ok(historyItem.task.includes("RESTART_PERSISTENCE_SMOKE"), "History title should persist after restart") + const conversationLength = await api.getTaskApiConversationHistoryLength(taskId) + assert.strictEqual( + conversationLength, + Number(createResult.values?.conversationLength), + "Conversation history length should persist", + ) + + await writePhaseResult(getResultsDir(), { + version: PHASE_RESULT_VERSION, + phase: "verify", + status: "passed", + values: { taskId, conversationLength: String(conversationLength) }, + }) + await quitGracefully() + } catch (error) { + await writePhaseResult(getResultsDir(), { + version: PHASE_RESULT_VERSION, + phase: "verify", + status: "failed", + error: serializePhaseError(error), + }) + throw error + } +} + +suite("Restart persistence", () => { + test("persists completed task across a fresh extension host", async () => { + const api = globalThis.api + if (process.env.E2E_PHASE === "create") { + await runCreate(api) + } else if (process.env.E2E_PHASE === "verify") { + await runVerify(api) + } else { + throw new Error(`Unknown E2E_PHASE: ${process.env.E2E_PHASE ?? "unset"}`) + } + }) +}) diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 569c846c29..87ff0dfec5 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -784,11 +784,6 @@ "count": 1 } }, - "core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts": { - "@typescript-eslint/no-explicit-any": { - "count": 3 - } - }, "core/task-persistence/__tests__/TaskHistoryStore.spec.ts": { "@typescript-eslint/no-explicit-any": { "count": 4 From 79463c3aacea86dd5a71d30a3f10ac0f65349fdd Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Mon, 10 Aug 2026 02:00:07 +0000 Subject: [PATCH 4/5] test(e2e): verify persistence across VS Code restart --- .github/workflows/e2e.yml | 6 + apps/vscode-e2e/src/runTest.ts | 10 +- src/core/task-persistence/TaskHistoryStore.ts | 150 +++++++++++------- .../TaskHistoryStore.reconciliation.spec.ts | 90 +++++++++-- .../__tests__/providerModelConfig.spec.ts | 6 +- .../settings/utils/providerModelConfig.ts | 8 +- 6 files changed, 184 insertions(+), 86 deletions(-) diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 2d1819b4e1..02d38ddfa0 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -150,6 +150,12 @@ jobs: if: github.event_name != 'pull_request' || steps.e2e-marker.outputs.cache-hit != 'true' run: xvfb-run -a pnpm --filter @roo-code/vscode-e2e test:ci:mock + - name: Run mocked restart-persistence E2E test + # Reuse the runner built by the full mocked suite; this direct invocation avoids + # test:run's dotenv loading and does not repeat the bundle or webview build. + if: github.event_name != 'pull_request' || steps.e2e-marker.outputs.cache-hit != 'true' + run: TEST_FILE=restart-persistence.test USE_MOCK=true xvfb-run -a pnpm --filter @roo-code/vscode-e2e exec node ./out/runTest.js + - name: Explain skipped mocked E2E pass marker if: steps.e2e-marker.outputs.cache-hit != 'true' && steps.run-e2e.outcome == 'success' && steps.vscode-fallback.outputs.used == 'true' run: echo "Skipping mocked E2E pass marker because tests ran against a stale cached VS Code binary (VS Code download endpoints unreachable)." diff --git a/apps/vscode-e2e/src/runTest.ts b/apps/vscode-e2e/src/runTest.ts index eaf7310207..8162f34068 100644 --- a/apps/vscode-e2e/src/runTest.ts +++ b/apps/vscode-e2e/src/runTest.ts @@ -45,8 +45,12 @@ function isBedrockTargetedRun(testFile?: string, testGrep?: string) { return testGrep?.toLowerCase().includes("bedrock") ?? false } -function isRestartPersistenceTargetedRun(testFile?: string): boolean { - return testFile?.toLowerCase().includes("restart-persistence") ?? false +function isRestartPersistenceTargetedRun(testFile?: string, testGrep?: string): boolean { + if (testFile?.toLowerCase().includes("restart-persistence")) { + return true + } + + return testGrep?.toLowerCase().includes("restart persistence") ?? false } async function main() { @@ -56,7 +60,7 @@ async function main() { const isDeepSeekTest = isDeepSeekTargetedRun(testFile, testGrep) const isGeminiTest = testFile?.toLowerCase().includes("gemini.test") ?? false const isBedrockTest = isBedrockTargetedRun(testFile, testGrep) - const isRestartPersistenceTest = isRestartPersistenceTargetedRun(testFile) + const isRestartPersistenceTest = isRestartPersistenceTargetedRun(testFile, testGrep) if (isRecord && isDeepSeekTest && !process.env.DEEPSEEK_API_KEY) { throw new Error("AIMOCK_RECORD=true requires DEEPSEEK_API_KEY to record DeepSeek fixtures") diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 01ec41e5d4..764266a825 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -2,6 +2,7 @@ import * as fs from "fs/promises" import * as fsSync from "fs" import * as path from "path" +import deepEqual from "fast-deep-equal" import type { HistoryItem } from "@roo-code/types" import { GlobalFileNames } from "../../shared/globalFileNames" @@ -96,6 +97,7 @@ export class TaskHistoryStore { private readonly globalStoragePath: string private readonly onWrite?: (items: HistoryItem[]) => Promise 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 @@ -137,10 +139,14 @@ export class TaskHistoryStore { await this.loadIndex() // 2. Reconcile cache against actual task directories on disk - await this.reconcile() + await this.reconcile({ forceRefresh: true }) // 3. Complete any two-record repair interrupted after its intent was durable. - await this.replayDelegationRepairIntent() + 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 await this.reconcileDelegationState() @@ -321,7 +327,7 @@ export class TaskHistoryStore { * - Tasks on disk but missing from cache: read and add * - Tasks in cache but missing from disk: remove */ - async reconcile(): Promise { + async reconcile(options: { forceRefresh?: boolean } = {}): Promise { // Run through the write lock to prevent interleaving with upsert/delete return this.withLock(async () => { const tasksDir = await this.getTasksDir() @@ -340,14 +346,29 @@ export class TaskHistoryStore { const cacheIds = new Set(this.cache.keys()) let changed = false - // Task files are authoritative. Always refresh entries from disk so a stale - // index cannot overwrite a repair or another instance's newer task state. + // 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) + if ( + !options.forceRefresh && + this.cache.has(taskId) && + this.taskFileMtimes.get(taskId) === mtimeMs + ) { + continue + } + const item = await this.readTaskFile(taskId) if (item) { - this.cache.set(taskId, item) - changed = true + const previous = this.cache.get(taskId) + this.taskFileMtimes.set(taskId, mtimeMs) + if (!deepEqual(previous, item)) { + this.cache.set(taskId, item) + changed = true + } } } catch { // Corrupted or missing file, skip @@ -358,6 +379,7 @@ export class TaskHistoryStore { for (const taskId of cacheIds) { if (!onDiskIds.has(taskId)) { this.cache.delete(taskId) + this.taskFileMtimes.delete(taskId) changed = true } } @@ -414,61 +436,66 @@ export class TaskHistoryStore { continue } - if (!item.awaitingChildId) { - await this.upsertCore( - { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, - ) - repairsInThisPass++ - continue - } + try { + if (!item.awaitingChildId) { + await this.upsertCore( + { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, + ) + repairsInThisPass++ + continue + } - const child = byId.get(item.awaitingChildId) - - if (!child) { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, - ) - repairsInThisPass++ - } else if ((child.status ?? "active") === "active" && persistedActiveIds.has(child.id)) { - // An active child persisted across startup cannot have a live task session - // behind it. Mark it interrupted before releasing the parent's delegation - // link so the normal resume/re-delegate flow can take over. This is an - // administrative recovery, not a runtime delegation transition. - await this.repairActiveDelegation(item, child) - console.warn( - `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, - ) - repairsInThisPass++ - } else if (child.status === "completed") { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - completedByChildId: child.id, - completionResultSummary: - child.completionResultSummary ?? "Task completed (recovered after interruption)", - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, - ) - repairsInThisPass++ + const child = byId.get(item.awaitingChildId) + + if (!child) { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, + ) + repairsInThisPass++ + } else if ((child.status ?? "active") === "active" && persistedActiveIds.has(child.id)) { + // An active child persisted across startup cannot have a live task session + // behind it. Mark it interrupted before releasing the parent's delegation + // link so the normal resume/re-delegate flow can take over. This is an + // administrative recovery, not a runtime delegation transition. + await this.repairActiveDelegation(item, child) + console.warn( + `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, + ) + repairsInThisPass++ + } else if (child.status === "completed") { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + completedByChildId: child.id, + completionResultSummary: + child.completionResultSummary ?? + "Task completed (recovered after interruption)", + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, + ) + repairsInThisPass++ + } + } catch (error) { + console.error(`[TaskHistoryStore] Failed to reconcile delegation for task ${item.id}:`, error) } // child.status === "interrupted" or "delegated" → leave as-is this pass } @@ -726,6 +753,7 @@ export class TaskHistoryStore { } else { this.cache.delete(taskId) } + this.taskFileMtimes.delete(taskId) } catch { this.cache.delete(taskId) } diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index d08ef999f9..b6e6f80c2d 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -144,6 +144,12 @@ describe("assertValidTransition", () => { describe("TaskHistoryStore reconcileDelegationState", () => { let tmpDir: string let store: TaskHistoryStore + const disposables = new Set() + + function registerStore(nextStore: TaskHistoryStore): TaskHistoryStore { + disposables.add(nextStore) + return nextStore + } async function seedItems(items: HistoryItem[]): Promise { const tasksDir = path.join(tmpDir, "tasks") @@ -157,12 +163,13 @@ describe("TaskHistoryStore reconcileDelegationState", () => { beforeEach(async () => { tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "reconcile-test-")) - store = new TaskHistoryStore(tmpDir) + store = registerStore(new TaskHistoryStore(tmpDir)) }) afterEach(async () => { safeWriteJsonMock.mockImplementation(writeJson) - store.dispose() + for (const disposable of disposables) disposable.dispose() + disposables.clear() await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}) }) @@ -336,14 +343,13 @@ describe("TaskHistoryStore reconcileDelegationState", () => { throw new Error("fault before child write") await writeJson(filePath, data) }) - await expect(store.initialize()).rejects.toThrow("fault before child write") + await expect(store.initialize()).resolves.toBeUndefined() store.dispose() safeWriteJsonMock.mockImplementation(writeJson) - const replayedStore = new TaskHistoryStore(tmpDir) + const replayedStore = registerStore(new TaskHistoryStore(tmpDir)) await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") - replayedStore.dispose() }) it("replays an intent after a failure before the parent write", async () => { @@ -364,14 +370,13 @@ describe("TaskHistoryStore reconcileDelegationState", () => { throw new Error("fault before parent write") await writeJson(filePath, data) }) - await expect(store.initialize()).rejects.toThrow("fault before parent write") + await expect(store.initialize()).resolves.toBeUndefined() store.dispose() safeWriteJsonMock.mockImplementation(writeJson) - const replayedStore = new TaskHistoryStore(tmpDir) + const replayedStore = registerStore(new TaskHistoryStore(tmpDir)) await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") - replayedStore.dispose() }) it("retains an intent when the callback fails after both writes", async () => { @@ -384,18 +389,19 @@ describe("TaskHistoryStore reconcileDelegationState", () => { }) await seedItems([parent, child]) store.dispose() - store = new TaskHistoryStore(tmpDir, { onWrite: vi.fn().mockRejectedValue(new Error("fault before cleanup")) }) - await expect(store.initialize()).rejects.toThrow("fault before cleanup") + store = registerStore( + new TaskHistoryStore(tmpDir, { onWrite: vi.fn().mockRejectedValue(new Error("fault before cleanup")) }), + ) + await expect(store.initialize()).resolves.toBeUndefined() const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) expect(await fs.readFile(intentPath, "utf8")).toContain(child.id) store.dispose() safeWriteJsonMock.mockImplementation(writeJson) - const replayedStore = new TaskHistoryStore(tmpDir) + const replayedStore = registerStore(new TaskHistoryStore(tmpDir)) await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") await expect(fs.access(intentPath)).rejects.toThrow() - replayedStore.dispose() }) it("quarantines malformed and stale intents without blocking unrelated startup", async () => { @@ -415,6 +421,60 @@ describe("TaskHistoryStore reconcileDelegationState", () => { ).toBe(true) }) + it("quarantines an intent with a missing task record without changing unrelated startup", async () => { + const unrelated = makeItem({ id: "unrelated-missing-intent", status: "active" }) + const missingChild = makeItem({ id: "missing-intent-child", status: "active" }) + const parent = makeItem({ id: "missing-intent-parent", status: "delegated", awaitingChildId: missingChild.id }) + await seedItems([unrelated]) + const tasksDir = path.join(tmpDir, "tasks") + const intentPath = path.join(tasksDir, GlobalFileNames.delegationRepairIntent) + await fs.writeFile(intentPath, JSON.stringify(makeRepairIntent(parent, missingChild))) + + await store.initialize() + + expect(store.get(unrelated.id)?.status).toBe("active") + expect(store.get(parent.id)).toBeUndefined() + expect( + (await fs.readdir(tasksDir)).some((name) => + name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), + ), + ).toBe(true) + }) + + it("quarantines an intent when the parent no longer matches its repair guard", async () => { + const child = makeItem({ + id: "mismatched-intent-child", + status: "interrupted", + parentTaskId: "mismatched-intent-parent", + }) + const parent = makeItem({ + id: "mismatched-intent-parent", + status: "completed", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + const tasksDir = path.join(tmpDir, "tasks") + const intentPath = path.join(tasksDir, GlobalFileNames.delegationRepairIntent) + await fs.writeFile(intentPath, JSON.stringify(makeRepairIntent({ ...parent, status: "delegated" }, child))) + const childPath = path.join(tasksDir, child.id, GlobalFileNames.historyItem) + const parentPath = path.join(tasksDir, parent.id, GlobalFileNames.historyItem) + const beforeChild = await fs.readFile(childPath, "utf8") + const beforeParent = await fs.readFile(parentPath, "utf8") + + await store.initialize() + + expect(store.get(child.id)?.status).toBe("interrupted") + expect(store.get(parent.id)?.status).toBe("completed") + expect(await fs.readFile(childPath, "utf8")).toBe(beforeChild) + expect(await fs.readFile(parentPath, "utf8")).toBe(beforeParent) + expect( + (await fs.readdir(tasksDir)).some((name) => + name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), + ), + ).toBe(true) + }) + it("repairs invalid delegation: delegated parent with no awaitingChildId → active (clears delegatedToId and awaitingChildId)", async () => { // awaitingChildId is falsy but explicitly set (empty string), delegatedToId is stale const parent = makeItem({ @@ -495,7 +555,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { const afterFirstChild = { ...store.get(child.id) } store.dispose() - const store2 = new TaskHistoryStore(tmpDir) + const store2 = registerStore(new TaskHistoryStore(tmpDir)) await store2.initialize() const afterSecondParent = { ...store2.get(parent.id) } const afterSecondChild = { ...store2.get(child.id) } @@ -516,7 +576,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { const afterFirst = { ...store.get("parent-6") } store.dispose() - const store2 = new TaskHistoryStore(tmpDir) + const store2 = registerStore(new TaskHistoryStore(tmpDir)) await store2.initialize() const afterSecond = { ...store2.get("parent-6") } store2.dispose() @@ -544,7 +604,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { it("invokes onWrite callback after startup repairs", async () => { const onWrite = vi.fn().mockResolvedValue(undefined) store.dispose() - store = new TaskHistoryStore(tmpDir, { onWrite }) + store = registerStore(new TaskHistoryStore(tmpDir, { onWrite })) const parent = makeItem({ id: "parent-onwrite", status: "delegated", awaitingChildId: "nonexistent-child" }) await seedItems([parent]) diff --git a/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts b/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts index 3eb8dc08b8..153f30b0e2 100644 --- a/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts +++ b/webview-ui/src/components/settings/utils/__tests__/providerModelConfig.spec.ts @@ -64,7 +64,7 @@ describe("providerModelConfig", () => { }) it("returns fallback config for unknown provider", () => { - const config = getProviderServiceConfig("unknown-provider") + const config = getProviderServiceConfig("unknown-provider" as any) expect(config.serviceName).toBe("unknown-provider") expect(config.serviceUrl).toBe("") }) @@ -88,7 +88,7 @@ describe("providerModelConfig", () => { }) it("returns empty string for unknown provider", () => { - const defaultId = getDefaultModelIdForProvider("unknown") + const defaultId = getDefaultModelIdForProvider("unknown" as any) expect(defaultId).toBe("") }) @@ -152,7 +152,7 @@ describe("providerModelConfig", () => { }) it("returns undefined for a provider with no model config entry", () => { - expect(getProviderModelConfig("unknown-provider")).toBeUndefined() + expect(getProviderModelConfig("unknown-provider" as any)).toBeUndefined() }) it("returns the static field config for a non-zai provider", () => { diff --git a/webview-ui/src/components/settings/utils/providerModelConfig.ts b/webview-ui/src/components/settings/utils/providerModelConfig.ts index 4396ddc68b..eccbf7ba1d 100644 --- a/webview-ui/src/components/settings/utils/providerModelConfig.ts +++ b/webview-ui/src/components/settings/utils/providerModelConfig.ts @@ -90,11 +90,11 @@ export const PROVIDER_DEFAULT_MODEL_IDS: Partial> = [providerIdentifiers.baseten]: basetenDefaultModelId, } -export const getProviderServiceConfig = (provider: string): ProviderServiceConfig => { - return PROVIDER_SERVICE_CONFIG[provider as ProviderName] ?? { serviceName: provider, serviceUrl: "" } +export const getProviderServiceConfig = (provider: ProviderName): ProviderServiceConfig => { + return PROVIDER_SERVICE_CONFIG[provider] ?? { serviceName: provider, serviceUrl: "" } } -export const getDefaultModelIdForProvider = (provider: string, apiConfiguration?: ProviderSettings): string => { +export const getDefaultModelIdForProvider = (provider: ProviderName, apiConfiguration?: ProviderSettings): string => { // Handle Z.ai's China/International entrypoint distinction if (provider === providerIdentifiers.zai && apiConfiguration) { return apiConfiguration.zaiApiLine === "china_coding" @@ -102,7 +102,7 @@ export const getDefaultModelIdForProvider = (provider: string, apiConfiguration? : internationalZAiDefaultModelId } - return PROVIDER_DEFAULT_MODEL_IDS[provider as ProviderName] ?? "" + return PROVIDER_DEFAULT_MODEL_IDS[provider] ?? "" } export type ProviderModelConfig = { From 075d3ce8e4afcbe3f3608e0d961b4bc92793c377 Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Mon, 10 Aug 2026 17:27:28 +0000 Subject: [PATCH 5/5] fix(task-persistence): harden delegation repair recovery --- src/core/task-persistence/TaskHistoryStore.ts | 290 ++++++++++-------- .../TaskHistoryStore.reconciliation.spec.ts | 204 +++++++++++- .../__tests__/TaskHistoryStore.spec.ts | 40 +++ 3 files changed, 405 insertions(+), 129 deletions(-) diff --git a/src/core/task-persistence/TaskHistoryStore.ts b/src/core/task-persistence/TaskHistoryStore.ts index 764266a825..e4707ee0a9 100644 --- a/src/core/task-persistence/TaskHistoryStore.ts +++ b/src/core/task-persistence/TaskHistoryStore.ts @@ -1,6 +1,7 @@ import * as fs from "fs/promises" import * as fsSync from "fs" import * as path from "path" +import crypto from "crypto" import deepEqual from "fast-deep-equal" import type { HistoryItem } from "@roo-code/types" @@ -140,6 +141,10 @@ export class TaskHistoryStore { // 2. Reconcile cache against actual task directories on disk 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. try { @@ -149,7 +154,7 @@ export class TaskHistoryStore { } // 4. Repair delegation inconsistencies left by a previous crash - await this.reconcileDelegationState() + await this.reconcileDelegationState(persistedActiveIds) // 5. Start fs.watch for cross-instance reactivity this.startWatcher() @@ -276,6 +281,7 @@ export class TaskHistoryStore { async delete(taskId: string): Promise { return this.withLock(async () => { this.cache.delete(taskId) + this.taskFileMtimes.delete(taskId) // Remove per-task file (best-effort) try { @@ -301,6 +307,7 @@ export class TaskHistoryStore { return this.withLock(async () => { for (const taskId of taskIds) { this.cache.delete(taskId) + this.taskFileMtimes.delete(taskId) try { const filePath = await this.getTaskFilePath(taskId) @@ -413,94 +420,107 @@ export class TaskHistoryStore { * resumable. An `active` child is treated as orphaned during startup recovery because * no live task session exists to own it. */ - private async reconcileDelegationState(): Promise { - return this.withLock(async () => { - // Only statuses loaded from persistence represent sessions that could have - // been orphaned by a crash. A delegated parent repaired to active earlier in - // this pass remains resumable and must not be mistaken for a second orphaned - // child in a delegation chain. - const persistedActiveIds = new Set( - Array.from(this.cache.values()) - .filter((item) => (item.status ?? "active") === "active") - .map((item) => item.id), - ) - let repairsInThisPass: number - do { - repairsInThisPass = 0 - // Rebuild the lookup map each pass so repairs from the previous pass - // are visible when evaluating chained delegations. - const byId = new Map(Array.from(this.cache.values()).map((i) => [i.id, i])) - - for (const [, item] of byId) { - if (item.status !== "delegated") { + private async reconcileDelegationState(persistedActiveIds: ReadonlySet): Promise { + return this.withLock(() => this.reconcileDelegationStateCore(persistedActiveIds)) + } + + /** + * Reconcile delegation state while the store lock is already held. + * + * Callers that do not hold the lock must use `reconcileDelegationState()`. + * Migration uses this core method so its cache/file/index updates and the + * follow-up repair remain one serialized operation without re-entering the + * non-reentrant lock. + */ + private async reconcileDelegationStateCore(persistedActiveIds: ReadonlySet): Promise { + // Only statuses loaded from persistence represent sessions that could have + // been orphaned by a crash. A delegated parent repaired to active earlier in + // this pass remains resumable and must not be mistaken for a second orphaned + // child in a delegation chain. The snapshot is intentionally captured before + // repair-intent replay and remains unchanged for the entire reconciliation. + let repairsInThisPass: number + do { + repairsInThisPass = 0 + // Rebuild the lookup map each pass so repairs from the previous pass + // are visible when evaluating chained delegations. + const byId = new Map(Array.from(this.cache.values()).map((i) => [i.id, i])) + + for (const [, item] of byId) { + if (item.status !== "delegated") { + continue + } + + try { + if (!item.awaitingChildId) { + await this.upsertCore( + { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, + ) + repairsInThisPass++ continue } - try { - if (!item.awaitingChildId) { - await this.upsertCore( - { ...item, status: "active", awaitingChildId: undefined, delegatedToId: undefined }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled invalid delegation: task ${item.id} → active (no awaitingChildId)`, - ) - repairsInThisPass++ - continue - } - - const child = byId.get(item.awaitingChildId) - - if (!child) { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, - ) - repairsInThisPass++ - } else if ((child.status ?? "active") === "active" && persistedActiveIds.has(child.id)) { - // An active child persisted across startup cannot have a live task session - // behind it. Mark it interrupted before releasing the parent's delegation - // link so the normal resume/re-delegate flow can take over. This is an - // administrative recovery, not a runtime delegation transition. - await this.repairActiveDelegation(item, child) - console.warn( - `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, - ) - repairsInThisPass++ - } else if (child.status === "completed") { - await this.upsertCore( - { - ...item, - status: "active", - awaitingChildId: undefined, - delegatedToId: undefined, - completedByChildId: child.id, - completionResultSummary: - child.completionResultSummary ?? - "Task completed (recovered after interruption)", - }, - { skipTransitionCheck: true }, - ) - console.warn( - `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, - ) - repairsInThisPass++ - } - } catch (error) { - console.error(`[TaskHistoryStore] Failed to reconcile delegation for task ${item.id}:`, error) + const child = byId.get(item.awaitingChildId) + + if (!child) { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled orphaned delegation: task ${item.id} → active (child ${item.awaitingChildId} not found)`, + ) + repairsInThisPass++ + } else if ((child.status ?? "active") === "active" && persistedActiveIds.has(child.id)) { + // An active child persisted across startup cannot have a live task session + // behind it. Mark it interrupted before releasing the parent's delegation + // link so the normal resume/re-delegate flow can take over. This is an + // administrative recovery, not a runtime delegation transition. + await this.repairActiveDelegation(item, child) + console.warn( + `[TaskHistoryStore] Reconciled orphaned active child: child ${child.id} → interrupted, task ${item.id} → active`, + ) + repairsInThisPass++ + } else if (child.status === "completed") { + await this.upsertCore( + { + ...item, + status: "active", + awaitingChildId: undefined, + delegatedToId: undefined, + completedByChildId: child.id, + completionResultSummary: + child.completionResultSummary ?? "Task completed (recovered after interruption)", + }, + { skipTransitionCheck: true }, + ) + console.warn( + `[TaskHistoryStore] Reconciled interrupted handoff: task ${item.id} → active (child ${item.awaitingChildId} already completed)`, + ) + repairsInThisPass++ } - // child.status === "interrupted" or "delegated" → leave as-is this pass + } catch (error) { + console.error(`[TaskHistoryStore] Failed to reconcile delegation for task ${item.id}:`, error) } - } while (repairsInThisPass > 0) - }) + // child.status === "interrupted" or "delegated" → leave as-is this pass + } + } while (repairsInThisPass > 0) + } + + private getPersistedActiveIds(): ReadonlySet { + return new Set( + Array.from(this.cache.values()) + .filter((item) => (item.status ?? "active") === "active") + .map((item) => item.id), + ) } /** @@ -508,6 +528,10 @@ export class TaskHistoryStore { * The expected fields are guards: an intent may update only the missing side * when the other side is already at its target, or when both records still * describe the original delegated handoff. + * + * This method acquires the store's non-reentrant promise-chain lock. It must be + * called outside an existing `withLock` callback; locked callers must use the + * corresponding core methods directly instead of awaiting this method. */ private async replayDelegationRepairIntent(): Promise { return this.withLock(async () => { @@ -527,20 +551,20 @@ export class TaskHistoryStore { } const childAtTarget = child.status === intent.target.childStatus - const parentAtTarget = + const parentMatchesTargetState = parent.status === intent.target.parentStatus && parent.awaitingChildId === undefined && parent.delegatedToId === undefined const childMatchesExpected = this.matchesDelegationRepairChildPreconditions(intent, child) const parentMatchesExpected = this.matchesDelegationRepairParentPreconditions(intent, parent) - if ((!childAtTarget && !childMatchesExpected) || (!parentAtTarget && !parentMatchesExpected)) { + if ((!childAtTarget && !childMatchesExpected) || (!parentMatchesTargetState && !parentMatchesExpected)) { await this.quarantineDelegationRepairIntent(intent, "task state no longer matches its guards") return } const repairedChild = childAtTarget ? child : { ...child, status: intent.target.childStatus } - const repairedParent = parentAtTarget + const repairedParent = parentMatchesTargetState ? parent : { ...parent, @@ -550,17 +574,22 @@ export class TaskHistoryStore { } if (!childAtTarget) await this.writeTaskFile(repairedChild) - if (!parentAtTarget) await this.writeTaskFile(repairedParent) + if (!parentMatchesTargetState) await this.writeTaskFile(repairedParent) this.cache.set(repairedChild.id, repairedChild) this.cache.set(repairedParent.id, repairedParent) - this.scheduleIndexWrite() // The journal is retained until the write-through callback succeeds. if (this.onWrite) { 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() }) } @@ -571,7 +600,7 @@ export class TaskHistoryStore { private async repairActiveDelegation(parent: HistoryItem, child: HistoryItem): Promise { const intent: DelegationRepairIntent = { version: 1, - operationId: `delegation-repair-${Date.now()}-${Math.random().toString(36).slice(2)}`, + operationId: crypto.randomUUID(), parentTaskId: parent.id, childTaskId: child.id, expected: { @@ -611,12 +640,14 @@ export class TaskHistoryStore { this.cache.set(repairedChild.id, repairedChild) this.cache.set(repairedParent.id, repairedParent) - this.scheduleIndexWrite() if (this.onWrite) { 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 { @@ -640,10 +671,8 @@ export class TaskHistoryStore { let parsed: unknown try { parsed = JSON.parse(await fs.readFile(intentPath, "utf8")) as unknown - } catch { - try { - await fs.access(intentPath) - } catch { + } catch (error) { + if (this.isFileNotFoundError(error)) { return null } await this.quarantineDelegationRepairIntent(null, "malformed JSON") @@ -673,6 +702,7 @@ export class TaskHistoryStore { ? (expectedRecord.child as Record) : null const target = candidate.target + const targetRecord = target && typeof target === "object" ? (target as Record) : null return ( candidate.version === 1 && typeof candidate.operationId === "string" && @@ -688,10 +718,9 @@ export class TaskHistoryStore { expectedChild.status === "active" && (expectedChild.parentTaskId === undefined || typeof expectedChild.parentTaskId === "string") && (expectedChild.rootTaskId === undefined || typeof expectedChild.rootTaskId === "string") && - !!target && - typeof target === "object" && - (target as Record).childStatus === "interrupted" && - (target as Record).parentStatus === "active" + !!targetRecord && + targetRecord.childStatus === "interrupted" && + targetRecord.parentStatus === "active" ) } @@ -734,6 +763,10 @@ export class TaskHistoryStore { ) } + private isFileNotFoundError(error: unknown): boolean { + return typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT" + } + private async getDelegationRepairIntentPath(): Promise { const tasksDir = await this.getTasksDir() return path.join(tasksDir, GlobalFileNames.delegationRepairIntent) @@ -782,40 +815,43 @@ export class TaskHistoryStore { return } - for (const item of taskHistoryEntries) { - if (!item.id) { - continue - } - - // Check if task directory exists on disk + await this.withLock(async () => { const tasksDir = await this.getTasksDir() - const taskDir = path.join(tasksDir, item.id) - try { - await fs.access(taskDir) - } catch { - // Task directory doesn't exist; skip this entry as it's orphaned in globalState - continue - } + for (const item of taskHistoryEntries) { + if (!item.id) { + continue + } - // Write history_item.json if it doesn't exist yet - const filePath = path.join(taskDir, GlobalFileNames.historyItem) - try { - await fs.access(filePath) - // File already exists, skip (don't overwrite existing per-task files) - } catch { - // File doesn't exist, write it - await safeWriteJson(filePath, item) - this.cache.set(item.id, item) + // Check if task directory exists on disk + const taskDir = path.join(tasksDir, item.id) + + try { + await fs.access(taskDir) + } catch { + // Task directory doesn't exist; skip this entry as it's orphaned in globalState + continue + } + + // Write history_item.json if it doesn't exist yet + const filePath = path.join(taskDir, GlobalFileNames.historyItem) + try { + await fs.access(filePath) + // File already exists, skip (don't overwrite existing per-task files) + } catch { + // File doesn't exist, write it + await safeWriteJson(filePath, item) + this.cache.set(item.id, item) + } } - } - // Write the index - await this.writeIndex() + // Write the index + await this.writeIndex() - // Repair any delegation inconsistencies introduced by the migrated entries. - // reconcileDelegationState() is idempotent so running it again is safe. - await this.reconcileDelegationState() + // 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 ────────────────────────────── diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts index b6e6f80c2d..e788b5d96a 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.reconciliation.spec.ts @@ -322,6 +322,16 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(store.get(child.id)?.status).toBe("interrupted") expect(store.get(parent.id)?.status).toBe("active") + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", parent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedChild).toMatchObject({ id: child.id, status: "interrupted" }) + expect(persistedParent).toMatchObject({ id: parent.id, status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() await expect(fs.access(intentPath)).rejects.toThrow() }) @@ -350,6 +360,16 @@ describe("TaskHistoryStore reconcileDelegationState", () => { await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", parent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedChild).toMatchObject({ id: child.id, status: "interrupted" }) + expect(persistedParent).toMatchObject({ id: parent.id, status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() }) it("replays an intent after a failure before the parent write", async () => { @@ -377,6 +397,16 @@ describe("TaskHistoryStore reconcileDelegationState", () => { await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", parent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedChild).toMatchObject({ id: child.id, status: "interrupted" }) + expect(persistedParent).toMatchObject({ id: parent.id, status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() }) it("retains an intent when the callback fails after both writes", async () => { @@ -401,6 +431,70 @@ describe("TaskHistoryStore reconcileDelegationState", () => { await replayedStore.initialize() expect(replayedStore.get(child.id)?.status).toBe("interrupted") expect(replayedStore.get(parent.id)?.status).toBe("active") + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", parent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedChild).toMatchObject({ id: child.id, status: "interrupted" }) + expect(persistedParent).toMatchObject({ id: parent.id, status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() + await expect(fs.access(intentPath)).rejects.toThrow() + }) + + it("replays a both-at-target intent without writing task files", async () => { + const child = makeItem({ id: "child-at-target", status: "interrupted", parentTaskId: "parent-at-target" }) + const parent = makeItem({ id: "parent-at-target", status: "active" }) + await seedItems([parent, child]) + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + await fs.writeFile( + intentPath, + JSON.stringify(makeRepairIntent({ ...parent, status: "delegated", awaitingChildId: child.id }, child)), + ) + + const writeCalls: string[] = [] + safeWriteJsonMock.mockImplementation(async (filePath, data) => { + writeCalls.push(filePath) + await writeJson(filePath, data) + }) + await store.initialize() + + expect(writeCalls.filter((filePath) => filePath.endsWith(GlobalFileNames.historyItem))).toEqual([]) + await expect(fs.access(intentPath)).rejects.toThrow() + }) + + it("does not schedule the derived index before repair-intent cleanup succeeds", async () => { + const child = makeItem({ id: "child-deferred-index", status: "active", parentTaskId: "parent-deferred-index" }) + const parent = makeItem({ + id: "parent-deferred-index", + status: "delegated", + awaitingChildId: child.id, + delegatedToId: child.id, + }) + await seedItems([parent, child]) + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + 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() }) @@ -419,6 +513,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), ), ).toBe(true) + await expect(fs.access(intentPath)).rejects.toThrow() }) it("quarantines an intent with a missing task record without changing unrelated startup", async () => { @@ -439,6 +534,7 @@ describe("TaskHistoryStore reconcileDelegationState", () => { name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), ), ).toBe(true) + await expect(fs.access(intentPath)).rejects.toThrow() }) it("quarantines an intent when the parent no longer matches its repair guard", async () => { @@ -473,6 +569,48 @@ describe("TaskHistoryStore reconcileDelegationState", () => { name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), ), ).toBe(true) + await expect(fs.access(intentPath)).rejects.toThrow() + }) + + it("quarantines an intent when the child no longer matches its repair guard", async () => { + const child = makeItem({ + id: "child-mismatched-intent", + status: "completed", + parentTaskId: "mismatched-child-parent", + }) + const parent = makeItem({ + id: "parent-mismatched-child-intent", + status: "active", + }) + await seedItems([parent, child]) + const tasksDir = path.join(tmpDir, "tasks") + const intentPath = path.join(tasksDir, GlobalFileNames.delegationRepairIntent) + await fs.writeFile( + intentPath, + JSON.stringify( + makeRepairIntent( + { ...parent, status: "delegated", awaitingChildId: child.id, delegatedToId: child.id }, + { ...child, status: "active" }, + ), + ), + ) + const childPath = path.join(tasksDir, child.id, GlobalFileNames.historyItem) + const parentPath = path.join(tasksDir, parent.id, GlobalFileNames.historyItem) + const beforeChild = await fs.readFile(childPath, "utf8") + const beforeParent = await fs.readFile(parentPath, "utf8") + + await store.initialize() + + expect(store.get(child.id)?.status).toBe("completed") + expect(store.get(parent.id)?.status).toBe("active") + expect(await fs.readFile(childPath, "utf8")).toBe(beforeChild) + expect(await fs.readFile(parentPath, "utf8")).toBe(beforeParent) + await expect(fs.access(intentPath)).rejects.toThrow() + expect( + (await fs.readdir(tasksDir)).some((name) => + name.startsWith(`${GlobalFileNames.delegationRepairIntent}.quarantine-`), + ), + ).toBe(true) }) it("repairs invalid delegation: delegated parent with no awaitingChildId → active (clears delegatedToId and awaitingChildId)", async () => { @@ -517,9 +655,9 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(store.get("parent-b")?.status).toBe("active") }) - it("handles chained delegation (A→B→C) until all orphaned links converge", async () => { + it("repairs an orphaned link in a chained delegation without repairing its grandparent", async () => { // C doesn't exist (orphaned). B is delegated waiting for C → repaired to active. - // A then sees B as an orphaned active child and is repaired as well. + // A sees B as delegated in the persisted startup snapshot and remains delegated. const parentA = makeItem({ id: "parent-a-chain", status: "delegated", awaitingChildId: "parent-b-chain" }) const parentB = makeItem({ id: "parent-b-chain", @@ -540,6 +678,68 @@ describe("TaskHistoryStore reconcileDelegationState", () => { expect(store.get("parent-b-chain")?.awaitingChildId).toBeUndefined() }) + it("does not repair a grandparent when replay repairs the middle node", async () => { + const grandparent = makeItem({ + id: "grandparent-replay-chain", + status: "delegated", + awaitingChildId: "parent-replay-chain", + delegatedToId: "parent-replay-chain", + }) + const parent = makeItem({ + id: "parent-replay-chain", + status: "delegated", + awaitingChildId: "child-replay-chain", + delegatedToId: "child-replay-chain", + parentTaskId: grandparent.id, + rootTaskId: grandparent.id, + }) + const child = makeItem({ + id: "child-replay-chain", + status: "active", + parentTaskId: parent.id, + rootTaskId: grandparent.id, + }) + await seedItems([grandparent, parent, child]) + const intentPath = path.join(tmpDir, "tasks", GlobalFileNames.delegationRepairIntent) + await fs.writeFile(intentPath, JSON.stringify(makeRepairIntent(parent, child))) + + await store.initialize() + + // Replay repairs B/C, but B was delegated at the persisted startup snapshot. + // A must remain delegated to the now-interrupted/resumable B. + expect(store.get(grandparent.id)).toMatchObject({ + id: grandparent.id, + status: "delegated", + awaitingChildId: parent.id, + delegatedToId: parent.id, + }) + expect(store.get(parent.id)).toMatchObject({ id: parent.id, status: "active" }) + expect(store.get(parent.id)?.awaitingChildId).toBeUndefined() + expect(store.get(parent.id)?.delegatedToId).toBeUndefined() + expect(store.get(child.id)).toMatchObject({ id: child.id, status: "interrupted" }) + + const persistedGrandparent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", grandparent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedParent = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", parent.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + const persistedChild = JSON.parse( + await fs.readFile(path.join(tmpDir, "tasks", child.id, GlobalFileNames.historyItem), "utf8"), + ) as HistoryItem + expect(persistedGrandparent).toMatchObject({ + id: grandparent.id, + status: "delegated", + awaitingChildId: parent.id, + delegatedToId: parent.id, + }) + expect(persistedParent).toMatchObject({ id: parent.id, status: "active" }) + expect(persistedParent.awaitingChildId).toBeUndefined() + expect(persistedParent.delegatedToId).toBeUndefined() + expect(persistedChild).toMatchObject({ id: child.id, status: "interrupted" }) + await expect(fs.access(intentPath)).rejects.toThrow() + }) + it("is idempotent when recovering an active child", async () => { const child = makeItem({ id: "child-active-idempotent", status: "active" }) const parent = makeItem({ diff --git a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts index 3b7e9041a4..3188e9c505 100644 --- a/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts +++ b/src/core/task-persistence/__tests__/TaskHistoryStore.spec.ts @@ -373,6 +373,46 @@ describe("TaskHistoryStore", () => { expect(store.get("idem-task")).toBeDefined() }) + + it("serializes migration cache and index updates behind the store lock", async () => { + const tasksDir = path.join(tmpDir, "tasks") + const migrated = makeHistoryItem({ id: "migration-locked" }) + const concurrent = makeHistoryItem({ id: "migration-concurrent" }) + const migratedFile = path.join(tasksDir, migrated.id, GlobalFileNames.historyItem) + await fs.mkdir(path.dirname(migratedFile), { recursive: true }) + + let releaseMigrationWrite!: () => void + const migrationWriteCanFinish = new Promise((resolve) => { + releaseMigrationWrite = resolve + }) + let signalMigrationWriteStarted!: () => void + 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 migration = store.migrateFromGlobalState([migrated]) + await migrationWriteStarted + const concurrentUpsert = store.upsert(concurrent) + + expect(store.get(concurrent.id)).toBeUndefined() + releaseMigrationWrite() + await Promise.all([migration, concurrentUpsert]) + + 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()", () => {