Merge pull request #1576 from Runfusion/feature/workflow-owned-merge-s03-generic-scheduler-claim

refactor(workflow): S03 generic scheduler claim path

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

@@ -8,9 +8,11 @@ import {
   workflowExtensionRegistryId,
   type Task,
   type TaskDetail,
+  type WorkflowWorkItem,
   type WorkflowIr,
 } from "@fusion/core";
 import { TaskExecutor } from "../executor.js";
+import { claimDueWorkflowWorkItem } from "../workflow-work-scheduler.js";
 
 describe("workflow work-engine dispatch", () => {
   afterEach(() => {
@@ -82,3 +84,71 @@ describe("workflow work-engine dispatch", () => {
     expect(store.updateTask).not.toHaveBeenCalled();
   });
 });
+
+describe("workflow work scheduler claims", () => {
+  function workItem(input: Partial<WorkflowWorkItem> & Pick<WorkflowWorkItem, "id" | "taskId" | "nodeId">): WorkflowWorkItem {
+    return {
+      runId: "run-1",
+      kind: "task",
+      state: "runnable",
+      attempt: 0,
+      retryAfter: null,
+      leaseOwner: null,
+      leaseExpiresAt: null,
+      lastError: null,
+      blockedReason: null,
+      createdAt: "2026-06-09T00:00:00.000Z",
+      updatedAt: "2026-06-09T00:00:00.000Z",
+      ...input,
+    };
+  }
+
+  it("claims the first due workflow work item without reading task columns", () => {
+    const item = workItem({ id: "work-1", taskId: "FN-1", nodeId: "node-a" });
+    const store = {
+      listDueWorkflowWorkItems: vi.fn(() => [item]),
+      acquireWorkflowWorkItemLease: vi.fn(() => ({ ...item, state: "running", leaseOwner: "scheduler-a" })),
+    };
+
+    const dispatch = claimDueWorkflowWorkItem(store, {
+      now: "2026-06-09T00:00:00.000Z",
+      leaseOwner: "scheduler-a",
+      leaseDurationMs: 60_000,
+      kinds: ["task"],
+    });
+
+    expect(store.listDueWorkflowWorkItems).toHaveBeenCalledWith({
+      now: "2026-06-09T00:00:00.000Z",
+      limit: 25,
+      kinds: ["task"],
+    });
+    expect(store.acquireWorkflowWorkItemLease).toHaveBeenCalledWith("work-1", "scheduler-a", {
+      now: "2026-06-09T00:00:00.000Z",
+      leaseDurationMs: 60_000,
+    });
+    expect(dispatch).toMatchObject({
+      runId: "run-1",
+      taskId: "FN-1",
+      nodeId: "node-a",
+      workItem: { state: "running", leaseOwner: "scheduler-a" },
+    });
+  });
+
+  it("skips contenders whose lease was already acquired", () => {
+    const first = workItem({ id: "work-1", taskId: "FN-1", nodeId: "node-a" });
+    const second = workItem({ id: "work-2", taskId: "FN-2", nodeId: "node-b" });
+    const store = {
+      listDueWorkflowWorkItems: vi.fn(() => [first, second]),
+      acquireWorkflowWorkItemLease: vi.fn((id: string) => (id === "work-2" ? { ...second, state: "running" } : null)),
+    };
+
+    const dispatch = claimDueWorkflowWorkItem(store, {
+      now: "2026-06-09T00:00:00.000Z",
+      leaseOwner: "scheduler-a",
+      leaseDurationMs: 60_000,
+    });
+
+    expect(dispatch?.workItem.id).toBe("work-2");
+    expect(store.acquireWorkflowWorkItemLease).toHaveBeenCalledTimes(2);
+  });
+});

packages/engine/src/index.ts

@@ -141,6 +141,12 @@ export {
 } from "./workflow-task-runtime.js";
 export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
 export { Scheduler, type SchedulerOptions } from "./scheduler.js";
+export {
+  claimDueWorkflowWorkItem,
+  type ClaimWorkflowWorkOptions,
+  type WorkflowWorkDispatch,
+  type WorkflowWorkSchedulerStore,
+} from "./workflow-work-scheduler.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-scheduler.ts

@@ -0,0 +1,52 @@
+import type { WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind } from "@fusion/core";
+
+export interface WorkflowWorkSchedulerStore {
+  listDueWorkflowWorkItems(filter?: WorkflowWorkItemDueFilter): WorkflowWorkItem[];
+  acquireWorkflowWorkItemLease(
+    id: string,
+    leaseOwner: string,
+    opts: { leaseDurationMs: number; now?: string },
+  ): WorkflowWorkItem | null;
+}
+
+export interface WorkflowWorkDispatch {
+  workItem: WorkflowWorkItem;
+  runId: string;
+  taskId: string;
+  nodeId: string;
+}
+
+export interface ClaimWorkflowWorkOptions {
+  now?: string;
+  limit?: number;
+  leaseOwner: string;
+  leaseDurationMs: number;
+  kinds?: WorkflowWorkItemKind[];
+}
+
+export function claimDueWorkflowWorkItem(
+  store: WorkflowWorkSchedulerStore,
+  opts: ClaimWorkflowWorkOptions,
+): WorkflowWorkDispatch | null {
+  const due = store.listDueWorkflowWorkItems({
+    now: opts.now,
+    limit: opts.limit ?? 25,
+    kinds: opts.kinds,
+  });
+
+  for (const candidate of due) {
+    const workItem = store.acquireWorkflowWorkItemLease(candidate.id, opts.leaseOwner, {
+      now: opts.now,
+      leaseDurationMs: opts.leaseDurationMs,
+    });
+    if (!workItem) continue;
+    return {
+      workItem,
+      runId: workItem.runId,
+      taskId: workItem.taskId,
+      nodeId: workItem.nodeId,
+    };
+  }
+
+  return null;
+}