root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

FN-8045: add transactional handoff failure-injection seam

Browse Source 
Ensure PostgreSQL review handoffs roll back all dependent writes after an injected late failure.

- Add a test-only failure injector after transactional handoff writes.
- Include workflow work in same-column retry transactions.
- Restore PG-backed handoff atomicity coverage and remove its quarantine.

Files changed:
 packages/core/src/store.ts                         |  24 +++
 packages/core/src/task-store/moves.ts              |  21 ++-
 .../in-review-handoff-atomic.test.ts               | 172 +++++++++++++--------
 packages/engine/vitest.config.ts                   |   1 -
 scripts/lib/test-quarantine.json                   |   5 -
 5 files changed, 151 insertions(+), 72 deletions(-)

Fusion-Task-Id: FN-8045

Fusion-Task-Lineage: 517e3000-9b88-4b0d-9b25-1a585eb8f322

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-16 00:44:35 -07:00
parent c5aa5ecfc7
commit a31c370375
5 changed files with 150 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
packages/core/src/store.ts
View File

@@ -335,6 +335,15 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
public readonly asyncLayer: AsyncDataLayer | null = null;
private pluginPostgresSchemaExecutor: ((contracts: readonly LoadedPluginSchemaContract[]) => Promise<void>) | 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<void>) | 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<TaskStoreEvents> {
async handoffToReview(taskId: string, opts: HandoffToReviewOptions): Promise<Task> {
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<void>) | null,
): void {
this.handoffMergeQueueFailureInjectorForTesting = injector;
}
/** @internal Invoked only from the late backend handoff transaction seam. */
public async __invokeHandoffMergeQueueFailureInjectorForTesting(taskId: string): Promise<void> {
await this.handoffMergeQueueFailureInjectorForTesting?.(taskId);
}
public resolveWorkflowMoveActor( moveSource: NonNullable<MoveTaskOptions["moveSource"]>, internal: MoveTaskInternalOptions, options?: MoveTaskOptions, ): WorkflowMovePolicyInput["actor"] { return resolveWorkflowMoveActorImpl(this, moveSource, internal, options);
}
public resolveWorkflowBypassGuards( moveSource: NonNullable<MoveTaskOptions["moveSource"]>, options?: MoveTaskOptions, ): boolean {

21
packages/core/src/task-store/moves.ts
View File

@@ -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 {

170
packages/engine/src/__tests__/reliability-interactions/in-review-handoff-atomic.test.ts
View File

@@ -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<Awaited<ReturnType<typeof makeReliabilityFixture>>> = [];
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<string, unknown> = {}) {
async function createInProgressTask(store: Awaited<ReturnType<typeof makeReliabilityFixture>>["store"], overrides: Record<string, unknown> = {}) {
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<ReturnType<typeof makeReliabilityFixture>>["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);
});
});

1
packages/engine/vitest.config.ts
View File

@@ -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",

5
scripts/lib/test-quarantine.json
View File

@@ -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.",