Merge pull request #1578 from Runfusion/feature/workflow-owned-merge-s05-runtime-work-item-driver

refactor(workflow): S05 runtime work-item driver

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`.

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";
@@ -32,7 +32,13 @@ function recordingPrimitives(
   overrides: Partial<Record<"prepare" | "execute" | "workflowStep", WorkflowNodeResult>> & {
     prepareData?: PreparedWorktree | null;
   } = {},
-  observed: { prepared?: PreparedWorktree; executedTasks?: TaskDetail[] } = {},
+  observed: {
+    prepared?: PreparedWorktree;
+    executedTasks?: TaskDetail[];
+    mergeAttempt?: number;
+    mergeRunId?: string;
+    mergeWorkflowId?: string;
+  } = {},
 ): WorkflowRuntimePrimitives {
   const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" };
   return {
@@ -93,8 +99,11 @@ function recordingPrimitives(
       calls.push("schedule");
       return { outcome: "success" };
     },
-    requestMerge: async () => {
+    requestMerge: async (ctx) => {
       calls.push("merge");
+      observed.mergeAttempt = ctx.node.attempt;
+      observed.mergeRunId = ctx.run.runId;
+      observed.mergeWorkflowId = ctx.run.workflowId;
       return { outcome: "success", value: "merged", data: { status: "merged" } };
     },
     abortRun: async () => ({ outcome: "success" }),
@@ -398,6 +407,268 @@ 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("routes merge-gate work items off when task auto-merge is disabled", async () => {
+    const transitions: Array<{ id: string; state: WorkflowWorkItemState; patch?: Record<string, unknown> }> = [];
+    const workItem = {
+      id: "work-merge-gate",
+      runId: "run-merge-gate",
+      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;
+    const runtime = new WorkflowTaskRuntime({
+      store: {
+        getTask: async () => ({ ...task, autoMerge: false } as TaskDetail),
+        getTaskWorkflowSelection: () => undefined,
+        getWorkflowDefinition: async () => undefined,
+        transitionWorkflowWorkItem: (id, state, patch) => {
+          transitions.push({ id, state, patch });
+          return { ...workItem, state };
+        },
+      },
+      primitives: recordingPrimitives([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+
+    const result = await runtime.runWorkItem(workItem, { ...flagOff, autoMerge: true } as Settings);
+
+    expect(result.disposition).toBe("completed");
+    expect(result.context["node:merge-gate:value"]).toBe("auto-off");
+    expect(transitions).toEqual([
+      {
+        id: "work-merge-gate",
+        state: "succeeded",
+        patch: { leaseOwner: null, leaseExpiresAt: null, lastError: null },
+      },
+    ]);
+  });
+
+  it("persists manual merge holds as manual-required work items", async () => {
+    const transitions: Array<{ id: string; state: WorkflowWorkItemState; patch?: Record<string, unknown> }> = [];
+    const workItem = {
+      id: "work-manual-hold",
+      runId: "run-manual-hold",
+      taskId: task.id,
+      nodeId: "merge-manual-hold",
+      kind: "manual-hold",
+      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: () => undefined,
+        getWorkflowDefinition: async () => undefined,
+        transitionWorkflowWorkItem: (id, state, patch) => {
+          transitions.push({ id, state, patch });
+          return { ...workItem, state };
+        },
+      },
+      primitives: recordingPrimitives([]),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+
+    const result = await runtime.runWorkItem(workItem, flagOff);
+
+    expect(result.disposition).toBe("manual-required");
+    expect(result.reason).toBe("manual-required");
+    expect(transitions).toEqual([
+      {
+        id: "work-manual-hold",
+        state: "manual-required",
+        patch: { leaseOwner: null, leaseExpiresAt: null, lastError: "manual-required" },
+      },
+    ]);
+  });
+
+  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; mergeRunId?: string; mergeWorkflowId?: string } = {};
+    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(observed.mergeRunId).toBe("run-merge-attempt");
+    expect(observed.mergeWorkflowId).toBe("builtin:coding");
+    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({

packages/engine/src/workflow-node-handlers.ts

@@ -1,5 +1,5 @@
 import { WorkflowIrError, getStepParser, instanceNodeId } from "@fusion/core";
-import type { NotificationEvent, NotificationPayload, TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";
+import type { NotificationEvent, NotificationPayload, Settings, TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";
 
 import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";
 import { createPrNodeHandlers, createAutoMergeGateHandler, type PrNodeDeps } from "./pr-nodes.js";
@@ -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,30 @@ export function createDefaultNodeHandlers(
     "parse-steps": parseSteps,
     code: createCodeNodeHandler(deps?.runCode),
     notify: createNotifyHandler(deps?.notifyDispatch),
+    "merge-gate": async (_node, ctx) => {
+      const settingsAutoMerge = (ctx.settings as Partial<Settings> | undefined)?.autoMerge;
+      const autoMerge = ctx.task.autoMerge !== false && settingsAutoMerge !== false;
+      return {
+        outcome: "success",
+        value: autoMerge ? "auto-on" : "auto-off",
+      };
+    },
+    "merge-attempt": async (_node, ctx) => {
+      if (!deps?.primitives) return { outcome: "failure", value: "merge-primitives-unwired" };
+      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" }),
+    "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,
   };
 }

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,
@@ -14,13 +14,15 @@ import {
   type WorkflowNodeOutcome,
 } from "./workflow-graph-executor.js";
 import {
+  WORKFLOW_ID_CONTEXT_KEY,
+  WORKFLOW_RUN_ID_CONTEXT_KEY,
   createDefaultNodeHandlers,
   createNoopLegacySeams,
   type WorkflowCustomNodeRunner,
 } from "./workflow-node-handlers.js";
 import type { WorkflowRuntimePrimitives } from "./runtime-primitives.js";
 
-export type WorkflowTaskRuntimeDisposition = "completed" | "failed";
+export type WorkflowTaskRuntimeDisposition = "completed" | "failed" | "manual-required";
 
 export interface WorkflowTaskRuntimeResult {
   disposition: WorkflowTaskRuntimeDisposition;
@@ -31,7 +33,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 +123,115 @@ export class WorkflowTaskRuntime {
     };
   }
 
+  public async runWorkItem(
+    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}`);
+    }
+
+    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_RUN_ID_CONTEXT_KEY]: workItem.runId,
+      [WORKFLOW_ID_CONTEXT_KEY]: target.workflowId,
+      "workflow:work-item-id": workItem.id,
+      "workflow:work-item-kind": workItem.kind,
+      "workflow:work-item-attempt": workItem.attempt,
+    };
+
+    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"
+      : reason === "manual-required"
+        ? "manual-required"
+        : "failed";
+    const terminalState: WorkflowWorkItemState = disposition === "completed"
+      ? "succeeded"
+      : disposition === "manual-required"
+        ? "manual-required"
+        : "failed";
+    this.deps.store.transitionWorkflowWorkItem(workItem.id, terminalState, {
+      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 {