refactor(workflow): add workflow task runtime facade
This commit is contained in:
gsxdsm
300
packages/engine/src/__tests__/workflow-task-runtime.test.ts
Normal file
300
packages/engine/src/__tests__/workflow-task-runtime.test.ts
Normal file
@@ -0,0 +1,300 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { Settings, TaskDetail, WorkflowIr } from "@fusion/core";
|
||||
|
||||
import { WorkflowTaskRuntime, type WorkflowTaskRuntimeDeps } from "../workflow-task-runtime.js";
|
||||
import type { WorkflowNodeResult } from "../workflow-graph-executor.js";
|
||||
import type { WorkflowLegacySeams } from "../workflow-node-handlers.js";
|
||||
|
||||
const task = { id: "FN-9002" } as TaskDetail;
|
||||
const flagOff = { experimentalFeatures: {} } as unknown as Pick<Settings, "experimentalFeatures">;
|
||||
|
||||
function selectedIr(): WorkflowIr {
|
||||
return {
|
||||
version: "v1",
|
||||
name: "selected",
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "prepare", kind: "prompt", config: { prompt: "prepare" } },
|
||||
{ id: "execute", kind: "prompt", config: { seam: "execute" } },
|
||||
{ id: "zend", kind: "end" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "prepare", condition: "success" },
|
||||
{ from: "prepare", to: "execute", condition: "success" },
|
||||
{ from: "execute", to: "zend", condition: "success" },
|
||||
{ from: "execute", to: "zend", condition: "failure" },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function recordingSeams(calls: string[], overrides: Partial<Record<string, WorkflowNodeResult>> = {}): WorkflowLegacySeams {
|
||||
const seam = (name: keyof WorkflowLegacySeams) => async (): Promise<WorkflowNodeResult> => {
|
||||
calls.push(name);
|
||||
return overrides[name] ?? { outcome: "success" };
|
||||
};
|
||||
return {
|
||||
planning: seam("planning"),
|
||||
execute: seam("execute"),
|
||||
review: seam("review"),
|
||||
merge: seam("merge"),
|
||||
schedule: seam("schedule"),
|
||||
};
|
||||
}
|
||||
|
||||
describe("WorkflowTaskRuntime", () => {
|
||||
it("requires execution wiring at the type boundary", () => {
|
||||
// @ts-expect-error WorkflowTaskRuntime is an execution entry point, so seams are required.
|
||||
const missingSeams: WorkflowTaskRuntimeDeps = {
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
};
|
||||
expect(missingSeams).toBeDefined();
|
||||
});
|
||||
|
||||
it("runs a selected workflow through the graph engine", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["custom:prepare", "execute"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "prepare", "execute"]);
|
||||
});
|
||||
|
||||
it("resolves an unselected task to the built-in coding workflow instead of falling back", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "execute", "review", "merge"]);
|
||||
});
|
||||
|
||||
it("turns selected workflow lookup failures into the built-in workflow via the shared resolver", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-MISSING", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
});
|
||||
|
||||
it("turns corrupt selected workflow definitions into the built-in workflow via the shared resolver", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-CORRUPT", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: "not a workflow ir" }),
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
});
|
||||
|
||||
it("forces only the graph executor flag while preserving other settings", async () => {
|
||||
let observedSettings: Pick<Settings, "experimentalFeatures"> | undefined;
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
handlers: {
|
||||
prompt: async (_node, context) => {
|
||||
observedSettings = context.settings;
|
||||
return { outcome: "success" };
|
||||
},
|
||||
},
|
||||
});
|
||||
const settings = {
|
||||
experimentalFeatures: { workflowColumns: true },
|
||||
testMode: true,
|
||||
} as unknown as Settings;
|
||||
|
||||
const result = await runtime.run(task, settings);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(observedSettings?.experimentalFeatures?.workflowGraphExecutor).toBe(true);
|
||||
expect(observedSettings?.experimentalFeatures?.workflowColumns).toBe(true);
|
||||
expect((observedSettings as Settings | undefined)?.testMode).toBe(true);
|
||||
});
|
||||
|
||||
it("uses a workflow-specific default run id", async () => {
|
||||
const observedRunIds: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
observedRunIds.push(runId);
|
||||
return [];
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await runtime.run(task, flagOff);
|
||||
|
||||
expect(observedRunIds).toContain("FN-9002:WF-001");
|
||||
});
|
||||
|
||||
it("uses the built-in workflow id in the default run id for unselected tasks", async () => {
|
||||
const observedRunIds: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
observedRunIds.push(runId);
|
||||
return [];
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await runtime.run(task, flagOff);
|
||||
|
||||
expect(observedRunIds).toContain("FN-9002:builtin:coding");
|
||||
});
|
||||
|
||||
it("surfaces graph failures as workflow-engine failures, not fallback", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams(calls, { execute: { outcome: "failure", value: "implementation-incomplete" } }),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(calls).toEqual(["custom:prepare", "execute"]);
|
||||
});
|
||||
|
||||
it("converts interpreter throws into workflow-engine failures", async () => {
|
||||
const badIr: WorkflowIr = {
|
||||
version: "v1",
|
||||
name: "bad",
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "zend", kind: "end" },
|
||||
],
|
||||
edges: [{ from: "start", to: "ghost" }],
|
||||
};
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: badIr }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toMatch(/workflow-execution-error/);
|
||||
});
|
||||
|
||||
it("preserves invoked node ids when the graph throws after side effects", async () => {
|
||||
const cyclicIr: WorkflowIr = {
|
||||
version: "v1",
|
||||
name: "cyclic",
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "prepare", kind: "prompt", config: { prompt: "prepare" } },
|
||||
{ id: "loop", kind: "prompt", config: { prompt: "loop" } },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "prepare", condition: "success" },
|
||||
{ from: "prepare", to: "loop", condition: "success" },
|
||||
{ from: "loop", to: "prepare", condition: "success" },
|
||||
],
|
||||
};
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: cyclicIr }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toMatch(/workflow-execution-error/);
|
||||
expect(result.visitedNodeIds).toEqual(["prepare", "loop"]);
|
||||
});
|
||||
|
||||
it("diagnostic event failures do not affect execution", async () => {
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
onEvent: () => {
|
||||
throw new Error("diagnostics failed");
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
});
|
||||
});
|
||||
@@ -107,6 +107,12 @@ export {
|
||||
type WorkflowGraphTaskRunResult,
|
||||
type WorkflowGraphTaskRunnerDeps,
|
||||
} from "./workflow-graph-task-runner.js";
|
||||
export {
|
||||
WorkflowTaskRuntime,
|
||||
type WorkflowTaskRuntimeDeps,
|
||||
type WorkflowTaskRuntimeDisposition,
|
||||
type WorkflowTaskRuntimeResult,
|
||||
} from "./workflow-task-runtime.js";
|
||||
export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
|
||||
export { Scheduler, type SchedulerOptions } from "./scheduler.js";
|
||||
export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js";
|
||||
|
||||
172
packages/engine/src/workflow-task-runtime.ts
Normal file
172
packages/engine/src/workflow-task-runtime.ts
Normal file
@@ -0,0 +1,172 @@
|
||||
import type { Settings, TaskDetail, WorkflowIr } from "@fusion/core";
|
||||
import { resolveWorkflowIrForTask, type WorkflowIrResolverStore } from "@fusion/core";
|
||||
|
||||
import {
|
||||
WorkflowGraphExecutor,
|
||||
type WorkflowGraphExecutorDeps,
|
||||
type WorkflowNodeOutcome,
|
||||
} from "./workflow-graph-executor.js";
|
||||
import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
|
||||
|
||||
export type WorkflowTaskRuntimeDisposition = "completed" | "failed";
|
||||
|
||||
export interface WorkflowTaskRuntimeResult {
|
||||
disposition: WorkflowTaskRuntimeDisposition;
|
||||
outcome: WorkflowNodeOutcome;
|
||||
visitedNodeIds: string[];
|
||||
context: Record<string, unknown>;
|
||||
reason?: string;
|
||||
}
|
||||
|
||||
export interface WorkflowTaskRuntimeDeps extends Omit<WorkflowGraphExecutorDeps, "seams" | "runCustomNode"> {
|
||||
store: WorkflowIrResolverStore;
|
||||
seams: WorkflowLegacySeams;
|
||||
runCustomNode: WorkflowCustomNodeRunner;
|
||||
onEvent?: (event: { type: "start" | "terminal"; taskId: string; detail: string }) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* WorkflowTaskRuntime is the workflow-engine execution facade.
|
||||
*
|
||||
* It always resolves a task to a workflow IR: explicit selections resolve to
|
||||
* their selected workflow, and tasks without a selection resolve to the built-in
|
||||
* coding workflow through `resolveWorkflowIrForTask`. This is intentionally
|
||||
* different from `WorkflowGraphTaskRunner`, whose current contract still models
|
||||
* "no selection" as legacy fallback.
|
||||
*/
|
||||
export class WorkflowTaskRuntime {
|
||||
public constructor(private readonly deps: WorkflowTaskRuntimeDeps) {}
|
||||
|
||||
private emit(type: "start" | "terminal", taskId: string, detail: string): void {
|
||||
try {
|
||||
this.deps.onEvent?.({ type, taskId, detail });
|
||||
} catch {
|
||||
// Diagnostics must never affect execution.
|
||||
}
|
||||
}
|
||||
|
||||
public async run(
|
||||
task: TaskDetail,
|
||||
settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
|
||||
): Promise<WorkflowTaskRuntimeResult> {
|
||||
this.emit("start", task.id, "resolve-workflow");
|
||||
|
||||
const workflowId = this.resolveWorkflowId(task.id);
|
||||
let ir: WorkflowIr;
|
||||
try {
|
||||
ir = await resolveWorkflowIrForTask(this.deps.store, task.id);
|
||||
} catch (err) {
|
||||
const reason = `workflow-resolution-error: ${err instanceof Error ? err.message : String(err)}`;
|
||||
this.emit("terminal", task.id, `failed:${reason}`);
|
||||
return {
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
visitedNodeIds: [],
|
||||
context: {},
|
||||
reason,
|
||||
};
|
||||
}
|
||||
|
||||
const invoked: string[] = [];
|
||||
const wrappedSeams = this.wrapSeams(invoked);
|
||||
const wrappedRunCustomNode: WorkflowCustomNodeRunner = (node, nodeTask, context) => {
|
||||
invoked.push(node.id);
|
||||
return this.deps.runCustomNode(node, nodeTask, context);
|
||||
};
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
...this.deps,
|
||||
seams: wrappedSeams,
|
||||
runCustomNode: wrappedRunCustomNode,
|
||||
// WorkflowTaskRuntime is the execution engine, so internally the graph
|
||||
// executor is authoritative even before the old feature flag plumbing is
|
||||
// deleted from legacy entry points.
|
||||
runId: this.deps.runId ?? `${task.id}:${workflowId}`,
|
||||
});
|
||||
|
||||
const runtimeSettings = forceWorkflowGraphExecutor(settings);
|
||||
let result: Awaited<ReturnType<WorkflowGraphExecutor["run"]>>;
|
||||
try {
|
||||
result = await executor.run(task, runtimeSettings, ir);
|
||||
} catch (err) {
|
||||
const reason = `workflow-execution-error: ${err instanceof Error ? err.message : String(err)}`;
|
||||
this.emit("terminal", task.id, `failed:${reason}`);
|
||||
return {
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
visitedNodeIds: invoked,
|
||||
context: {},
|
||||
reason,
|
||||
};
|
||||
}
|
||||
const disposition: WorkflowTaskRuntimeDisposition = result.outcome === "success" ? "completed" : "failed";
|
||||
this.emit("terminal", task.id, disposition);
|
||||
return {
|
||||
disposition,
|
||||
outcome: result.outcome,
|
||||
visitedNodeIds: result.visitedNodeIds,
|
||||
context: result.context,
|
||||
};
|
||||
}
|
||||
|
||||
private resolveWorkflowId(taskId: string): string {
|
||||
try {
|
||||
return this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId ?? "builtin:coding";
|
||||
} catch {
|
||||
return "builtin:coding";
|
||||
}
|
||||
}
|
||||
|
||||
private wrapSeams(invoked: string[]): WorkflowLegacySeams {
|
||||
const seams = this.deps.seams;
|
||||
return {
|
||||
planning: (task, context) => {
|
||||
invoked.push("planning");
|
||||
return seams.planning(task, context);
|
||||
},
|
||||
execute: (task, context) => {
|
||||
invoked.push("execute");
|
||||
return seams.execute(task, context);
|
||||
},
|
||||
review: (task, context) => {
|
||||
invoked.push("review");
|
||||
return seams.review(task, context);
|
||||
},
|
||||
merge: (task, context) => {
|
||||
invoked.push("merge");
|
||||
return seams.merge(task, context);
|
||||
},
|
||||
schedule: (task, context) => {
|
||||
invoked.push("schedule");
|
||||
return seams.schedule(task, context);
|
||||
},
|
||||
...(seams.stepExecute
|
||||
? {
|
||||
stepExecute: (task, context) => {
|
||||
invoked.push("step-execute");
|
||||
return seams.stepExecute!(task, context);
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
...(seams.stepReview
|
||||
? {
|
||||
stepReview: (task, context, config) => {
|
||||
invoked.push("step-review");
|
||||
return seams.stepReview!(task, context, config);
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function forceWorkflowGraphExecutor(
|
||||
settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
|
||||
): Pick<Settings, "experimentalFeatures"> & Partial<Settings> {
|
||||
return {
|
||||
...(settings ?? {}),
|
||||
experimentalFeatures: {
|
||||
...(settings?.experimentalFeatures ?? {}),
|
||||
workflowGraphExecutor: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user