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

feat(FN-000): run workflow work items through runtime

Browse Source 
Fusion-Task-Id: FN-000
This commit is contained in:
gsxdsm
2026-06-09 17:16:55 -07:00
parent 265496d0b6
commit fa4445138a
4 changed files with 275 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
packages/engine/src/__tests__/workflow-task-runtime.test.ts
View File

@@ -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<string, unknown> }> = [];
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<string, unknown> }> = [];
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({

24
packages/engine/src/workflow-node-handlers.ts
View File

@@ -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,
};
}

100
packages/engine/src/workflow-task-runtime.ts
View File

@@ -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<WorkflowGraphExecutorDeps, "seams" | "runCustomNode"> {
store: WorkflowIrResolverStore;
store: WorkflowIrResolverStore & {
getTask?: (taskId: string) => Promise<TaskDetail>;
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<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
): Promise<WorkflowTaskRuntimeResult> {
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<string, unknown> = {
"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<WorkflowRuntimeTarget> {
let workflowId: string | undefined;
try {