From 0ec7b496aedac23b590e1f63380451a129e5a64f Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 17:15:59 -0700 Subject: [PATCH] feat(FN-000): claim workflow work in scheduler substrate Fusion-Task-Id: FN-000 --- .../s03-generic-scheduler-claim.md | 46 ++++++++++++ .../workflow-work-engine-dispatch.test.ts | 70 +++++++++++++++++++ packages/engine/src/index.ts | 6 ++ .../engine/src/workflow-work-scheduler.ts | 52 ++++++++++++++ 4 files changed, 174 insertions(+) create mode 100644 docs/plans/workflow-owned-merge-stack/s03-generic-scheduler-claim.md create mode 100644 packages/engine/src/workflow-work-scheduler.ts diff --git a/docs/plans/workflow-owned-merge-stack/s03-generic-scheduler-claim.md b/docs/plans/workflow-owned-merge-stack/s03-generic-scheduler-claim.md new file mode 100644 index 0000000000..dfa89bbd7f --- /dev/null +++ b/docs/plans/workflow-owned-merge-stack/s03-generic-scheduler-claim.md @@ -0,0 +1,46 @@ +--- +title: "S03: generic scheduler claim path" +type: refactor +status: draft-stack-handoff +date: 2026-06-09 +slice: S03 +milestone: "Foundation" +origin: docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md +stack_base: feature/workflow-owned-merge-s02-merge-request-projection +--- + +# S03: generic scheduler claim path + +## Stack Role + +This draft PR reserves the S03 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 + +Foundation + +## Depends On + +S1 workflow work-item schema and store API. + +## Goal + +Teach Scheduler to claim due workflow work items while preserving existing task dispatch behavior. + +## Expected File Scope + +packages/engine/src/scheduler.ts; packages/engine/src/workflow-task-runtime.ts; packages/engine/src/project-engine.ts; scheduler/workflow dispatch tests. + +## Expected Tests + +Due-work claiming, retryAfter delay, capacity holds, user pause exclusion, stale lease reclaim, and remote dispatch. + +## Exit Gate + +A workflow work item can be dispatched end to end in tests without constructing a merge queue branch. + +## 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-work-engine-dispatch.test.ts b/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts index 789023f47a..42cc9aa1c3 100644 --- a/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts +++ b/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 & Pick): 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); + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 76826ac8ae..70bd3a348d 100644 --- a/packages/engine/src/index.ts +++ b/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"; diff --git a/packages/engine/src/workflow-work-scheduler.ts b/packages/engine/src/workflow-work-scheduler.ts new file mode 100644 index 0000000000..8ae7206a86 --- /dev/null +++ b/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; +}