root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(FN-6035): route execution through workflow primitives

Browse Source 
Fusion-Task-Id: FN-6035
This commit is contained in:
gsxdsm
2026-06-08 19:13:14 -07:00
parent 8bc3d7b0a5
commit 8f42098cc3
17 changed files with 1383 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
packages/engine/src/workflow-node-handlers.ts
View File

@@ -3,6 +3,11 @@ import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";
import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";
import { createPrNodeHandlers, createAutoMergeGateHandler, type PrNodeDeps } from "./pr-nodes.js";
import {
primitiveNodeContext,
type WorkflowPrimitiveContext,
type WorkflowRuntimePrimitives,
} from "./runtime-primitives.js";
export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute";
@@ -104,6 +109,12 @@ export const SPLIT_ACTIVE_CONTEXT_KEY = "split:active";
*/
export const INTEGRATION_CONFLICT_CONTEXT_KEY = "integration:conflict";
/** Reserved graph context key for the current workflow run id. */
export const WORKFLOW_RUN_ID_CONTEXT_KEY = "workflow:run-id";
/** Reserved graph context key for the current workflow id. */
export const WORKFLOW_ID_CONTEXT_KEY = "workflow:id";
/** Shape of the value stored under {@link FOREACH_ACTIVE_CONTEXT_KEY}. */
export interface ForeachActiveContext {
foreachNodeId: string;
@@ -149,6 +160,34 @@ export type WorkflowCustomNodeRunner = (
context: Record<string, unknown>,
) => Promise<WorkflowNodeResult>;
function primitiveContextForNode(
node: WorkflowIrNode,
task: TaskDetail,
context: Record<string, unknown>,
attempt?: number,
): WorkflowPrimitiveContext {
return primitiveNodeContext(
{
runId: typeof context[WORKFLOW_RUN_ID_CONTEXT_KEY] === "string"
? context[WORKFLOW_RUN_ID_CONTEXT_KEY]
: `${task.id}:workflow`,
taskId: task.id,
workflowId: typeof context[WORKFLOW_ID_CONTEXT_KEY] === "string"
? context[WORKFLOW_ID_CONTEXT_KEY]
: "unknown",
},
node,
{
attempt,
context,
effectivePrincipalId:
typeof context["workflow:effective-principal-id"] === "string"
? context["workflow:effective-principal-id"]
: undefined,
},
);
}
/** Resolve a node's seam name, or undefined for custom (non-seam) nodes. */
export function resolveSeamName(node: { config?: Record<string, unknown> }): WorkflowSeamName | undefined {
const seam = node.config?.seam;
@@ -220,6 +259,83 @@ export function createPromptLikeHandler(
};
}
export function createPrimitivePromptLikeHandler(
primitives: WorkflowRuntimePrimitives,
runCustomNode?: WorkflowCustomNodeRunner,
): WorkflowNodeHandler {
return async (node, context) => {
const seam = resolveSeamName(node);
if (seam === "step-execute") {
const active = context.context[FOREACH_ACTIVE_CONTEXT_KEY] as
| ForeachActiveContext
| undefined;
if (!active || typeof active.stepIndex !== "number") {
throw new WorkflowIrError(
`step-execute node '${node.id}' reached without an active foreach instance context`,
);
}
context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = instanceNodeId(
active.foreachNodeId,
active.stepIndex,
node.id,
);
const result = await primitives.runTaskStep(
primitiveContextForNode(node, context.task, context.context),
context.task,
active.stepIndex,
);
active.baselineSha = result.baselineSha;
active.checkpointId = result.checkpointId;
return {
outcome: result.outcome,
value: result.outcome === "success" ? "step-done" : "step-failed",
contextPatch: {
[FOREACH_ACTIVE_CONTEXT_KEY]: active,
},
};
}
if (seam) {
context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = node.id;
const primitiveCtx = primitiveContextForNode(node, context.task, context.context);
if (seam === "planning") {
const result = await primitives.runPlanningSession(primitiveCtx, context.task);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
if (seam === "execute") {
const prepared = await primitives.prepareWorktree(primitiveCtx, context.task);
if (prepared.outcome !== "success" || !prepared.data) {
return {
outcome: prepared.outcome,
value: prepared.value ?? "prepare-worktree-failed",
contextPatch: prepared.contextPatch,
};
}
const result = await primitives.runCodingSession(primitiveCtx, context.task, prepared.data);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
if (seam === "review") {
const result = await primitives.runReview(primitiveCtx, context.task, { type: "code" });
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
if (seam === "merge") {
const result = await primitives.requestMerge(primitiveCtx, context.task);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
if (seam === "schedule") {
const result = await primitives.transitionTask(primitiveCtx, context.task, {
reason: "workflow-schedule",
preserveProgress: true,
});
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
}
if (!runCustomNode) {
throw new WorkflowIrError(`No custom-node runner registered for node: ${node.id}`);
}
return runCustomNode(node, context.task, context.context);
};
}
/**
* Gate handler. Two forms:
* - Context gate (original scaffold contract): `config.expect` compared against
@@ -337,6 +453,55 @@ export function createStepReviewHandler(seams: WorkflowLegacySeams): WorkflowNod
};
}
export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrimitives): WorkflowNodeHandler {
return async (node, ctx) => {
const active = ctx.context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined;
if (!active || typeof active.stepIndex !== "number") {
throw new WorkflowIrError(
`step-review node '${node.id}' reached without an active foreach instance context`,
);
}
const advisory = ctx.context[SPLIT_ACTIVE_CONTEXT_KEY] === true;
const config = resolveStepReviewConfig(node, advisory);
let result: StepReviewSeamResult = {
verdict: "UNAVAILABLE",
};
for (let attempt = 0; attempt <= STEP_REVIEW_UNAVAILABLE_RETRY_CAP; attempt++) {
const primitiveResult = await primitives.runReview(
primitiveContextForNode(node, ctx.task, ctx.context, attempt + 1),
ctx.task,
{
type: config.type,
stepIndex: active.stepIndex,
baselineSha: config.type === "code" ? active.baselineSha : undefined,
},
);
result = primitiveResult.data ?? { verdict: "UNAVAILABLE" as const };
if (result.verdict !== "UNAVAILABLE") break;
}
if (!advisory) {
active.verdict = result.verdict;
}
const patch: Record<string, unknown> = {
[FOREACH_ACTIVE_CONTEXT_KEY]: active,
[`node:${node.id}:verdict`]: result.verdict,
};
const value =
result.verdict === "APPROVE"
? "approve"
: result.verdict === "REVISE"
? "revise"
: result.verdict === "RETHINK"
? "rethink"
: "unavailable";
return { outcome: "success", value, contextPatch: patch };
};
}
// ── parse-steps node (U12, KTD-12) ──────────────────────────────────────────
/** The implicit default step-source artifact when a workflow declares no
@@ -542,6 +707,8 @@ export function createCodeNodeHandler(runCode?: CodeNodeRunner): WorkflowNodeHan
}
export interface DefaultNodeHandlerDeps {
/** Workflow-native runtime primitives. When present they replace legacy seams. */
primitives?: WorkflowRuntimePrimitives;
/** parse-steps node deps (U12). When absent, a parse-steps node fails cleanly. */
parseSteps?: ParseStepsHandlerDeps;
/** code node runner (U14). When absent, a code node fails cleanly. */
@@ -566,7 +733,9 @@ export function createDefaultNodeHandlers(
| "pr-merge",
WorkflowNodeHandler
> {
const promptLike = createPromptLikeHandler(seams, runCustomNode);
const promptLike = deps?.primitives
? createPrimitivePromptLikeHandler(deps.primitives, runCustomNode)
: createPromptLikeHandler(seams, runCustomNode);
// parse-steps without deps fails closed (would otherwise have no handler at
// all and throw "No handler registered"); a clean failure is the safe posture.
const parseSteps: WorkflowNodeHandler = deps?.parseSteps
@@ -596,7 +765,9 @@ export function createDefaultNodeHandlers(
prompt: promptLike,
script: promptLike,
gate,
"step-review": createStepReviewHandler(seams),
"step-review": deps?.primitives
? createPrimitiveStepReviewHandler(deps.primitives)
: createStepReviewHandler(seams),
"parse-steps": parseSteps,
code: createCodeNodeHandler(deps?.runCode),
...prNodes,