feat(FN-000): claim workflow work in scheduler substrate
Fusion-Task-Id: FN-000
This commit is contained in:
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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";
|
||||
|
||||
52
packages/engine/src/workflow-work-scheduler.ts
Normal file
52
packages/engine/src/workflow-work-scheduler.ts
Normal file
@@ -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;
|
||||
}
|
||||
Reference in New Issue
Block a user