diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 4fc25fde99..1dceb10b1a 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -335,6 +335,15 @@ export class TaskStore extends EventEmitter { public readonly asyncLayer: AsyncDataLayer | null = null; private pluginPostgresSchemaExecutor: ((contracts: readonly LoadedPluginSchemaContract[]) => Promise) | null = null; + /* + FNXC:HandoffFailureInjection 2026-07-15-12:00: + PostgreSQL handoffs call enqueueMergeQueueInTransaction directly, bypassing the + legacy enqueueMergeQueueSyncInternal spy. Keep this test-only hook dormant in + production so VAL-DATA-013 can inject a late transaction failure and prove every + handoff sub-write rolls back without adding queries or runtime behavior. + */ + private handoffMergeQueueFailureInjectorForTesting: ((taskId: string) => void | Promise) | null = null; + /** True when the mandatory production AsyncDataLayer was injected. */ /** @internal TaskStore decomposition: accessible to extracted modules */ public get backendMode(): boolean { @@ -1112,6 +1121,21 @@ export class TaskStore extends EventEmitter { async handoffToReview(taskId: string, opts: HandoffToReviewOptions): Promise { return handoffToReviewImpl(this, taskId, opts); } + /** + * FNXC:HandoffFailureInjection 2026-07-15-12:00: + * Test-only PostgreSQL handoff seam. Tests arm it after the transaction's + * column, merge-queue, workflow-work, and audit writes so VAL-DATA-013 proves + * they roll back together; null is the strict production no-op. + */ + public __setHandoffMergeQueueFailureInjectorForTesting( + injector: ((taskId: string) => void | Promise) | null, + ): void { + this.handoffMergeQueueFailureInjectorForTesting = injector; + } + /** @internal Invoked only from the late backend handoff transaction seam. */ + public async __invokeHandoffMergeQueueFailureInjectorForTesting(taskId: string): Promise { + await this.handoffMergeQueueFailureInjectorForTesting?.(taskId); + } public resolveWorkflowMoveActor( moveSource: NonNullable, internal: MoveTaskInternalOptions, options?: MoveTaskOptions, ): WorkflowMovePolicyInput["actor"] { return resolveWorkflowMoveActorImpl(this, moveSource, internal, options); } public resolveWorkflowBypassGuards( moveSource: NonNullable, options?: MoveTaskOptions, ): boolean { diff --git a/packages/core/src/task-store/moves.ts b/packages/core/src/task-store/moves.ts index f5a5656ed8..beec1bdb8d 100644 --- a/packages/core/src/task-store/moves.ts +++ b/packages/core/src/task-store/moves.ts @@ -219,11 +219,14 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum agentId: internal.runContext?.agentId, runId: internal.runContext?.runId, }); + // FNXC:PostgresCutover 2026-07-15-12:00: + // Same-column retries must share the outer handoff transaction too, + // so workflow work cannot survive a rolled-back queue/audit handoff. await store.createCompletionHandoffWorkflowWork(task, { runId: internal.runContext?.runId, now: internal.now, source: internal.evidence?.reason, - }); + }, tx); await recordRunAuditEventWithinTransaction(tx, { taskId: id, agentId: internal.runContext?.agentId ?? "system", @@ -241,6 +244,14 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum alreadyEnqueued: existing, }, }); + /* + FNXC:HandoffFailureInjection 2026-07-15-12:00: + Backend handoffs bypass the legacy enqueueMergeQueueSyncInternal spy. + This test-only no-op seam runs after every VAL-DATA-013 sub-write + (move, queue, workflow work, and handoff audit), so an injected throw + proves this transaction rolls all of them back. + */ + await store.__invokeHandoffMergeQueueFailureInjectorForTesting(id); }); return task; } @@ -830,6 +841,14 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum alreadyEnqueued, }, }); + /* + FNXC:HandoffFailureInjection 2026-07-15-12:00: + Backend handoffs bypass the legacy enqueueMergeQueueSyncInternal spy. + This test-only no-op seam runs after every VAL-DATA-013 sub-write + (move, queue, workflow work, and handoff audit), so an injected throw + proves this transaction rolls all of them back. + */ + await store.__invokeHandoffMergeQueueFailureInjectorForTesting(id); } }); } else { diff --git a/packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts b/packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts index 55bdc12e89..4c378aedb6 100644 --- a/packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts @@ -1,36 +1,19 @@ -import { mkdtempSync, readFileSync, rmSync } from "node:fs"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { HandoffInvariantViolationError, TaskStore } from "@fusion/core"; -import { SelfHealingManager } from "../../self-healing.js"; +import { readFileSync } from "node:fs"; +import { afterEach, describe, expect, it } from "vitest"; +import { HandoffInvariantViolationError } from "@fusion/core"; +import { hasGit, hasPg, makeReliabilityFixture } from "./_helpers.js"; -function taskTempDir(): string { - return mkdtempSync(join(tmpdir(), "fn-5241-reliability-")); -} +const describeIfGit = hasGit && hasPg ? describe : describe.skip; +const handoffMutationTypes = ["task:move", "mergeQueue:enqueue", "task:handoff"] as const; -describe("FN-5241 reliability interactions: in-review handoff atomic", () => { - let rootDir: string; - let globalDir: string; - let store: TaskStore; +describeIfGit("FN-5241 reliability interactions: in-review handoff atomic", () => { + const fixtures: Array>> = []; - beforeEach(async () => { - rootDir = taskTempDir(); - globalDir = join(rootDir, ".fusion-global"); - store = new TaskStore(rootDir, globalDir); - await store.init(); + afterEach(async () => { + while (fixtures.length) await fixtures.pop()!.cleanup(); }); - afterEach(() => { - try { - vi.restoreAllMocks(); - store.close(); - } finally { - rmSync(rootDir, { recursive: true, force: true }); - } - }); - - async function createInProgressTask(overrides: Record = {}) { + async function createInProgressTask(store: Awaited>["store"], overrides: Record = {}) { const task = await store.createTask({ description: "handoff reliability", priority: "high" }); await store.moveTask(task.id, "todo"); await store.moveTask(task.id, "in-progress"); @@ -40,30 +23,88 @@ describe("FN-5241 reliability interactions: in-review handoff atomic", () => { return (await store.getTask(task.id))!; } - it("rolls back column move and queue insert when enqueueMergeQueue throws, then succeeds on retry", async () => { - const task = await createInProgressTask(); - vi.spyOn(store as never, "enqueueMergeQueueSyncInternal").mockImplementationOnce((() => { - throw new Error("boom"); - }) as never); + async function handoffAudits(store: Awaited>["store"], taskId: string) { + const events = await store.getRunAuditEventsAsync({ taskId, limit: 50 }); + return events.filter((event) => handoffMutationTypes.includes(event.mutationType as typeof handoffMutationTypes[number])); + } - await expect(store.handoffToReview(task.id, { + it("rolls back every column-change handoff write when the PG seam throws, then succeeds on retry", async () => { + const fx = await makeReliabilityFixture({ taskId: "FN-5241-column-change" }); + fixtures.push(fx); + const task = await createInProgressTask(fx.store); + const auditsBefore = await handoffAudits(fx.store, task.id); + let injectorCallCount = 0; + fx.store.__setHandoffMergeQueueFailureInjectorForTesting((taskId) => { + injectorCallCount += 1; + expect(taskId).toBe(task.id); + throw new Error("boom"); + }); + + await expect(fx.store.handoffToReview(task.id, { ownerAgentId: "executor-agent", evidence: { reason: "fn_task_done", runId: "run-1", agentId: "executor-agent" }, })).rejects.toThrow("boom"); - expect((await store.getTask(task.id))?.column).toBe("in-progress"); - expect(await store.peekMergeQueue()).toHaveLength(0); - expect(store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 20 })).toHaveLength(0); + expect(injectorCallCount).toBe(1); + expect((await fx.store.getTask(task.id))?.column).toBe("in-progress"); + expect(await fx.store.peekMergeQueue()).toHaveLength(0); + expect(await fx.store.listWorkflowWorkItemsForTask(task.id)).toEqual([]); + expect(await handoffAudits(fx.store, task.id)).toEqual(auditsBefore); - await store.handoffToReview(task.id, { + fx.store.__setHandoffMergeQueueFailureInjectorForTesting(null); + await fx.store.handoffToReview(task.id, { ownerAgentId: "executor-agent", evidence: { reason: "fn_task_done", runId: "run-2", agentId: "executor-agent" }, }); - expect((await store.getTask(task.id))?.column).toBe("in-review"); - expect(await store.peekMergeQueue()).toEqual([ + expect((await fx.store.getTask(task.id))?.column).toBe("in-review"); + expect(await fx.store.peekMergeQueue()).toEqual([ expect.objectContaining({ taskId: task.id, priority: task.priority }), ]); + expect(await fx.store.listWorkflowWorkItemsForTask(task.id)).toHaveLength(1); + }); + + it("rolls back every same-column retry write when the PG seam throws, then retries idempotently", async () => { + const fx = await makeReliabilityFixture({ taskId: "FN-5241-same-column" }); + fixtures.push(fx); + const task = await createInProgressTask(fx.store); + const handoff = { + ownerAgentId: "executor-agent", + evidence: { reason: "fn_task_done", runId: "run-1", agentId: "executor-agent" }, + }; + await fx.store.handoffToReview(task.id, handoff); + + const queueBefore = await fx.store.peekMergeQueue(); + const workBefore = await fx.store.listWorkflowWorkItemsForTask(task.id); + const auditsBefore = await handoffAudits(fx.store, task.id); + let injectorCallCount = 0; + fx.store.__setHandoffMergeQueueFailureInjectorForTesting((taskId) => { + injectorCallCount += 1; + expect(taskId).toBe(task.id); + throw new Error("boom"); + }); + + await expect(fx.store.handoffToReview(task.id, { + ...handoff, + evidence: { reason: "fn_task_done", runId: "run-2", agentId: "executor-agent" }, + })).rejects.toThrow("boom"); + + expect(injectorCallCount).toBe(1); + expect((await fx.store.getTask(task.id))?.column).toBe("in-review"); + expect(await fx.store.peekMergeQueue()).toEqual(queueBefore); + expect(await fx.store.listWorkflowWorkItemsForTask(task.id)).toEqual(workBefore); + expect(await handoffAudits(fx.store, task.id)).toEqual(auditsBefore); + + fx.store.__setHandoffMergeQueueFailureInjectorForTesting(null); + await fx.store.handoffToReview(task.id, { + ...handoff, + evidence: { reason: "fn_task_done", runId: "run-3", agentId: "executor-agent" }, + }); + + expect((await fx.store.getTask(task.id))?.column).toBe("in-review"); + expect(await fx.store.peekMergeQueue()).toEqual(queueBefore); + const workAfterRetry = await fx.store.listWorkflowWorkItemsForTask(task.id); + expect(workAfterRetry.filter((item) => item.state === "runnable")).toHaveLength(1); }); it("contains no direct moveTask(..., \"in-review\") writes outside allowlisted same-line comments", () => { @@ -82,43 +123,45 @@ describe("FN-5241 reliability interactions: in-review handoff atomic", () => { }); it("keeps autoMerge-false handoffs parked in in-review with queue state intact across self-healing sweeps", async () => { - await store.updateSettings({ autoMerge: false } as any); - const task = await createInProgressTask(); - await store.handoffToReview(task.id, { + const fx = await makeReliabilityFixture({ taskId: "FN-5241-auto-merge", settings: { autoMerge: false } }); + fixtures.push(fx); + const task = await createInProgressTask(fx.store); + await fx.store.handoffToReview(task.id, { ownerAgentId: "executor-agent", evidence: { reason: "fn_task_done", runId: "run-1", agentId: "executor-agent" }, }); - const manager = new SelfHealingManager(store, { rootDir }); - await manager.recoverCompletionHandoffLimbo(); - expect(await manager.surfaceInReviewStalls()).toBe(0); - expect(await manager.surfaceInReviewStalled()).toBe(0); + await fx.manager.recoverCompletionHandoffLimbo(); + expect(await fx.manager.surfaceInReviewStalls()).toBe(0); + expect(await fx.manager.surfaceInReviewStalled()).toBe(0); - const latest = await store.getTask(task.id); + const latest = await fx.store.getTask(task.id); expect(latest?.column).toBe("in-review"); expect(latest?.paused ?? false).toBe(false); expect(latest?.status ?? null).toBeNull(); - expect(await store.peekMergeQueue()).toEqual([ + expect(await fx.store.peekMergeQueue()).toEqual([ expect.objectContaining({ taskId: task.id }), ]); - expect(store.getRunAuditEvents({ taskId: task.id, limit: 50 }).filter((event) => event.mutationType.startsWith("task:auto-recover"))).toEqual([]); + expect((await fx.store.getRunAuditEventsAsync({ taskId: task.id, limit: 50 })) + .filter((event) => event.mutationType.startsWith("task:auto-recover"))).toEqual([]); }); it("composes no-progress churn terminalization with atomic handoff + queue insertion", async () => { - const task = await createInProgressTask({ stuckKillCount: 2, lineageId: "lin-5241" }); - const manager = new SelfHealingManager(store, { rootDir }); + const fx = await makeReliabilityFixture({ taskId: "FN-5241-churn" }); + fixtures.push(fx); + const task = await createInProgressTask(fx.store, { stuckKillCount: 2, lineageId: "lin-5241" }); - const result = await manager.checkStuckBudget(task.id, "no-progress-churn", { ignoredStepUpdateCount: 25 }); + const result = await fx.manager.checkStuckBudget(task.id, "no-progress-churn", { ignoredStepUpdateCount: 25 }); expect(result).toBe(false); - const latest = await store.getTask(task.id); + const latest = await fx.store.getTask(task.id); expect(latest?.column).toBe("in-review"); expect(latest?.status).toBe("failed"); expect(latest?.error).toMatch(/^STUCK_NO_PROGRESS_CHURN:/); - expect(await store.peekMergeQueue()).toEqual([ + expect(await fx.store.peekMergeQueue()).toEqual([ expect.objectContaining({ taskId: task.id, priority: task.priority }), ]); - const handoff = store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })[0]; + const handoff = (await fx.store.getRunAuditEventsAsync({ taskId: task.id, mutationType: "task:handoff", limit: 10 }))[0]; expect(handoff?.metadata).toMatchObject({ taskId: task.id, reason: "stuck-no-progress-churn", @@ -129,18 +172,17 @@ describe("FN-5241 reliability interactions: in-review handoff atomic", () => { }); it("rejects soft-deleted tasks without creating mergeQueue state", async () => { - const task = await createInProgressTask(); - store.getDatabase().prepare('UPDATE tasks SET "deletedAt" = ? WHERE id = ?').run( - "2026-05-19T00:00:00.000Z", - task.id, - ); + const fx = await makeReliabilityFixture({ taskId: "FN-5241-deleted" }); + fixtures.push(fx); + const task = await createInProgressTask(fx.store); + await fx.store.deleteTask(task.id); - await expect(store.handoffToReview(task.id, { + await expect(fx.store.handoffToReview(task.id, { ownerAgentId: "executor-agent", evidence: { reason: "fn_task_done", runId: "run-1", agentId: "executor-agent" }, })).rejects.toBeInstanceOf(HandoffInvariantViolationError); - expect(await store.peekMergeQueue()).toHaveLength(0); - expect(store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })).toHaveLength(0); + expect(await fx.store.peekMergeQueue()).toHaveLength(0); + expect(await fx.store.getRunAuditEventsAsync({ taskId: task.id, mutationType: "task:handoff", limit: 10 })).toHaveLength(0); }); }); diff --git a/packages/engine/vitest.config.ts b/packages/engine/vitest.config.ts index 1e29b627c1..aca155bb5c 100644 --- a/packages/engine/vitest.config.ts +++ b/packages/engine/vitest.config.ts @@ -379,7 +379,6 @@ export default defineConfig({ // (now PG-backed) but fail on sync SQLite APIs (getRunAuditEvents, getDatabase) that // return [] / throw in backend mode, or on mock drift from the async-satellite cutover. // Quarantined on sight per AGENTS.md; mirrored in scripts/lib/test-quarantine.json. - "src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts", "src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts", "src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts", "src/__tests__/reliability-interactions/integration-worktree-state.test.ts", diff --git a/scripts/lib/test-quarantine.json b/scripts/lib/test-quarantine.json index d34b9e91f1..41324e627a 100644 --- a/scripts/lib/test-quarantine.json +++ b/scripts/lib/test-quarantine.json @@ -126,11 +126,6 @@ "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", "quarantinedAt": "2026-07-14" }, - { - "file": "packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: reliability fixture uses PG-backed store but sync APIs (getRunAuditEvents, getDatabase) fail in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, { "file": "packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts", "reason": "VAL-REMOVAL-005 PG migration: reliability fixture uses PG-backed store but sync APIs (getRunAuditEvents, getDatabase) fail in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.",