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

fix(engine): address workflow cutover review feedback

Browse Source
This commit is contained in:
gsxdsm
2026-06-22 22:17:30 -07:00
parent 65f509e193
commit e60b1378b8
6 changed files with 209 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts
View File

@@ -110,6 +110,27 @@ describe("fast mode workflow/runtime invariants", () => {
);
});
it("falls back to the runner task when prepareWorktree cannot trust the live row", async () => {
const store = createMockStore();
store.getTask.mockResolvedValue({ ...task({ id: "FN-OTHER", worktree: "/tmp/wrong" }) });
const executor = new TaskExecutor(store, "/tmp/test");
const result = await (executor as any)
.createAuthoritativeWorkflowPrimitives({ experimentalFeatures: { workflowGraphExecutor: true } })
.prepareWorktree(
{ run: { taskId: "FN-6226" }, node: { node: { id: "execute" }, context: {} } },
task({ id: "FN-6226", worktree: "/tmp/right", branch: "fusion/fn-6226" }),
);
expect(result).toMatchObject({
outcome: "success",
data: {
worktreePath: "/tmp/right",
branchName: "fusion/fn-6226",
},
});
});
it("graph executor with builtin:coding selection skips the workflow-step seam in fast mode", async () => {
const { executor } = makeExecutorForTask(task({ executionMode: "fast", worktree: "/tmp/wt" }));
const runWorkflowSteps = vi.spyOn(executor as any, "runWorkflowSteps").mockResolvedValue(workflowResult());

44
packages/engine/src/__tests__/hold-release.test.ts
View File

@@ -19,7 +19,7 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { execSync } from "node:child_process";
import { TaskStore, type WorkflowIr } from "@fusion/core";
import { TaskStore, type Task, type WorkflowIr } from "@fusion/core";
import {
runHoldReleaseSweep,
promoteHeldTask,
@@ -104,6 +104,48 @@ describe("hold-release sweep (U6)", () => {
expect((await store.getTask(id))?.column).toBe("in-progress");
});
it("does not let unrelated moved events disable the current task's eventless release fallback", async () => {
const held = {
id: "FN-777",
title: "Held",
description: "",
column: "todo",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as Task;
let onMoved: ((data: { task: Task; to: string }) => void) | undefined;
const release = vi.fn();
const fakeStore = {
getSettings: vi.fn(async () => ({
maxConcurrent: 4,
experimentalFeatures: { workflowColumns: true },
})),
listTasks: vi.fn(async () => [held]),
moveTask: vi.fn(async () => {
onMoved?.({ task: { ...held, id: "FN-OTHER" }, to: "in-progress" });
held.column = "in-progress";
return held;
}),
getTaskWorkflowSelection: vi.fn(() => null),
on: vi.fn((_event: string, listener: (data: { task: Task; to: string }) => void) => {
onMoved = listener;
}),
off: vi.fn(),
} as unknown as TaskStore;
const result = await runHoldReleaseSweep(fakeStore, {
now: () => Date.now(),
reserveSlot: () => ({ release }),
});
expect(result.released).toEqual(["FN-777"]);
expect(release).not.toHaveBeenCalled();
});
it("two holds, one slot: exactly one releases; the other releases next sweep after the slot frees", async () => {
await store.updateSettings({ maxConcurrent: 1 } as Parameters<typeof store.updateSettings>[0]);
const a = await seedTodoCard();

80
packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts
View File

@@ -96,6 +96,86 @@ describe("Scheduler workflow cutover", () => {
expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-100", column: "in-progress" }));
});
it("queues without dispatch when ephemeral agents are disabled and no agent store is available", async () => {
const ready = task({ id: "FN-101" });
const store = storeWith([ready], { ephemeralAgentsEnabled: false });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.updateTask).toHaveBeenCalledWith("FN-101", { status: "queued" });
expect(store.logEntry).toHaveBeenCalledWith(
"FN-101",
"queued — permanent executor selection unavailable (ephemeral agents disabled)",
);
expect(store.moveTask).not.toHaveBeenCalledWith("FN-101", "in-progress", expect.anything());
expect(onSchedule).not.toHaveBeenCalled();
expect(ready.column).toBe("todo");
});
it("passes worktree naming and directory settings to the workflow release allocator", async () => {
const ready = task({ id: "FN-102" });
const store = storeWith([ready], {
worktreeNaming: "task-id",
worktreesDir: "custom-worktrees",
});
const scheduler = new Scheduler(store, { onSchedule: vi.fn() });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
const moveOptions = vi.mocked(store.moveTask).mock.calls[0]?.[2] as {
allocateWorktree?: (reservedNames: Set<string>) => string | null;
};
expect(moveOptions.allocateWorktree?.(new Set())).toBe("/tmp/project/custom-worktrees/fn-102");
});
it("continues executor handoff for all released tasks when post-release metadata or logs fail", async () => {
const first = task({ id: "FN-201", status: "queued" });
const second = task({ id: "FN-202", status: "queued" });
const store = storeWith([first, second], { maxConcurrent: 4, maxWorktrees: 4 });
const updateImpl = vi.mocked(store.updateTask).getMockImplementation()!;
vi.mocked(store.updateTask).mockImplementation(async (id, patch) => {
if (id === "FN-201" && "lastDispatchAt" in patch) {
throw new Error("metadata write failed");
}
return updateImpl(id, patch);
});
vi.mocked(store.logEntry).mockImplementation(async (id, message) => {
if (id === "FN-201" && message.startsWith("Node routing resolved")) {
throw new Error("log write failed");
}
});
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({
id: "FN-201",
column: "in-progress",
status: undefined,
effectiveNodeSource: "local",
}));
expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({
id: "FN-202",
column: "in-progress",
status: undefined,
effectiveNodeSource: "local",
}));
expect(store.updateTask).toHaveBeenCalledWith("FN-202", expect.objectContaining({
status: null,
effectiveNodeSource: "local",
}));
expect(store.logEntry).toHaveBeenCalledWith(
"FN-202",
"Node routing resolved: local (source: local)",
);
});
it("keeps dependency-blocked todo tasks queued on the workflow sweep path", async () => {
const blocker = task({ id: "FN-001", column: "todo" });
const dependent = task({ id: "FN-002", dependencies: ["FN-001"] });

10
packages/engine/src/executor.ts
View File

@@ -5155,14 +5155,18 @@ export class TaskExecutor {
return {
prepareWorktree: async (_ctx, task) => {
const live = await this.store.getTask(task.id);
const live = await this.store.getTask(task.id).catch(() => null);
const liveTask = live?.id === task.id ? live : null;
/*
FNXC:WorkflowExecution 2026-06-23-11:49:
The workflow execute node must not perform a second worktree acquisition ahead of the authoritative executor. Passing the repo root as a prepared worktree makes the inner execute() reject a valid fresh-worktree task as repo-root reuse; pass only an existing task worktree and let execute() acquire when none exists.
FNXC:WorkflowExecution 2026-06-23-22:31:
Upgrade safety requires the graph primitive to tolerate older or minimal stores that return null or a mismatched row during startup/cutover. Only trust the live row when it is for the requested task; otherwise fall back to the runner snapshot.
*/
const prepared: PreparedWorktree = {
worktreePath: live.worktree || task.worktree || "",
branchName: live.branch || task.branch,
worktreePath: liveTask?.worktree || task.worktree || "",
branchName: liveTask?.branch || task.branch,
};
return { outcome: "success", value: "worktree-ready", data: prepared };
},

13
packages/engine/src/hold-release.ts
View File

@@ -429,8 +429,12 @@ async function issueRelease(
// the real mover; any other call that reserved performed a redundant no-op and
// must release the slot it grabbed (FN-1415).
const movedTaskObjects = new Set<object>();
const onMoved = (data: { task: object; to: string }): void => {
if (data.to === target) movedTaskObjects.add(data.task);
let sawMovedEventForTask = false;
const onMoved = (data: { task: Task; to: string }): void => {
if (data.to === target && data.task.id === task.id) {
sawMovedEventForTask = true;
movedTaskObjects.add(data.task);
}
};
store.on?.("task:moved", onMoved);
@@ -446,8 +450,11 @@ async function issueRelease(
/*
FNXC:WorkflowScheduling 2026-06-23-21:57:
The cutover scheduler uses hold/release in tests and older embedded stores that may not expose task:moved events. Treat a returned task that clearly moved from the original column to the target as the committed release so minimal stores do not leak reservations or falsely report a racing same-column no-op.
FNXC:WorkflowScheduling 2026-06-23-22:39:
Eventless-release fallback is scoped to the current task. Other cards moving to the same target column during the same sweep must not disable this task's fallback and leak its reservation.
*/
const returnedMovedTask = movedTaskObjects.size === 0
const returnedMovedTask = !sawMovedEventForTask
&& (
result === undefined
|| (result.id === task.id && result.column === target && originalColumn !== target)

59
packages/engine/src/scheduler.ts
View File

@@ -2376,7 +2376,23 @@ export class Scheduler {
}
}
if (latestSettings.ephemeralAgentsEnabled === false && !freshTask.assignedAgentId && this.options.agentStore) {
if (latestSettings.ephemeralAgentsEnabled === false && !freshTask.assignedAgentId) {
/*
FNXC:WorkflowScheduling 2026-06-23-22:33:
The workflow cutover path must not silently dispatch unassigned work when ephemeral agents are disabled. Queue until permanent-agent selection is available so upgrades preserve the executor contract instead of falling through to local execution.
*/
if (!this.options.agentStore) {
await this.store.updateTask(task.id, { status: "queued" });
if (!this.wasPermanentAgentUnavailable.has(task.id)) {
await this.logDispatchQueuedReason(
task.id,
"queued — permanent executor selection unavailable (ephemeral agents disabled)",
);
this.wasPermanentAgentUnavailable.add(task.id);
}
return null;
}
const selectedAgent = await selectPermanentAgentForTask({
task: freshTask,
agentStore: this.options.agentStore,
@@ -2564,23 +2580,21 @@ export class Scheduler {
};
},
allocateWorktree: (task, reservedNames) =>
planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}),
this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings),
});
for (const taskId of result.released) {
const prep = dispatchPrepByTaskId.get(taskId);
if (!prep) continue;
/*
FNXC:WorkflowScheduling 2026-06-23-21:49:
A workflow hold release is not a committed dispatch until moveTask succeeds and appears in result.released. Only then may the scheduler emit "Starting" and clear queued state. Call onSchedule before best-effort metadata/log writes so a post-release store/log failure does not strand an in-progress task without executor handoff.
A workflow hold release is not a committed dispatch until moveTask succeeds and appears in result.released. Only then may the scheduler emit "Starting" and clear queued state.
FNXC:WorkflowScheduling 2026-06-23-22:36:
Persist dispatch metadata before executor handoff when possible, but isolate update/log failures per task. A metadata failure must not block later released tasks or strand an already released task without onSchedule handoff.
*/
schedulerLog.log(`Starting ${taskId}: ${prep.task.title || taskId} (deps satisfied)`);
const latest = await this.store.getTask(taskId).catch(() => null);
try {
this.options.onSchedule?.(latest ?? prep.task);
} catch (error) {
schedulerLog.error(`onSchedule failed for ${taskId}:`, error);
}
await this.store.updateTask(taskId, {
const dispatchUpdate = {
status: null,
blockedBy: null,
executionStartBranch: prep.baseBranch ?? undefined,
@@ -2589,13 +2603,36 @@ export class Scheduler {
mergeRetries: 0,
dispatchStormCount: prep.dispatchStormCount,
lastDispatchAt: prep.dispatchTimestamp,
});
};
const scheduledTask = {
...(latest?.id === taskId ? latest : prep.task),
...dispatchUpdate,
status: undefined,
blockedBy: undefined,
effectiveNodeId: prep.effectiveNodeId ?? undefined,
effectiveNodeSource: prep.effectiveNodeSource as Task["effectiveNodeSource"],
column: "in-progress" as const,
};
try {
await this.store.updateTask(taskId, dispatchUpdate);
} catch (error) {
schedulerLog.error(`Post-release dispatch metadata update failed for ${taskId}:`, error);
}
try {
this.options.onSchedule?.(scheduledTask);
} catch (error) {
schedulerLog.error(`onSchedule failed for ${taskId}:`, error);
}
this.recentEngineTodoRequeues.delete(taskId);
this.wasNodeBlocked.delete(taskId);
this.wasNodeDispatchValidationBlocked.delete(taskId);
this.wasPermanentAgentUnavailable.delete(taskId);
this.clearDispatchQueuedReasonMemo(taskId);
await this.store.logEntry(taskId, `Node routing resolved: ${prep.effectiveNodeId ?? "local"} (source: ${prep.effectiveNodeSource})`);
try {
await this.store.logEntry(taskId, `Node routing resolved: ${prep.effectiveNodeId ?? "local"} (source: ${prep.effectiveNodeSource})`);
} catch (error) {
schedulerLog.error(`Post-release dispatch log failed for ${taskId}:`, error);
}
}
} catch (error) {
schedulerLog.error("Hold/release sweep failed:", error);