From ba27e499b4271e094d13bfb3106244479075cf80 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 10:26:33 -0700 Subject: [PATCH] feat(engine): executable custom nodes in the workflow graph interpreter (CU-U1) Non-seam prompt/script nodes now dispatch to an injected WorkflowCustomNodeRunner instead of throwing; gate nodes support an executable (prompt/script-backed) form alongside the original context-gate contract. WorkflowGraphExecutor accepts the runner via deps. --- .../__tests__/workflow-node-handlers.test.ts | 74 +++++++++++++++++ .../engine/src/workflow-graph-executor.ts | 11 ++- packages/engine/src/workflow-node-handlers.ts | 83 +++++++++++++++---- 3 files changed, 150 insertions(+), 18 deletions(-) diff --git a/packages/engine/src/__tests__/workflow-node-handlers.test.ts b/packages/engine/src/__tests__/workflow-node-handlers.test.ts index 2412e5495a..369f8d32a9 100644 --- a/packages/engine/src/__tests__/workflow-node-handlers.test.ts +++ b/packages/engine/src/__tests__/workflow-node-handlers.test.ts @@ -47,4 +47,78 @@ describe("workflow node handlers", () => { expect(result).toEqual({ outcome: "failure", value: "gate-mismatch" }); }); + + const noopSeams = () => ({ + execute: vi.fn(async () => ({ outcome: "success" as const })), + review: vi.fn(async () => ({ outcome: "success" as const })), + merge: vi.fn(async () => ({ outcome: "success" as const })), + schedule: vi.fn(async () => ({ outcome: "success" as const })), + }); + + it("dispatches a custom (non-seam) prompt node to the custom-node runner", async () => { + const seams = noopSeams(); + const runCustomNode = vi.fn(async () => ({ outcome: "success" as const, value: "APPROVE" })); + const handlers = createDefaultNodeHandlers(seams, runCustomNode); + + const customNode: WorkflowIrNode = { id: "spec-check", kind: "prompt", config: { prompt: "Check the spec" } }; + const result = await handlers.prompt(customNode, { task, settings: undefined, context: {} }); + + expect(runCustomNode).toHaveBeenCalledWith(customNode, task, {}); + expect(result.value).toBe("APPROVE"); + expect(seams.execute).not.toHaveBeenCalled(); + expect(seams.review).not.toHaveBeenCalled(); + }); + + it("dispatches a custom script node to the custom-node runner", async () => { + const runCustomNode = vi.fn(async () => ({ outcome: "failure" as const, value: "exit-1" })); + const handlers = createDefaultNodeHandlers(noopSeams(), runCustomNode); + + const result = await handlers.script( + { id: "lint", kind: "script", config: { scriptName: "lint" } }, + { task, settings: undefined, context: {} }, + ); + + expect(runCustomNode).toHaveBeenCalledOnce(); + expect(result.outcome).toBe("failure"); + }); + + it("throws for a custom node when no runner is registered", async () => { + const handlers = createDefaultNodeHandlers(noopSeams()); + await expect( + handlers.prompt({ id: "p", kind: "prompt", config: { prompt: "x" } }, { task, settings: undefined, context: {} }), + ).rejects.toThrow(/custom-node runner/i); + }); + + it("throws for an unknown seam string", async () => { + const handlers = createDefaultNodeHandlers(noopSeams(), vi.fn()); + await expect( + handlers.prompt(node("prompt", "deploy"), { task, settings: undefined, context: {} }), + ).rejects.toThrow(/Unsupported workflow seam/i); + }); + + it("runs an executable gate (prompt-backed) through the runner and gates on its outcome", async () => { + const runCustomNode = vi.fn(async () => ({ outcome: "failure" as const, value: "REVISE" })); + const handlers = createDefaultNodeHandlers(noopSeams(), runCustomNode); + + const result = await handlers.gate( + { id: "quality-gate", kind: "gate", config: { prompt: "Block on regressions", gateMode: "gate" } }, + { task, settings: undefined, context: {} }, + ); + + expect(runCustomNode).toHaveBeenCalledOnce(); + expect(result.outcome).toBe("failure"); + }); + + it("context gate still takes precedence over executable config", async () => { + const runCustomNode = vi.fn(async () => ({ outcome: "failure" as const })); + const handlers = createDefaultNodeHandlers(noopSeams(), runCustomNode); + + const result = await handlers.gate( + { id: "g", kind: "gate", config: { expect: "done", contextKey: "phase", prompt: "x" } }, + { task, settings: undefined, context: { phase: "done" } }, + ); + + expect(result.outcome).toBe("success"); + expect(runCustomNode).not.toHaveBeenCalled(); + }); }); diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 1b8699f1f7..0b772253c9 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -1,7 +1,12 @@ import type { Settings, TaskDetail, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled } from "@fusion/core"; -import { createDefaultNodeHandlers, createNoopLegacySeams, type WorkflowLegacySeams } from "./workflow-node-handlers.js"; +import { + createDefaultNodeHandlers, + createNoopLegacySeams, + type WorkflowCustomNodeRunner, + type WorkflowLegacySeams, +} from "./workflow-node-handlers.js"; export type WorkflowNodeOutcome = "success" | "failure"; @@ -22,6 +27,8 @@ export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeEx export interface WorkflowGraphExecutorDeps { handlers?: Partial>; seams?: WorkflowLegacySeams; + /** Executes custom (non-seam) prompt/script/gate nodes. */ + runCustomNode?: WorkflowCustomNodeRunner; maxRetriesPerNode?: number; } @@ -47,7 +54,7 @@ export class WorkflowGraphExecutor { public constructor(private readonly deps: WorkflowGraphExecutorDeps) { this.maxRetriesPerNode = Math.max(1, Math.floor(deps.maxRetriesPerNode ?? 2)); this.handlers = { - ...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams()), + ...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams(), deps.runCustomNode), ...(deps.handlers ?? {}), }; } diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 962f0b465d..db40a5de7d 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -1,5 +1,5 @@ import { WorkflowIrError } from "@fusion/core"; -import type { TaskDetail } from "@fusion/core"; +import type { TaskDetail, WorkflowIrNode } from "@fusion/core"; import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js"; @@ -12,39 +12,90 @@ export interface WorkflowLegacySeams { schedule: (task: TaskDetail, context: Record) => Promise; } -function resolveSeam(node: { config?: Record }): WorkflowSeamName { +/** + * Runs a custom (non-seam) prompt/script/gate node for a task — typically by + * delegating to the WorkflowStep prompt-session/script machinery. Injected so + * the graph layer stays engine-agnostic and unit-testable with fakes. + */ +export type WorkflowCustomNodeRunner = ( + node: WorkflowIrNode, + task: TaskDetail, + context: Record, +) => Promise; + +/** Resolve a node's seam name, or undefined for custom (non-seam) nodes. */ +export function resolveSeamName(node: { config?: Record }): WorkflowSeamName | undefined { const seam = node.config?.seam; + if (seam === undefined) return undefined; if (seam === "execute" || seam === "review" || seam === "merge" || seam === "schedule") { return seam; } throw new WorkflowIrError(`Unsupported workflow seam: ${String(seam)}`); } -export function createPromptLikeHandler(seams: WorkflowLegacySeams): WorkflowNodeHandler { +/** + * Prompt/script handler: seam-configured nodes delegate to the legacy seam; + * custom nodes run through the injected custom-node runner. + */ +export function createPromptLikeHandler( + seams: WorkflowLegacySeams, + runCustomNode?: WorkflowCustomNodeRunner, +): WorkflowNodeHandler { return async (node, context) => { - const seam = resolveSeam(node); - return seams[seam](context.task, context.context); + const seam = resolveSeamName(node); + if (seam) { + return seams[seam](context.task, context.context); + } + if (!runCustomNode) { + throw new WorkflowIrError(`No custom-node runner registered for node: ${node.id}`); + } + return runCustomNode(node, context.task, context.context); }; } -export const gateNodeHandler: WorkflowNodeHandler = async (node, context) => { - const expected = node.config?.expect; - const actual = context.context[String(node.config?.contextKey ?? "outcome")]; - if (typeof expected === "string" && actual !== expected) { - return { outcome: "failure", value: "gate-mismatch" }; - } - return { outcome: "success" }; -}; +/** + * Gate handler. Two forms: + * - Context gate (original scaffold contract): `config.expect` compared against + * a context key — pure, no execution. + * - Executable gate: a gate node carrying a prompt/script config runs through + * the custom-node runner; its outcome decides whether the gate passes. + */ +export function createGateHandler(runCustomNode?: WorkflowCustomNodeRunner): WorkflowNodeHandler { + return async (node, context) => { + const expected = node.config?.expect; + if (typeof expected === "string") { + const actual = context.context[String(node.config?.contextKey ?? "outcome")]; + if (actual !== expected) { + return { outcome: "failure", value: "gate-mismatch" }; + } + return { outcome: "success" }; + } -export function createDefaultNodeHandlers(seams: WorkflowLegacySeams): Record<"prompt" | "script" | "gate", WorkflowNodeHandler> { - const promptLike = createPromptLikeHandler(seams); + const hasExecutableConfig = + typeof node.config?.prompt === "string" || typeof node.config?.scriptName === "string"; + if (hasExecutableConfig && runCustomNode) { + return runCustomNode(node, context.task, context.context); + } + + return { outcome: "success" }; + }; +} + +export function createDefaultNodeHandlers( + seams: WorkflowLegacySeams, + runCustomNode?: WorkflowCustomNodeRunner, +): Record<"prompt" | "script" | "gate", WorkflowNodeHandler> { + const promptLike = createPromptLikeHandler(seams, runCustomNode); return { prompt: promptLike, script: promptLike, - gate: gateNodeHandler, + gate: createGateHandler(runCustomNode), }; } +/** Back-compat export: the original context-only gate handler. */ +export const gateNodeHandler: WorkflowNodeHandler = createGateHandler(); + export function createNoopLegacySeams(): WorkflowLegacySeams { const success = async (): Promise => ({ outcome: "success" }); return {