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

feat(FN-5241): add atomic in-review handoff seam with executor/self-healing

Browse Source 
The merge introduces an atomic review handoff seam in the core store (`packages/core/src/store.ts`) and migrates executor and self-healing transitions to use it, replacing the previous multi-step mutable-state handoff with a single transactional operation. Extensive reliability backstops and regress

Fusion-Task-Id: FN-5241
This commit is contained in:
Fusion (runfusion.ai)
2026-05-20 00:54:03 -07:00
committed by gsxdsm
parent b7ddfc9d20
commit 93b11c6c0c
17 changed files with 999 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
packages/engine/src/__tests__/executor-task-done-invariant.test.ts
View File

@@ -1,8 +1,13 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { beforeEach, afterEach, describe, expect, it, vi } from "vitest";
import { mkdtempSync, mkdirSync } from "node:fs";
import { rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import "./executor-test-helpers.js";
import { TaskExecutor } from "../executor.js";
import * as worktreePool from "../worktree-pool.js";
import { createMockStore, mockedCreateFnAgent, mockedExecSync, resetExecutorMocks } from "./executor-test-helpers.js";
import { TaskStore } from "@fusion/core";
import { createMockStore, mockedCreateFnAgent, mockedExec, mockedExecSync, resetExecutorMocks } from "./executor-test-helpers.js";
function baseTask(overrides: Record<string, unknown> = {}) {
return {
@@ -32,6 +37,10 @@ async function setup(overrides: Record<string, unknown> = {}) {
store.moveTask.mockImplementation(async (id: string, column: string) => {
task = { ...task, id, column, paused: false, pausedByAgentId: null, status: null, error: null };
});
store.handoffToReview.mockImplementation(async (id: string) => {
task = { ...task, id, column: "in-review", paused: false, pausedByAgentId: null };
return task;
});
mockedCreateFnAgent.mockImplementation(async ({ customTools }: any) => {
tool = customTools.find((t: any) => t.name === "fn_task_done");
@@ -140,3 +149,126 @@ describe("FN-4114 fn_task_done invariants", () => {
expect(store.updateStep).toHaveBeenCalled();
});
});
describe("FN-5241 executor handoff auditing", () => {
let rootDir: string;
let globalDir: string;
let store: TaskStore;
beforeEach(async () => {
rootDir = mkdtempSync(join(tmpdir(), "fn-5241-executor-"));
globalDir = join(rootDir, ".fusion-global");
store = new TaskStore(rootDir, globalDir);
await store.init();
resetExecutorMocks();
vi.spyOn(worktreePool, "isUsableTaskWorktree").mockResolvedValue(true);
});
afterEach(async () => {
store.close();
await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
});
async function createExecutorTask(taskDoneRetryCount = 0) {
const created = await store.createTask({ description: "Invariant test", priority: "high" });
await store.moveTask(created.id, "todo");
await store.moveTask(created.id, "in-progress");
const worktreePath = join(rootDir, ".worktrees", "swift-falcon");
mkdirSync(worktreePath, { recursive: true });
const branch = `fusion/${created.id.toLowerCase()}`;
await store.updateTask(created.id, {
worktree: worktreePath,
branch,
baseCommitSha: "abc123",
taskDoneRetryCount,
steps: [{ name: "Step 1", status: "in-progress" }],
currentStep: 0,
});
const task = (await store.getTask(created.id))!;
return {
task: {
...task,
prompt: "# Test\n## Steps\n### Step 1: Implement\n- [ ] check",
},
worktreePath,
};
}
it("emits task:handoff and enqueues merge work on successful fn_task_done", async () => {
const { task, worktreePath } = await createExecutorTask();
mockedExec.mockImplementation(((cmd: string, _opts: unknown, cb?: (err: Error | null, stdout: string, stderr: string) => void) => {
if (!cb) return undefined as any;
if (cmd.includes("rev-parse --show-toplevel")) return cb(null, `${worktreePath}\n`, "");
if (cmd.includes("rev-parse --abbrev-ref HEAD")) return cb(null, `${task.branch}\n`, "");
if (cmd.includes("rev-list --count")) return cb(null, "1\n", "");
if (cmd.includes("rev-parse HEAD")) return cb(null, "def456\n", "");
return cb(null, "", "");
}) as any);
mockedCreateFnAgent.mockImplementation(async ({ customTools }: any) => ({
session: {
prompt: vi.fn().mockImplementation(async () => {
const taskDoneTool = customTools.find((tool: any) => tool.name === "fn_task_done");
await taskDoneTool.execute("tool-1", {});
}),
dispose: vi.fn(),
subscribe: vi.fn(),
on: vi.fn(),
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
state: {},
},
}) as any);
const executor = new TaskExecutor(store as any, rootDir);
await executor.execute(task as any);
expect((await store.getTask(task.id))?.column).toBe("in-review");
expect(store.peekMergeQueue()).toEqual([
expect.objectContaining({ taskId: task.id, priority: task.priority }),
]);
const handoff = store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })[0];
expect(handoff?.metadata).toMatchObject({
taskId: task.id,
reason: "fn_task_done",
alreadyEnqueued: false,
});
});
it("emits failed-status handoff auditing when no-fn_task_done retry budget is exhausted", async () => {
const { task, worktreePath } = await createExecutorTask(3);
mockedExec.mockImplementation(((cmd: string, _opts: unknown, cb?: (err: Error | null, stdout: string, stderr: string) => void) => {
if (!cb) return undefined as any;
if (cmd.includes("rev-parse --show-toplevel")) return cb(null, `${worktreePath}\n`, "");
if (cmd.includes("rev-parse --abbrev-ref HEAD")) return cb(null, `${task.branch}\n`, "");
if (cmd.includes("rev-list --count")) return cb(null, "1\n", "");
if (cmd.includes("rev-parse HEAD")) return cb(null, "def456\n", "");
return cb(null, "", "");
}) as any);
mockedCreateFnAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
subscribe: vi.fn(),
on: vi.fn(),
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
state: {},
},
} as any);
const executor = new TaskExecutor(store as any, rootDir);
await executor.execute(task as any);
const latest = await store.getTask(task.id);
expect(latest?.column).toBe("in-review");
expect(latest?.status).toBe("failed");
expect(String(latest?.error ?? "")).toContain("without calling fn_task_done");
expect(store.peekMergeQueue()).toEqual([
expect.objectContaining({ taskId: task.id, priority: task.priority }),
]);
const handoff = store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })[0];
expect(handoff?.metadata).toMatchObject({
taskId: task.id,
reason: "max-task-done-retries-exhausted",
alreadyEnqueued: false,
});
});
});

1
packages/engine/src/__tests__/executor-test-helpers.ts
View File

@@ -321,6 +321,7 @@ export function createMockStore() {
}),
updateTask: vi.fn().mockResolvedValue({}),
moveTask: vi.fn().mockResolvedValue({}),
handoffToReview: vi.fn().mockImplementation(async (id: string) => store.moveTask(id, "in-review")),
mergeTask: vi.fn().mockResolvedValue({}),
createTask: vi.fn().mockImplementation(async (input: Record<string, unknown>) => ({
id: "FN-002",

2
packages/engine/src/__tests__/reliability-interactions/completion-handoff-limbo.test.ts
View File

@@ -28,6 +28,7 @@ function createStore(task: Task) {
return current;
}),
moveTask: vi.fn(async () => undefined),
enqueueMergeQueue: vi.fn(async () => undefined),
logEntry: vi.fn(async () => undefined),
recordRunAuditEvent: vi.fn(async () => undefined),
_get: () => current,
@@ -52,6 +53,7 @@ describe("FN-4999 reliability interactions: completion-handoff-limbo", () => {
expect(requeueForAutoMerge).toHaveBeenCalledTimes(1);
expect(requeueForAutoMerge).toHaveBeenCalledWith("FN-4999-T");
expect(store.enqueueMergeQueue).toHaveBeenCalledWith("FN-4999-T");
expect(store.logEntry).toHaveBeenCalledWith("FN-4999-T", expect.stringMatching(/Auto-recovered \(FN-4999\)/));
expect(store.moveTask).not.toHaveBeenCalled();
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({

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

@@ -0,0 +1,146 @@
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";
function taskTempDir(): string {
return mkdtempSync(join(tmpdir(), "fn-5241-reliability-"));
}
describe("FN-5241 reliability interactions: in-review handoff atomic", () => {
let rootDir: string;
let globalDir: string;
let store: TaskStore;
beforeEach(async () => {
rootDir = taskTempDir();
globalDir = join(rootDir, ".fusion-global");
store = new TaskStore(rootDir, globalDir);
await store.init();
});
afterEach(() => {
try {
vi.restoreAllMocks();
store.close();
} finally {
rmSync(rootDir, { recursive: true, force: true });
}
});
async function createInProgressTask(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");
if (Object.keys(overrides).length > 0) {
await store.updateTask(task.id, overrides as any);
}
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, "enqueueMergeQueue").mockImplementationOnce(() => {
throw new Error("boom");
});
await expect(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(store.peekMergeQueue()).toHaveLength(0);
expect(store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 20 })).toHaveLength(0);
await 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(store.peekMergeQueue()).toEqual([
expect.objectContaining({ taskId: task.id, priority: task.priority }),
]);
});
it("contains no direct moveTask(..., \"in-review\") writes outside allowlisted same-line comments", () => {
const regex = /moveTask\([^\n]+,\s*"in-review"\)/g;
for (const path of [
new URL("../../executor.ts", import.meta.url),
new URL("../../self-healing.ts", import.meta.url),
]) {
const source = readFileSync(path, "utf8");
const offenders = source
.split("\n")
.filter((line) => regex.test(line) && !/\/\/ handoff-invariant-violation-allowlist: .+/.test(line));
expect(offenders).toEqual([]);
regex.lastIndex = 0;
}
});
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, {
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);
const latest = await store.getTask(task.id);
expect(latest?.column).toBe("in-review");
expect(latest?.paused ?? false).toBe(false);
expect(latest?.status ?? null).toBeNull();
expect(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([]);
});
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 result = await manager.checkStuckBudget(task.id, "no-progress-churn", { ignoredStepUpdateCount: 25 });
expect(result).toBe(false);
const latest = await store.getTask(task.id);
expect(latest?.column).toBe("in-review");
expect(latest?.status).toBe("failed");
expect(latest?.error).toMatch(/^STUCK_NO_PROGRESS_CHURN:/);
expect(store.peekMergeQueue()).toEqual([
expect.objectContaining({ taskId: task.id, priority: task.priority }),
]);
const handoff = store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })[0];
expect(handoff?.metadata).toMatchObject({
taskId: task.id,
reason: "stuck-no-progress-churn",
agentId: "self-healing",
ownerAgentId: null,
alreadyEnqueued: false,
});
});
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,
);
await expect(store.handoffToReview(task.id, {
ownerAgentId: "executor-agent",
evidence: { reason: "fn_task_done", runId: "run-1", agentId: "executor-agent" },
})).rejects.toBeInstanceOf(HandoffInvariantViolationError);
expect(store.peekMergeQueue()).toHaveLength(0);
expect(store.getRunAuditEvents({ taskId: task.id, mutationType: "task:handoff", limit: 10 })).toHaveLength(0);
});
});

10
packages/engine/src/__tests__/reliability-interactions/non-progress-churn.test.ts
View File

@@ -31,6 +31,11 @@ function createStore(task: Task, settings: Record<string, unknown> = {}): TaskSt
task.column = column as any;
task.updatedAt = new Date(Date.now()).toISOString();
});
(emitter as any).handoffToReview = vi.fn().mockImplementation(async (_taskId: string, _opts: any) => {
task.column = "in-review" as any;
task.updatedAt = new Date(Date.now()).toISOString();
return task;
});
(emitter as any).logEntry = vi.fn().mockImplementation(async (_taskId: string, action: string) => {
task.log = task.log ?? [];
task.log.push({ timestamp: new Date(Date.now()).toISOString(), action });
@@ -126,7 +131,10 @@ describe("reliability interactions: non-progress churn", () => {
expect(task.status).toBe("failed");
expect(task.column).toBe("in-review");
expect(task.error).toMatch(/^STUCK_NO_PROGRESS_CHURN:/);
expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-review");
expect(store.handoffToReview).toHaveBeenCalledWith(task.id, expect.objectContaining({
ownerAgentId: null,
evidence: expect.objectContaining({ reason: "stuck-no-progress-churn", agentId: "self-healing" }),
}));
manager.stop();
});

4
packages/engine/src/__tests__/restart.integration.test.ts
View File

@@ -270,7 +270,7 @@ const DEFAULT_SETTINGS: Settings = {
function createMockStore(overrides: Record<string, any> = {}) {
const listeners = new Map<string, Function[]>();
return {
const store = {
on: vi.fn((event: string, fn: Function) => {
const existing = listeners.get(event) || [];
existing.push(fn);
@@ -307,6 +307,8 @@ function createMockStore(overrides: Record<string, any> = {}) {
_listeners: listeners,
...overrides,
} as any;
store.handoffToReview ??= vi.fn().mockImplementation(async (id: string) => store.moveTask(id, "in-review"));
return store;
}
function makeTask(id: string, column: Column, overrides: Partial<Task> = {}): Task {

5
packages/engine/src/__tests__/self-healing-pr-conflict.test.ts
View File

@@ -56,6 +56,11 @@ function makeStore(
task.column = column;
return task;
}),
handoffToReview: vi.fn(async (id: string) => {
if (!task || id !== task.id) return null;
task.column = "in-review";
return task;
}),
logEntry: vi.fn(async () => undefined),
appendAgentLog: vi.fn(async () => undefined),
clearStaleExecutionStartBranchReferences: vi.fn(() => []),

3
packages/engine/src/__tests__/self-healing-reclaim-paused-review.test.ts
View File

@@ -11,6 +11,9 @@ function createStore(): TaskStore & EventEmitter {
(emitter as any).listTasks = vi.fn();
(emitter as any).updateTask = vi.fn().mockResolvedValue(undefined);
(emitter as any).moveTask = vi.fn().mockResolvedValue(undefined);
(emitter as any).handoffToReview = vi.fn().mockImplementation(async (taskId: string) => {
await (emitter as any).moveTask(taskId, "in-review");
});
(emitter as any).logEntry = vi.fn().mockResolvedValue(undefined);
(emitter as any).recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
return emitter;

28
packages/engine/src/__tests__/self-healing.test.ts
View File

@@ -153,6 +153,8 @@ function createMockStore(overrides: Record<string, unknown> = {}): TaskStore & E
updateTask: vi.fn().mockResolvedValue({} as Task),
logEntry: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(undefined),
handoffToReview: vi.fn().mockResolvedValue(undefined),
enqueueMergeQueue: vi.fn().mockResolvedValue(undefined),
mergeTask: vi.fn().mockResolvedValue(undefined),
archiveTaskAndCleanup: vi.fn().mockResolvedValue({} as Task),
walCheckpoint: vi.fn().mockReturnValue({ busy: 0, log: 5, checkpointed: 5 }),
@@ -398,7 +400,10 @@ describe("SelfHealingManager", () => {
status: "failed",
error: "STUCK_LOOP_EXHAUSTED: stuck kill budget exhausted (7/6) after last reason=loop.",
});
expect(store.moveTask).toHaveBeenLastCalledWith("FN-001", "in-review");
expect(store.handoffToReview).toHaveBeenLastCalledWith("FN-001", expect.objectContaining({
ownerAgentId: null,
evidence: expect.objectContaining({ reason: "stuck-loop-exhausted", agentId: "self-healing" }),
}));
expect(store.logEntry).toHaveBeenLastCalledWith(
"FN-001",
"STUCK_LOOP_EXHAUSTED: stuck kill budget exhausted (7/6), last reason=loop. No further automatic retries will run. Manually retry, pause, or move the task to triage to resume work.",
@@ -423,7 +428,10 @@ describe("SelfHealingManager", () => {
status: "failed",
error: "STUCK_NO_PROGRESS_CHURN: detected 25 ignored step-update rebuffs after compact-and-resume failed to recover progress. Task is likely too large; decompose via fn_task_create child tasks or rescope. No further automatic retries will run.",
});
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");
expect(store.handoffToReview).toHaveBeenCalledWith("FN-001", expect.objectContaining({
ownerAgentId: null,
evidence: expect.objectContaining({ reason: "stuck-no-progress-churn", agentId: "self-healing" }),
}));
expect(store.logEntry).toHaveBeenCalledWith(
"FN-001",
"STUCK_NO_PROGRESS_CHURN: detected 25 ignored step-update rebuffs after compact-and-resume failed to recover progress. No further automatic retries will run. Pause the task, manually decompose the work via fn_task_create child tasks, or move it to triage to rescope.",
@@ -477,7 +485,7 @@ describe("SelfHealingManager", () => {
id: "FN-001",
stuckKillCount: 6,
} as unknown as Task);
(store.moveTask as ReturnType<typeof vi.fn>).mockRejectedValue(new Error("concurrent move"));
(store.handoffToReview as ReturnType<typeof vi.fn>).mockRejectedValue(new Error("concurrent move"));
manager.start();
@@ -490,7 +498,7 @@ describe("SelfHealingManager", () => {
error: "STUCK_LOOP_EXHAUSTED: stuck kill budget exhausted (7/6) after last reason=loop.",
});
expect(getSelfHealingLogger().warn).toHaveBeenCalledWith(
expect.stringContaining("moveTask(\"in-review\") failed (concurrent move)"),
expect.stringContaining("handoffTaskToReview failed (concurrent move)"),
);
});
@@ -7424,7 +7432,9 @@ describe("SelfHealingManager reclaimSelfOwnedBranchConflicts", () => {
paused: true,
pausedReason: "branch-conflict-unrecoverable",
}));
expect(store.moveTask).toHaveBeenCalledWith("FN-503", "in-review");
expect(store.handoffToReview).toHaveBeenCalledWith("FN-503", expect.objectContaining({
evidence: expect.objectContaining({ reason: "branch-conflict-unrecoverable-repromote" }),
}));
});
it("preserves dirty worktree as recovery patch before unrecoverable escalation", async () => {
@@ -7448,7 +7458,9 @@ describe("SelfHealingManager reclaimSelfOwnedBranchConflicts", () => {
const recovered = await manager.reclaimSelfOwnedBranchConflicts();
expect(recovered).toBe(0);
expect(store.moveTask).toHaveBeenCalledWith("FN-504", "in-review");
expect(store.handoffToReview).toHaveBeenCalledWith("FN-504", expect.objectContaining({
evidence: expect.objectContaining({ reason: "branch-conflict-unrecoverable-repromote" }),
}));
const recoveryDir = join(fixtureRoot, ".fusion", "recovery");
const files = await readdir(recoveryDir);
@@ -7472,7 +7484,9 @@ describe("SelfHealingManager reclaimSelfOwnedBranchConflicts", () => {
paused: true,
pausedReason: "branch-conflict-unrecoverable",
}));
expect(store.moveTask).toHaveBeenCalledWith("FN-502", "in-review");
expect(store.handoffToReview).toHaveBeenCalledWith("FN-502", expect.objectContaining({
evidence: expect.objectContaining({ reason: "branch-conflict-unrecoverable-repromote" }),
}));
});
});