fix(FN-000): harden workflow work item dispatch

Address PR #1578 feedback by failing unwired work item dispatch without a no-op persistence path and forwarding work item attempts into merge primitive context.
This commit is contained in:
gsxdsm
2026-06-11 08:30:31 -07:00
parent c7e278c3db
commit 3af587af6e
3 changed files with 95 additions and 7 deletions

View File

@@ -32,7 +32,7 @@ function recordingPrimitives(
overrides: Partial<Record<"prepare" | "execute" | "workflowStep", WorkflowNodeResult>> & {
prepareData?: PreparedWorktree | null;
} = {},
observed: { prepared?: PreparedWorktree } = {},
observed: { prepared?: PreparedWorktree; mergeAttempt?: number } = {},
): WorkflowRuntimePrimitives {
const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" };
return {
@@ -92,8 +92,9 @@ function recordingPrimitives(
calls.push("schedule");
return { outcome: "success" };
},
requestMerge: async () => {
requestMerge: async (ctx) => {
calls.push("merge");
observed.mergeAttempt = ctx.node.attempt;
return { outcome: "success", value: "merged", data: { status: "merged" } };
},
abortRun: async () => ({ outcome: "success" }),
@@ -501,6 +502,81 @@ describe("WorkflowTaskRuntime", () => {
]);
});
it("returns failed without persisting when work item store transitions are unwired", async () => {
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => undefined,
getWorkflowDefinition: async () => undefined,
},
primitives: recordingPrimitives([]),
runCustomNode: async () => ({ outcome: "success" }),
});
const workItem = {
id: "work-unwired",
runId: "run-unwired",
taskId: task.id,
nodeId: "merge-gate",
kind: "merge",
state: "running",
attempt: 0,
retryAfter: null,
leaseOwner: "scheduler-a",
leaseExpiresAt: "2026-06-09T00:01:00.000Z",
lastError: null,
blockedReason: null,
createdAt: "2026-06-09T00:00:00.000Z",
updatedAt: "2026-06-09T00:00:00.000Z",
} satisfies WorkflowWorkItem;
await expect(runtime.runWorkItem(workItem, flagOff)).resolves.toEqual(expect.objectContaining({
disposition: "failed",
reason: "workflow-work-item-store-unwired",
}));
});
it("threads work item attempt into merge primitive context", async () => {
const observed: { mergeAttempt?: number } = {};
const transitions: Array<{ id: string; state: WorkflowWorkItemState; patch?: Record<string, unknown> }> = [];
const workItem = {
id: "work-merge-attempt",
runId: "run-merge-attempt",
taskId: task.id,
nodeId: "merge-attempt",
kind: "merge",
state: "running",
attempt: 3,
retryAfter: null,
leaseOwner: "scheduler-a",
leaseExpiresAt: "2026-06-09T00:01:00.000Z",
lastError: null,
blockedReason: null,
createdAt: "2026-06-09T00:00:00.000Z",
updatedAt: "2026-06-09T00:00:00.000Z",
} satisfies WorkflowWorkItem;
const runtime = new WorkflowTaskRuntime({
store: {
getTask: async () => task,
getTaskWorkflowSelection: () => undefined,
getWorkflowDefinition: async () => undefined,
transitionWorkflowWorkItem: (id, state, patch) => {
transitions.push({ id, state, patch });
return { ...workItem, state };
},
},
primitives: recordingPrimitives([], {}, observed),
runCustomNode: async () => ({ outcome: "success" }),
});
const result = await runtime.runWorkItem(workItem, flagOff);
expect(result.disposition).toBe("completed");
expect(result.context["workflow:work-item-attempt"]).toBe(3);
expect(observed.mergeAttempt).toBe(3);
expect(transitions).toEqual([
expect.objectContaining({ id: "work-merge-attempt", state: "succeeded" }),
]);
});
it("uses the built-in workflow id in the default run id for unselected tasks", async () => {
const observedRunIds: string[] = [];
const runtime = new WorkflowTaskRuntime({

View File

@@ -935,7 +935,10 @@ export function createDefaultNodeHandlers(
},
"merge-attempt": async (_node, ctx) => {
if (!deps?.primitives) return { outcome: "failure", value: "merge-primitives-unwired" };
const result = await deps.primitives.requestMerge(primitiveContextForNode(_node, ctx.task, ctx.context), ctx.task);
const attempt = typeof ctx.context["workflow:work-item-attempt"] === "number"
? ctx.context["workflow:work-item-attempt"]
: undefined;
const result = await deps.primitives.requestMerge(primitiveContextForNode(_node, ctx.task, ctx.context, attempt), ctx.task);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
},
"manual-merge-hold": async () => ({ outcome: "failure", value: "manual-required" }),

View File

@@ -125,12 +125,20 @@ export class WorkflowTaskRuntime {
workItem: WorkflowWorkItem,
settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
): Promise<WorkflowTaskRuntimeResult> {
if (!this.deps.store.getTask || !this.deps.store.transitionWorkflowWorkItem) {
const reason = "workflow-work-item-store-unwired";
this.emit("terminal", workItem.taskId, `work-item:failed:${reason}`);
return {
disposition: "failed",
outcome: "failure",
visitedNodeIds: [],
context: {},
reason,
};
}
if (workItem.state !== "running") {
return this.failWorkItem(workItem, `workflow-work-item-not-running:${workItem.state}`);
}
if (!this.deps.store.getTask || !this.deps.store.transitionWorkflowWorkItem) {
return this.failWorkItem(workItem, "workflow-work-item-store-unwired");
}
let task: TaskDetail;
try {
@@ -163,6 +171,7 @@ export class WorkflowTaskRuntime {
let context: Record<string, unknown> = {
"workflow:work-item-id": workItem.id,
"workflow:work-item-kind": workItem.kind,
"workflow:work-item-attempt": workItem.attempt,
};
try {
@@ -204,7 +213,7 @@ export class WorkflowTaskRuntime {
}
private failWorkItem(workItem: WorkflowWorkItem, reason: string): WorkflowTaskRuntimeResult {
this.deps.store.transitionWorkflowWorkItem?.(workItem.id, "failed", {
this.deps.store.transitionWorkflowWorkItem!(workItem.id, "failed", {
leaseOwner: null,
leaseExpiresAt: null,
lastError: reason,