feat(FN-000): process workflow-owned merge work

Fusion-Task-Id: FN-000

docs/plans/workflow-owned-merge-stack/s08-workflow-owned-merge-processing.md

@@ -0,0 +1,46 @@
+---
+title: "S08: workflow-owned merge queue processing"
+type: refactor
+status: draft-stack-handoff
+date: 2026-06-09
+slice: S08
+milestone: "Gate B"
+origin: docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md
+stack_base: feature/workflow-owned-merge-s07-completion-handoff-merge-work
+---
+
+# S08: workflow-owned merge queue processing
+
+## Stack Role
+
+This draft PR reserves the S08 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
+
+Gate B
+
+## Depends On
+
+S3 scheduler claim path, S6 merge capabilities, and S7 completion handoff.
+
+## Goal
+
+Process merge work items through workflow runtime instead of ProjectEngine's in-memory merge queue loop.
+
+## Expected File Scope
+
+packages/engine/src/project-engine.ts; packages/engine/src/scheduler.ts; packages/engine/src/merger.ts; packages/core/src/store.ts; merge lifecycle tests.
+
+## Expected Tests
+
+Serialized merge claim, successful finalize, transient retry, permanent conflict routing, duplicate lease blocking, hard cancel cancellation.
+
+## Exit Gate
+
+Production merge processing no longer depends on a hidden mergeQueue dequeue loop.
+
+## Full Plan
+
+See `docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md`.

packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts

@@ -1,7 +1,12 @@
 // @vitest-environment node
 
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
 import {
+  TaskStore,
   WORKFLOW_EXTENSION_SCHEMA_VERSION,
   __resetWorkflowExtensionRegistryForTests,
   getWorkflowExtensionRegistry,
@@ -13,6 +18,9 @@ import {
 } from "@fusion/core";
 import { TaskExecutor } from "../executor.js";
 import { claimDueWorkflowWorkItem } from "../workflow-work-scheduler.js";
+import { processDueWorkflowWorkItem, workflowMergeWorkKinds } from "../workflow-work-processor.js";
+import { WorkflowTaskRuntime } from "../workflow-task-runtime.js";
+import type { WorkflowRuntimePrimitives } from "../runtime-primitives.js";
 
 describe("workflow work-engine dispatch", () => {
   afterEach(() => {
@@ -152,3 +160,72 @@ describe("workflow work scheduler claims", () => {
     expect(store.acquireWorkflowWorkItemLease).toHaveBeenCalledTimes(2);
   });
 });
+
+describe("workflow work processor", () => {
+  let rootDir: string;
+  let store: TaskStore;
+
+  beforeEach(async () => {
+    rootDir = mkdtempSync(join(tmpdir(), "kb-workflow-work-processor-"));
+    store = new TaskStore(rootDir, join(rootDir, ".fusion-global"));
+    await store.init();
+  });
+
+  afterEach(async () => {
+    store.close();
+    await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+  });
+
+  function primitives(): WorkflowRuntimePrimitives {
+    const success = async () => ({ outcome: "success" as const });
+    return {
+      prepareWorktree: async () => ({ outcome: "success", data: { worktreePath: rootDir } }),
+      readArtifact: async () => undefined,
+      writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }),
+      runPlanningSession: success,
+      runCodingSession: async () => ({ outcome: "success", data: { taskDone: true, modifiedFiles: [] } }),
+      runTaskStep: success,
+      resetTaskStep: async () => ({ ok: true }),
+      runReview: async () => ({ outcome: "success", data: { verdict: "APPROVE" } }),
+      runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }),
+      runWorkflowStep: success,
+      updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }),
+      transitionTask: success,
+      requestMerge: async () => ({ outcome: "success", data: { status: "merged" } }),
+      abortRun: success,
+      audit: vi.fn(),
+    };
+  }
+
+  it("claims due merge work and runs it through workflow runtime", async () => {
+    const task = await store.createTask({ description: "processor task" });
+    await store.moveTask(task.id, "todo");
+    await store.moveTask(task.id, "in-progress");
+    await store.handoffToReview(task.id, {
+      ownerAgentId: "agent-test",
+      evidence: { reason: "fn_task_done", runId: "run-processor", agentId: "agent-test" },
+      now: "2026-06-09T00:00:00.000Z",
+    });
+    const runtime = new WorkflowTaskRuntime({
+      store,
+      primitives: primitives(),
+      runCustomNode: async () => ({ outcome: "success" }),
+    });
+
+    const result = await processDueWorkflowWorkItem(store, runtime, { experimentalFeatures: {} } as any, {
+      now: "2026-06-09T00:00:00.000Z",
+      leaseOwner: "processor-a",
+      leaseDurationMs: 60_000,
+      kinds: workflowMergeWorkKinds(),
+    });
+
+    expect(result).toMatchObject({
+      claimed: true,
+      taskId: task.id,
+      runtime: { disposition: "completed" },
+    });
+    expect(store.listWorkflowWorkItemsForTask(task.id, { kinds: ["merge"] })).toEqual([
+      expect.objectContaining({ state: "succeeded", leaseOwner: null, leaseExpiresAt: null }),
+    ]);
+  });
+});

packages/engine/src/index.ts

@@ -152,6 +152,12 @@ export {
   runWorkflowMergeAttemptNode,
   type WorkflowMergeNodeDeps,
 } from "./workflow-merge-nodes.js";
+export {
+  processDueWorkflowWorkItem,
+  workflowMergeWorkKinds,
+  type WorkflowWorkProcessorOptions,
+  type WorkflowWorkProcessorResult,
+} from "./workflow-work-processor.js";
 export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js";
 export { MissionAutopilot, type MissionAutopilotOptions } from "./mission-autopilot.js";
 export { MissionExecutionLoop, type MissionExecutionLoopOptions, type ValidationResult, loopLog } from "./mission-execution-loop.js";

packages/engine/src/workflow-work-processor.ts

@@ -0,0 +1,44 @@
+import type { Settings, WorkflowWorkItemKind } from "@fusion/core";
+import { claimDueWorkflowWorkItem, type WorkflowWorkSchedulerStore } from "./workflow-work-scheduler.js";
+import { WorkflowTaskRuntime, type WorkflowTaskRuntimeResult } from "./workflow-task-runtime.js";
+
+export interface WorkflowWorkProcessorOptions {
+  leaseOwner: string;
+  leaseDurationMs: number;
+  now?: string;
+  kinds?: WorkflowWorkItemKind[];
+}
+
+export interface WorkflowWorkProcessorResult {
+  claimed: boolean;
+  workItemId?: string;
+  taskId?: string;
+  runtime?: WorkflowTaskRuntimeResult;
+}
+
+export async function processDueWorkflowWorkItem(
+  store: WorkflowWorkSchedulerStore,
+  runtime: WorkflowTaskRuntime,
+  settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
+  opts: WorkflowWorkProcessorOptions,
+): Promise<WorkflowWorkProcessorResult> {
+  const dispatch = claimDueWorkflowWorkItem(store, {
+    now: opts.now,
+    leaseOwner: opts.leaseOwner,
+    leaseDurationMs: opts.leaseDurationMs,
+    kinds: opts.kinds,
+  });
+  if (!dispatch) return { claimed: false };
+
+  const runtimeResult = await runtime.runWorkItem(dispatch.workItem, settings);
+  return {
+    claimed: true,
+    workItemId: dispatch.workItem.id,
+    taskId: dispatch.taskId,
+    runtime: runtimeResult,
+  };
+}
+
+export function workflowMergeWorkKinds(): WorkflowWorkItemKind[] {
+  return ["merge", "manual-hold"];
+}