diff --git a/packages/engine/src/__tests__/workflow-task-runtime.test.ts b/packages/engine/src/__tests__/workflow-task-runtime.test.ts index b37953e171..14b7c847e5 100644 --- a/packages/engine/src/__tests__/workflow-task-runtime.test.ts +++ b/packages/engine/src/__tests__/workflow-task-runtime.test.ts @@ -32,7 +32,7 @@ function recordingPrimitives( overrides: Partial> & { 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 }> = []; + 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({ diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 9d7fcc4378..c6107d3b94 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -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" }), diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index 9d64177c90..73c00b5e06 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -125,12 +125,20 @@ export class WorkflowTaskRuntime { workItem: WorkflowWorkItem, settings: (Pick & Partial) | undefined, ): Promise { + 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 = { "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,