From fa4445138a9c6721019bb75b034e2e595180ba2b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 17:16:55 -0700 Subject: [PATCH] feat(FN-000): run workflow work items through runtime Fusion-Task-Id: FN-000 --- .../s05-runtime-work-item-driver.md | 46 ++++++++ .../__tests__/workflow-task-runtime.test.ts | 109 +++++++++++++++++- packages/engine/src/workflow-node-handlers.ts | 24 ++++ packages/engine/src/workflow-task-runtime.ts | 100 +++++++++++++++- 4 files changed, 275 insertions(+), 4 deletions(-) create mode 100644 docs/plans/workflow-owned-merge-stack/s05-runtime-work-item-driver.md diff --git a/docs/plans/workflow-owned-merge-stack/s05-runtime-work-item-driver.md b/docs/plans/workflow-owned-merge-stack/s05-runtime-work-item-driver.md new file mode 100644 index 0000000000..ac433eba34 --- /dev/null +++ b/docs/plans/workflow-owned-merge-stack/s05-runtime-work-item-driver.md @@ -0,0 +1,46 @@ +--- +title: "S05: runtime work-item driver" +type: refactor +status: draft-stack-handoff +date: 2026-06-09 +slice: S05 +milestone: "Runtime" +origin: docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md +stack_base: feature/workflow-owned-merge-s04-builtin-ir-regions +--- + +# S05: runtime work-item driver + +## Stack Role + +This draft PR reserves the S05 review slot in the workflow-owned merge, +retry, scheduling, and recovery migration stack. It is intentionally a handoff +artifact, not the completed implementation for this slice. + +## Milestone + +Runtime + +## Depends On + +S1 workflow work items, S3 generic scheduler claim path, and S4 built-in IR regions. + +## Goal + +Let WorkflowTaskRuntime start from a workflow work item and persist node/work-item outcomes. + +## Expected File Scope + +packages/engine/src/workflow-task-runtime.ts; workflow graph executor and node handler files; runtime tests. + +## Expected Tests + +Runnable completion, retrying work creation, manual hold creation, restart resume, and duplicate lease refusal. + +## Exit Gate + +Runtime can progress workflow work without old merge queue callbacks. + +## Full Plan + +See `docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md`. diff --git a/packages/engine/src/__tests__/workflow-task-runtime.test.ts b/packages/engine/src/__tests__/workflow-task-runtime.test.ts index ea4d7282d4..6edb205b2a 100644 --- a/packages/engine/src/__tests__/workflow-task-runtime.test.ts +++ b/packages/engine/src/__tests__/workflow-task-runtime.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import type { Settings, TaskDetail, WorkflowIr } from "@fusion/core"; +import type { Settings, TaskDetail, WorkflowIr, WorkflowWorkItem, WorkflowWorkItemState } from "@fusion/core"; import { WorkflowTaskRuntime, type WorkflowTaskRuntimeDeps } from "../workflow-task-runtime.js"; import type { WorkflowNodeResult } from "../workflow-graph-executor.js"; @@ -193,7 +193,17 @@ describe("WorkflowTaskRuntime", () => { expect(result.disposition).toBe("completed"); expect(calls).toEqual(["planning", "prepare-worktree", "execute", "workflow-step", "review", "merge"]); - expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "workflow-step", "review", "merge"]); + expect(result.visitedNodeIds).toEqual([ + "start", + "planning", + "execute", + "workflow-step", + "review", + "merge-gate", + "branch-group-member-integration", + "branch-group-promotion", + "merge-attempt", + ]); }); it("stops the built-in workflow before review when workflow-step remediation is scheduled", async () => { @@ -306,6 +316,101 @@ describe("WorkflowTaskRuntime", () => { expect(observedRunIds).toContain("FN-9002:WF-001"); }); + it("runs a leased workflow work item at its addressed node and persists success", async () => { + const calls: string[] = []; + const transitions: Array<{ id: string; state: WorkflowWorkItemState; patch?: Record }> = []; + const runtime = new WorkflowTaskRuntime({ + store: { + getTask: async () => task, + getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), + getWorkflowDefinition: async () => ({ ir: selectedIr() }), + transitionWorkflowWorkItem: (id, state, patch) => { + transitions.push({ id, state, patch }); + return { ...workItem, state }; + }, + }, + primitives: recordingPrimitives(calls), + runCustomNode: async (node) => { + calls.push(`custom:${node.id}`); + return { outcome: "success" }; + }, + }); + const workItem = { + id: "work-1", + runId: "run-1", + taskId: task.id, + nodeId: "execute", + kind: "task", + 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; + + const result = await runtime.runWorkItem(workItem, flagOff); + + expect(result.disposition).toBe("completed"); + expect(calls).toEqual(["prepare-worktree", "execute"]); + expect(result.visitedNodeIds).toEqual(["execute"]); + expect(transitions).toEqual([ + { + id: "work-1", + state: "succeeded", + patch: { leaseOwner: null, leaseExpiresAt: null, lastError: null }, + }, + ]); + }); + + it("fails and releases a workflow work item when the addressed node fails", async () => { + const transitions: Array<{ id: string; state: WorkflowWorkItemState; patch?: Record }> = []; + const workItem = { + id: "work-2", + runId: "run-1", + taskId: task.id, + nodeId: "execute", + kind: "task", + 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; + const runtime = new WorkflowTaskRuntime({ + store: { + getTask: async () => task, + getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), + getWorkflowDefinition: async () => ({ ir: selectedIr() }), + transitionWorkflowWorkItem: (id, state, patch) => { + transitions.push({ id, state, patch }); + return { ...workItem, state }; + }, + }, + primitives: recordingPrimitives([], { execute: { outcome: "failure", value: "implementation-incomplete" } }), + runCustomNode: async () => ({ outcome: "success" }), + }); + + const result = await runtime.runWorkItem(workItem, flagOff); + + expect(result.disposition).toBe("failed"); + expect(result.reason).toBe("implementation-incomplete"); + expect(transitions).toEqual([ + { + id: "work-2", + state: "failed", + patch: { leaseOwner: null, leaseExpiresAt: null, lastError: "implementation-incomplete" }, + }, + ]); + }); + 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 612614f57e..091fdc6f64 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -875,6 +875,13 @@ export function createDefaultNodeHandlers( | "parse-steps" | "code" | "notify" + | "merge-gate" + | "merge-attempt" + | "manual-merge-hold" + | "retry-backoff" + | "recovery-router" + | "branch-group-member-integration" + | "branch-group-promotion" | "pr-create" | "pr-respond" | "pr-merge", @@ -918,6 +925,23 @@ export function createDefaultNodeHandlers( "parse-steps": parseSteps, code: createCodeNodeHandler(deps?.runCode), notify: createNotifyHandler(deps?.notifyDispatch), + "merge-gate": async (_node, ctx) => ({ + outcome: "success", + value: ctx.context.autoMerge === false ? "auto-off" : "auto-on", + }), + "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); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + }, + "manual-merge-hold": async () => ({ outcome: "failure", value: "manual-required" }), + "retry-backoff": async () => ({ outcome: "success" }), + "recovery-router": async (_node, ctx) => ({ + outcome: "success", + value: typeof ctx.context.recoveryOutcome === "string" ? ctx.context.recoveryOutcome : "wake-merge", + }), + "branch-group-member-integration": async () => ({ outcome: "success" }), + "branch-group-promotion": async () => ({ outcome: "success" }), ...prNodes, }; } diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index ecf5d15b2d..eacdb4b7ae 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -1,4 +1,4 @@ -import type { Settings, TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core"; +import type { Settings, TaskDetail, WorkflowIr, WorkflowIrNode, WorkflowWorkItem, WorkflowWorkItemState } from "@fusion/core"; import { BUILTIN_CODING_WORKFLOW_IR, getBuiltinWorkflow, @@ -31,7 +31,14 @@ export interface WorkflowTaskRuntimeResult { } export interface WorkflowTaskRuntimeDeps extends Omit { - store: WorkflowIrResolverStore; + store: WorkflowIrResolverStore & { + getTask?: (taskId: string) => Promise; + transitionWorkflowWorkItem?: ( + id: string, + state: WorkflowWorkItemState, + patch?: { now?: string; lastError?: string | null; leaseOwner?: string | null; leaseExpiresAt?: string | null }, + ) => WorkflowWorkItem; + }; primitives: WorkflowRuntimePrimitives; runCustomNode: WorkflowCustomNodeRunner; onEvent?: (event: { type: "start" | "terminal"; taskId: string; detail: string }) => void; @@ -114,6 +121,95 @@ export class WorkflowTaskRuntime { }; } + public async runWorkItem( + workItem: WorkflowWorkItem, + settings: (Pick & Partial) | undefined, + ): Promise { + 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 { + task = await this.deps.store.getTask(workItem.taskId); + } catch (err) { + return this.failWorkItem(workItem, `workflow-work-item-task-missing:${err instanceof Error ? err.message : String(err)}`); + } + + let target: WorkflowRuntimeTarget; + try { + target = await this.resolveRuntimeTarget(workItem.taskId); + } catch (err) { + return this.failWorkItem(workItem, `workflow-resolution-error: ${err instanceof Error ? err.message : String(err)}`); + } + + const node = target.ir.nodes.find((candidate) => candidate.id === workItem.nodeId); + if (!node) { + return this.failWorkItem(workItem, `workflow-work-item-node-missing:${workItem.nodeId}`); + } + + const invoked: string[] = []; + const handler = this.recordingHandlers(invoked)[node.kind]; + if (!handler && node.kind !== "start" && node.kind !== "end") { + return this.failWorkItem(workItem, `workflow-work-item-node-unhandled:${node.kind}`); + } + + const runtimeSettings = forceWorkflowGraphExecutor(settings); + let outcome: WorkflowNodeOutcome = "success"; + let reason: string | undefined; + let context: Record = { + "workflow:work-item-id": workItem.id, + "workflow:work-item-kind": workItem.kind, + }; + + try { + const result = handler + ? await handler(node, { task, settings: runtimeSettings, context }) + : { outcome: "success" as const }; + outcome = result.outcome; + if (result.value !== undefined) context[`node:${node.id}:value`] = result.value; + context = { ...context, ...(result.contextPatch ?? {}) }; + reason = result.outcome === "failure" ? result.value ?? "workflow-work-item-node-failed" : undefined; + } catch (err) { + outcome = "failure"; + reason = `workflow-work-item-node-error:${err instanceof Error ? err.message : String(err)}`; + } + + const disposition: WorkflowTaskRuntimeDisposition = outcome === "success" ? "completed" : "failed"; + this.deps.store.transitionWorkflowWorkItem(workItem.id, disposition === "completed" ? "succeeded" : "failed", { + leaseOwner: null, + leaseExpiresAt: null, + lastError: reason ?? null, + }); + this.emit("terminal", workItem.taskId, `work-item:${disposition}`); + return { + disposition, + outcome, + visitedNodeIds: invoked.length > 0 ? invoked : [node.id], + context, + reason, + }; + } + + private failWorkItem(workItem: WorkflowWorkItem, reason: string): WorkflowTaskRuntimeResult { + this.deps.store.transitionWorkflowWorkItem?.(workItem.id, "failed", { + leaseOwner: null, + leaseExpiresAt: null, + lastError: reason, + }); + this.emit("terminal", workItem.taskId, `work-item:failed:${reason}`); + return { + disposition: "failed", + outcome: "failure", + visitedNodeIds: [], + context: {}, + reason, + }; + } + private async resolveRuntimeTarget(taskId: string): Promise { let workflowId: string | undefined; try {