Merge pull request #1536 from Runfusion/feature/workflow-mapping
feat(FN-6035): route execution through workflow primitives
This commit is contained in:
gsxdsm
@@ -36,7 +36,7 @@ describe("builtin coding workflow ir", () => {
|
||||
const seams = BUILTIN_CODING_WORKFLOW_IR.nodes
|
||||
.map((node) => String(node.config?.seam ?? ""))
|
||||
.filter((seam) => seam.length > 0);
|
||||
expect(seams).toEqual(expect.arrayContaining(["execute", "review", "merge"]));
|
||||
expect(seams).toEqual(expect.arrayContaining(["execute", "workflow-step", "review", "merge"]));
|
||||
expect(seams).not.toContain("triage");
|
||||
});
|
||||
|
||||
@@ -66,13 +66,15 @@ describe("builtin coding workflow ir", () => {
|
||||
it("places seam nodes in their columns", () => {
|
||||
const byId = new Map(BUILTIN_CODING_WORKFLOW_IR.nodes.map((n) => [n.id, n]));
|
||||
expect(byId.get("execute")?.column).toBe("in-progress");
|
||||
expect(byId.get("workflow-step")?.column).toBe("in-progress");
|
||||
expect(byId.get("review")?.column).toBe("in-review");
|
||||
expect(byId.get("merge")?.column).toBe("in-review");
|
||||
});
|
||||
|
||||
it("assigns descriptive names to execute/review/merge seam nodes", () => {
|
||||
it("assigns descriptive names to execute/workflow-step/review/merge seam nodes", () => {
|
||||
const byId = new Map(BUILTIN_CODING_WORKFLOW_IR.nodes.map((n) => [n.id, n]));
|
||||
expect(byId.get("execute")?.config?.name).toBe("Execute");
|
||||
expect(byId.get("workflow-step")?.config?.name).toBe("Pre-merge workflow steps");
|
||||
expect(byId.get("review")?.config?.name).toBe("Review");
|
||||
expect(byId.get("merge")?.config?.name).toBe("Merge boundary");
|
||||
});
|
||||
@@ -85,8 +87,10 @@ describe("builtin coding workflow ir", () => {
|
||||
expect(config.maxRetries).toBeLessThanOrEqual(10);
|
||||
|
||||
const byId = new Map(BUILTIN_CODING_WORKFLOW_IR.nodes.map((n) => [n.id, n]));
|
||||
expect(byId.get("workflow-step")?.config?.name).toBe("Pre-merge workflow steps");
|
||||
expect(byId.get("review")?.config?.name).toBe("Review");
|
||||
expect(byId.get("merge")?.config?.name).toBe("Merge boundary");
|
||||
expect(byId.get("workflow-step")?.config?.maxRetries).toBeUndefined();
|
||||
expect(byId.get("review")?.config?.maxRetries).toBeUndefined();
|
||||
expect(byId.get("merge")?.config?.maxRetries).toBeUndefined();
|
||||
});
|
||||
|
||||
@@ -133,6 +133,7 @@ describe("built-in workflows", () => {
|
||||
|
||||
const byId = new Map(ir.nodes.map((node) => [node.id, node]));
|
||||
expect(byId.get("execute")?.column).toBe("in-progress");
|
||||
expect(byId.get("workflow-step")?.column).toBe("in-progress");
|
||||
expect(byId.get("review")?.column).toBe("in-review");
|
||||
expect(byId.get("merge")?.column).toBe("in-review");
|
||||
expect(ir.settings).toEqual(BUILTIN_WORKFLOW_SETTINGS);
|
||||
@@ -169,8 +170,10 @@ describe("built-in workflows", () => {
|
||||
expect(executeConfig?.maxRetries).toBeLessThanOrEqual(10);
|
||||
|
||||
const byId = new Map(candidate.nodes.map((node) => [node.id, node]));
|
||||
expect(byId.get("workflow-step")?.config?.name).toBe("Pre-merge workflow steps");
|
||||
expect(byId.get("review")?.config?.name).toBe("Review");
|
||||
expect(byId.get("merge")?.config?.name).toBe("Merge boundary");
|
||||
expect(byId.get("workflow-step")?.config?.maxRetries).toBeUndefined();
|
||||
expect(byId.get("review")?.config?.maxRetries).toBeUndefined();
|
||||
expect(byId.get("merge")?.config?.maxRetries).toBeUndefined();
|
||||
}
|
||||
|
||||
@@ -145,7 +145,7 @@ describe("compileWorkflowToSteps (U2)", () => {
|
||||
expect(err).toBeInstanceOf(WorkflowCompileError);
|
||||
});
|
||||
|
||||
it("rejects seams that are out of the execute -> review -> merge order", () => {
|
||||
it("rejects seams that are out of the planning -> execute -> workflow-step -> review -> merge order", () => {
|
||||
const ir: WorkflowIr = {
|
||||
version: "v1",
|
||||
name: "misordered-seams",
|
||||
@@ -163,7 +163,7 @@ describe("compileWorkflowToSteps (U2)", () => {
|
||||
};
|
||||
const err = validateLinearity(ir);
|
||||
expect(err).toBeInstanceOf(WorkflowCompileError);
|
||||
expect(err?.message).toMatch(/execute -> review -> merge order/);
|
||||
expect(err?.message).toMatch(/planning -> execute -> workflow-step -> review -> merge order/);
|
||||
});
|
||||
|
||||
it("rejects a graph with a duplicated seam role", () => {
|
||||
|
||||
@@ -18,9 +18,11 @@ import { builtinPromptConfig } from "./builtin-workflow-prompts.js";
|
||||
* done = complete
|
||||
* archived = archived
|
||||
*
|
||||
* The seam nodes (execute/review/merge) are placed in their columns; the graph
|
||||
* walk (edges) is byte-identical to the prior v1 coding pipeline, so the graph
|
||||
* executor continues to drive execute → review → merge unchanged.
|
||||
* The lifecycle seam nodes are placed in their columns. Planning is explicit so
|
||||
* the built-in workflow owns the specification phase rather than relying on
|
||||
* triage code that runs outside the graph; workflow-step keeps the legacy
|
||||
* pre-merge quality gate between implementation and review; execute/review/
|
||||
* merge keep the same observable pipeline and failure routing.
|
||||
*/
|
||||
const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = {
|
||||
version: "v2",
|
||||
@@ -47,22 +49,40 @@ const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = {
|
||||
],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start", column: "triage" },
|
||||
{
|
||||
id: "planning",
|
||||
kind: "prompt",
|
||||
column: "triage",
|
||||
config: builtinPromptConfig("planning", "Plan / specify"),
|
||||
},
|
||||
{
|
||||
id: "execute",
|
||||
kind: "prompt",
|
||||
column: "in-progress",
|
||||
config: { ...builtinPromptConfig("execute", "Execute"), maxRetries: 2 },
|
||||
},
|
||||
{
|
||||
id: "workflow-step",
|
||||
kind: "prompt",
|
||||
column: "in-progress",
|
||||
config: builtinPromptConfig("workflow-step", "Pre-merge workflow steps"),
|
||||
},
|
||||
{ id: "review", kind: "prompt", column: "in-review", config: builtinPromptConfig("review", "Review") },
|
||||
{ id: "merge", kind: "prompt", column: "in-review", config: builtinPromptConfig("merge", "Merge boundary") },
|
||||
{ id: "end", kind: "end", column: "done" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "execute" },
|
||||
{ from: "execute", to: "review", condition: "success" },
|
||||
{ from: "start", to: "planning" },
|
||||
{ from: "planning", to: "execute", condition: "success" },
|
||||
{ from: "execute", to: "workflow-step", condition: "success" },
|
||||
{ from: "workflow-step", to: "review", condition: "success" },
|
||||
{ from: "workflow-step", to: "end", condition: "outcome:remediation-scheduled" },
|
||||
{ from: "workflow-step", to: "end", condition: "outcome:deferred-paused" },
|
||||
{ from: "review", to: "merge", condition: "success" },
|
||||
{ from: "merge", to: "end", condition: "success" },
|
||||
{ from: "planning", to: "end", condition: "failure" },
|
||||
{ from: "execute", to: "end", condition: "failure" },
|
||||
{ from: "workflow-step", to: "end", condition: "failure" },
|
||||
{ from: "review", to: "end", condition: "failure" },
|
||||
{ from: "merge", to: "end", condition: "failure" },
|
||||
],
|
||||
|
||||
@@ -9,6 +9,7 @@ const BUILTIN_SEAM_PROMPTS: Record<string, string> = {
|
||||
execute: DEFAULT_EXECUTOR_PROMPT,
|
||||
planning: DEFAULT_TRIAGE_PROMPT,
|
||||
"step-execute": DEFAULT_EXECUTOR_PROMPT,
|
||||
"workflow-step": DEFAULT_REVIEWER_PROMPT,
|
||||
review: DEFAULT_REVIEWER_PROMPT,
|
||||
merge: DEFAULT_MERGER_PROMPT,
|
||||
};
|
||||
|
||||
@@ -16,8 +16,9 @@ export class WorkflowCompileError extends Error {
|
||||
}
|
||||
|
||||
/** Seam anchor kinds, encoded on IR nodes as `config.seam`. These map to the
|
||||
* fixed execute → review → merge pipeline and are not emitted as steps. */
|
||||
const SEAM_NAMES = new Set(["execute", "review", "merge"]);
|
||||
* fixed planning → execute → workflow-step → review → merge pipeline and are
|
||||
* not emitted as steps. */
|
||||
const SEAM_NAMES = new Set(["planning", "execute", "workflow-step", "review", "merge"]);
|
||||
|
||||
function seamOf(node: WorkflowIrNode): string | undefined {
|
||||
const seam = node.config?.seam;
|
||||
@@ -77,7 +78,8 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null {
|
||||
const seam = seamOf(node);
|
||||
if (seam) {
|
||||
const failureEdges = outs.filter((edge) => edge.condition === "failure");
|
||||
const mainEdges = outs.filter((edge) => edge.condition !== "failure");
|
||||
const mainEdges = outs.filter((edge) => !edge.condition || edge.condition === "success");
|
||||
const outcomeEdges = outs.filter((edge) => edge.condition?.startsWith("outcome:"));
|
||||
if (mainEdges.length !== 1) {
|
||||
return new WorkflowCompileError(`seam '${node.id}' must have exactly one success path`);
|
||||
}
|
||||
@@ -87,6 +89,10 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null {
|
||||
if (failureEdges[0] && failureEdges[0].to !== endNode.id) {
|
||||
return new WorkflowCompileError(`seam '${node.id}' failure edge must target the end node`);
|
||||
}
|
||||
const nonTerminalOutcomeEdge = outcomeEdges.find((edge) => edge.to !== endNode.id);
|
||||
if (nonTerminalOutcomeEdge) {
|
||||
return new WorkflowCompileError(`seam '${node.id}' outcome edge must target the end node`);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -106,12 +112,12 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null {
|
||||
}
|
||||
|
||||
// Reachability: the single main path must reach end and cover every node.
|
||||
// While walking, enforce the canonical seam pipeline: each of execute/review/
|
||||
// merge may appear at most once and only in that order. The compiler treats
|
||||
// seams as a fixed execute → review → merge boundary (merge flips pre- to
|
||||
// post-merge), so out-of-order or duplicate seams would compile inconsistently
|
||||
// with the runtime contract.
|
||||
const expectedSeamOrder = ["execute", "review", "merge"] as const;
|
||||
// While walking, enforce the canonical seam pipeline: each of planning/
|
||||
// execute/workflow-step/review/merge may appear at most once and only in that
|
||||
// order. The compiler treats seams as a fixed lifecycle boundary (merge flips
|
||||
// pre- to post-merge), so out-of-order or duplicate seams would compile
|
||||
// inconsistently with the runtime contract.
|
||||
const expectedSeamOrder = ["planning", "execute", "workflow-step", "review", "merge"] as const;
|
||||
const seenSeams = new Set<string>();
|
||||
let nextExpectedSeamIndex = 0;
|
||||
const visited = new Set<string>();
|
||||
@@ -131,7 +137,9 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null {
|
||||
nextExpectedSeamIndex += 1;
|
||||
}
|
||||
if (expectedSeamOrder[nextExpectedSeamIndex] !== seam) {
|
||||
return new WorkflowCompileError("seams must follow the execute -> review -> merge order");
|
||||
return new WorkflowCompileError(
|
||||
"seams must follow the planning -> execute -> workflow-step -> review -> merge order",
|
||||
);
|
||||
}
|
||||
seenSeams.add(seam);
|
||||
nextExpectedSeamIndex += 1;
|
||||
|
||||
@@ -138,6 +138,7 @@ export const DEFAULT_WORKFLOW_COLUMN_IDS = [
|
||||
function defaultColumnForNode(node: WorkflowIrNode): string {
|
||||
const seam = node.config?.seam;
|
||||
if (seam === "execute") return "in-progress";
|
||||
if (seam === "workflow-step") return "in-progress";
|
||||
if (seam === "review") return "in-review";
|
||||
if (seam === "merge") return "in-review";
|
||||
return "todo";
|
||||
|
||||
57
packages/engine/src/__tests__/runtime-primitives.test.ts
Normal file
57
packages/engine/src/__tests__/runtime-primitives.test.ts
Normal file
@@ -0,0 +1,57 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { markSideEffectsStarted, primitiveNodeContext } from "../runtime-primitives.js";
|
||||
|
||||
describe("runtime primitives", () => {
|
||||
it("creates a workflow primitive context from a run and node", () => {
|
||||
const run = {
|
||||
runId: "run-1",
|
||||
taskId: "FN-1",
|
||||
workflowId: "coding",
|
||||
};
|
||||
const node = {
|
||||
id: "execute",
|
||||
kind: "prompt" as const,
|
||||
column: "in-progress",
|
||||
config: { prompt: "implement" },
|
||||
};
|
||||
|
||||
const ctx = primitiveNodeContext(run, node, {
|
||||
effectivePrincipalId: "agent:builder",
|
||||
attempt: 2,
|
||||
context: { priorOutcome: "revise" },
|
||||
});
|
||||
|
||||
expect(ctx).toEqual({
|
||||
run,
|
||||
node: {
|
||||
node,
|
||||
effectivePrincipalId: "agent:builder",
|
||||
attempt: 2,
|
||||
context: { priorOutcome: "revise" },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("marks side effects on an immutable context copy", () => {
|
||||
const ctx = primitiveNodeContext(
|
||||
{
|
||||
runId: "run-1",
|
||||
taskId: "FN-1",
|
||||
workflowId: "coding",
|
||||
},
|
||||
{ id: "execute", kind: "prompt" as const },
|
||||
);
|
||||
|
||||
const marked = markSideEffectsStarted(ctx);
|
||||
|
||||
expect(marked).toEqual({
|
||||
...ctx,
|
||||
run: {
|
||||
...ctx.run,
|
||||
sideEffectsStarted: true,
|
||||
},
|
||||
});
|
||||
expect(ctx.run.sideEffectsStarted).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -1,7 +1,8 @@
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// PARITY SUBJECT (test-file ownership, U7 / KTD-9):
|
||||
// This suite owns DEFAULT-WORKFLOW BYTE-IDENTITY parity — it proves the graph
|
||||
// executor reproduces the legacy monolithic execute → review → merge seam
|
||||
// executor reproduces the workflow-native planning → execute → workflow-step
|
||||
// → review → merge seam
|
||||
// sequence exactly (the parity ORACLE per KTD-1). It deliberately does NOT
|
||||
// cover per-step / updateStep-trajectory parity.
|
||||
//
|
||||
@@ -22,13 +23,27 @@ import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
|
||||
import type { WorkflowLegacySeams } from "../workflow-node-handlers.js";
|
||||
|
||||
const task = { id: "FN-5767" } as TaskDetail;
|
||||
type BaseSeam = "planning" | "execute" | "workflow-step" | "review" | "merge" | "schedule";
|
||||
|
||||
function runBaseSeam(seams: WorkflowLegacySeams, seam: BaseSeam, task: TaskDetail, context: Record<string, unknown>) {
|
||||
if (seam === "workflow-step") {
|
||||
return seams.workflowStep?.(task, context) ?? Promise.resolve({ outcome: "success" as const });
|
||||
}
|
||||
return seams[seam](task, context);
|
||||
}
|
||||
|
||||
function runLegacy(seams: WorkflowLegacySeams) {
|
||||
return async () => {
|
||||
const events: string[] = [];
|
||||
const planning = await seams.planning(task, {});
|
||||
events.push(`planning:${planning.outcome}`);
|
||||
if (planning.outcome !== "success") return events;
|
||||
const execute = await seams.execute(task, {});
|
||||
events.push(`execute:${execute.outcome}`);
|
||||
if (execute.outcome !== "success") return events;
|
||||
const workflowStep = await seams.workflowStep?.(task, {}) ?? { outcome: "success" as const };
|
||||
events.push(`workflow-step:${workflowStep.outcome}`);
|
||||
if (workflowStep.outcome !== "success") return events;
|
||||
const review = await seams.review(task, {});
|
||||
events.push(`review:${review.outcome}`);
|
||||
if (review.outcome !== "success") return events;
|
||||
@@ -47,20 +62,20 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
|
||||
expect(prompt).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("matches legacy execute-review-merge success path", async () => {
|
||||
it("matches default planning-execute-review-merge success path", async () => {
|
||||
const events: string[] = [];
|
||||
const seams: WorkflowLegacySeams = {
|
||||
planning: async () => ({ outcome: "success" }),
|
||||
execute: async () => ({ outcome: "success" }),
|
||||
workflowStep: async () => ({ outcome: "success" }),
|
||||
review: async () => ({ outcome: "success" }),
|
||||
merge: async () => ({ outcome: "success" }),
|
||||
schedule: async () => ({ outcome: "success" }),
|
||||
};
|
||||
const legacyEvents = await runLegacy(seams)();
|
||||
type BaseSeam = "planning" | "execute" | "review" | "merge" | "schedule";
|
||||
const executor = new WorkflowGraphExecutor({ seams, handlers: { prompt: async (node, ctx) => {
|
||||
const seam = String(node.config?.seam);
|
||||
const result = await seams[seam as BaseSeam](ctx.task, ctx.context);
|
||||
const seam = String(node.config?.seam) as BaseSeam;
|
||||
const result = await runBaseSeam(seams, seam, ctx.task, ctx.context);
|
||||
events.push(`${seam}:${result.outcome}`);
|
||||
return result;
|
||||
} } });
|
||||
@@ -74,6 +89,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
|
||||
const seams: WorkflowLegacySeams = {
|
||||
planning: async () => ({ outcome: "success" }),
|
||||
execute: async () => ({ outcome: "success" }),
|
||||
workflowStep: async () => ({ outcome: "success" }),
|
||||
review: async () => ({ outcome: "success" }),
|
||||
merge: async () => ({ outcome: "failure", value: "FileScopeViolationError" }),
|
||||
schedule: async () => ({ outcome: "success" }),
|
||||
@@ -82,7 +98,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
|
||||
const executor = new WorkflowGraphExecutor({ seams });
|
||||
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(legacyEvents).toEqual(["execute:success", "review:success", "merge:failure"]);
|
||||
expect(legacyEvents).toEqual(["planning:success", "execute:success", "workflow-step:success", "review:success", "merge:failure"]);
|
||||
});
|
||||
|
||||
it("preserves autoMerge:false terminal in-review semantics via review failure", async () => {
|
||||
@@ -159,18 +175,18 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
|
||||
const seams: WorkflowLegacySeams = {
|
||||
planning: async () => ({ outcome: "success" }),
|
||||
execute: async () => ({ outcome: "success" }),
|
||||
workflowStep: async () => ({ outcome: "success" }),
|
||||
review: async () => ({ outcome: "success" }),
|
||||
merge: async () => ({ outcome: "success" }),
|
||||
schedule: async () => ({ outcome: "success" }),
|
||||
};
|
||||
type BaseSeam = "planning" | "execute" | "review" | "merge" | "schedule";
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
seams,
|
||||
handlers: {
|
||||
prompt: async (node, ctx) => {
|
||||
const seam = String(node.config?.seam) as BaseSeam;
|
||||
stages.push(seam);
|
||||
return seams[seam](ctx.task, ctx.context);
|
||||
return runBaseSeam(seams, seam, ctx.task, ctx.context);
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -182,13 +198,16 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
|
||||
// Bind the invariant to actual executor behavior (PR #1432 review): the
|
||||
// observation below derives from the run-captured seam sequence, so seam
|
||||
// drift fails here instead of being masked by a hard-coded literal.
|
||||
expect(stages).toEqual(["execute", "review", "merge"]);
|
||||
expect(stages).toEqual(["planning", "execute", "workflow-step", "review", "merge"]);
|
||||
|
||||
// Legacy authoritative observation: a clean run that lands in `done`/merged.
|
||||
const legacyObs = buildWorkflowObservationFromTask(
|
||||
{ column: "done", status: "done", review: { verdict: "approve" } },
|
||||
{ columnSequence: ["todo", "in-progress", "in-review", "done"] },
|
||||
);
|
||||
const legacyObs = buildWorkflowObservation({
|
||||
stageTransitions: ["triage", "planning", "execute", "workflow-step", "review", "merge"],
|
||||
terminalColumn: "done",
|
||||
terminalStatus: "done",
|
||||
reviewVerdict: "approve",
|
||||
mergeOutcome: "merged",
|
||||
});
|
||||
// Interpreter (binding-free) observation assembled from the same run.
|
||||
const interpreterObs = buildWorkflowObservation({
|
||||
stageTransitions: ["triage", ...stages] as WorkflowStage[],
|
||||
|
||||
@@ -69,6 +69,7 @@ function recordingSeams(calls: string[], overrides: Partial<Record<string, Workf
|
||||
return {
|
||||
planning: seam("planning"),
|
||||
execute: seam("execute"),
|
||||
workflowStep: seam("workflow-step"),
|
||||
review: seam("review"),
|
||||
merge: seam("merge"),
|
||||
schedule: seam("schedule"),
|
||||
@@ -236,7 +237,7 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
|
||||
const result = await runner.run(task, flagOn);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
expect(calls).toEqual(["planning", "execute", "workflow-step", "review", "merge"]);
|
||||
expect(result.reason).toBeUndefined();
|
||||
expect(getWorkflowDefinition).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
@@ -3,7 +3,7 @@ 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";
|
||||
import type { PreparedWorktree, WorkflowRuntimePrimitives } from "../runtime-primitives.js";
|
||||
|
||||
const task = { id: "FN-9002" } as TaskDetail;
|
||||
const flagOff = { experimentalFeatures: {} } as unknown as Pick<Settings, "experimentalFeatures">;
|
||||
@@ -27,35 +27,96 @@ function selectedIr(): WorkflowIr {
|
||||
};
|
||||
}
|
||||
|
||||
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" };
|
||||
};
|
||||
function recordingPrimitives(
|
||||
calls: string[],
|
||||
overrides: Partial<Record<"prepare" | "execute" | "workflowStep", WorkflowNodeResult>> & {
|
||||
prepareData?: PreparedWorktree | null;
|
||||
} = {},
|
||||
observed: { prepared?: PreparedWorktree } = {},
|
||||
): WorkflowRuntimePrimitives {
|
||||
const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" };
|
||||
return {
|
||||
planning: seam("planning"),
|
||||
execute: seam("execute"),
|
||||
review: seam("review"),
|
||||
merge: seam("merge"),
|
||||
schedule: seam("schedule"),
|
||||
prepareWorktree: async () => {
|
||||
calls.push("prepare-worktree");
|
||||
return {
|
||||
outcome: overrides.prepare?.outcome ?? "success",
|
||||
value: overrides.prepare?.value,
|
||||
contextPatch: overrides.prepare?.contextPatch,
|
||||
data: overrides.prepare?.outcome === "failure"
|
||||
? undefined
|
||||
: overrides.prepareData === null
|
||||
? undefined
|
||||
: overrides.prepareData ?? prepared,
|
||||
};
|
||||
},
|
||||
readArtifact: async () => undefined,
|
||||
writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }),
|
||||
runPlanningSession: async () => {
|
||||
calls.push("planning");
|
||||
return { outcome: "success", data: { approved: true, artifactKeys: [] } };
|
||||
},
|
||||
runCodingSession: async (_ctx, _task, preparedWorktree) => {
|
||||
calls.push("execute");
|
||||
observed.prepared = preparedWorktree;
|
||||
const override = overrides.execute;
|
||||
return {
|
||||
outcome: override?.outcome ?? "success",
|
||||
value: override?.value ?? "implemented",
|
||||
contextPatch: override?.contextPatch,
|
||||
data: { taskDone: override?.outcome !== "failure", modifiedFiles: [] },
|
||||
};
|
||||
},
|
||||
runTaskStep: async () => ({ outcome: "success" }),
|
||||
resetTaskStep: async () => ({ ok: true }),
|
||||
runReview: async (_ctx, _task, input) => {
|
||||
calls.push(input.stepIndex === undefined ? "review" : "step-review");
|
||||
return {
|
||||
outcome: "success",
|
||||
value: input.stepIndex === undefined ? "in-review" : "approve",
|
||||
data: { verdict: "APPROVE" },
|
||||
};
|
||||
},
|
||||
runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }),
|
||||
runWorkflowStep: async () => {
|
||||
calls.push("workflow-step");
|
||||
const override = overrides.workflowStep;
|
||||
return {
|
||||
outcome: override?.outcome ?? "success",
|
||||
value: override?.value ?? "workflow-steps-passed",
|
||||
contextPatch: override?.contextPatch,
|
||||
data: { allPassed: override?.value !== "remediation-scheduled" },
|
||||
};
|
||||
},
|
||||
updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }),
|
||||
transitionTask: async () => {
|
||||
calls.push("schedule");
|
||||
return { outcome: "success" };
|
||||
},
|
||||
requestMerge: async () => {
|
||||
calls.push("merge");
|
||||
return { outcome: "success", value: "merged", data: { status: "merged" } };
|
||||
},
|
||||
abortRun: async () => ({ outcome: "success" }),
|
||||
audit: () => undefined,
|
||||
};
|
||||
}
|
||||
|
||||
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 = {
|
||||
// @ts-expect-error WorkflowTaskRuntime is an execution entry point, so primitives are required.
|
||||
const missingPrimitives: WorkflowTaskRuntimeDeps = {
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
};
|
||||
expect(missingSeams).toBeDefined();
|
||||
expect(missingPrimitives).toBeDefined();
|
||||
});
|
||||
|
||||
it("runs a selected workflow through the graph engine", async () => {
|
||||
const calls: string[] = [];
|
||||
const observed: { prepared?: PreparedWorktree } = {};
|
||||
let workflowSelectionReads = 0;
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
@@ -65,7 +126,14 @@ describe("WorkflowTaskRuntime", () => {
|
||||
},
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
primitives: recordingPrimitives(
|
||||
calls,
|
||||
{
|
||||
prepare: { outcome: "success", contextPatch: { preparedKey: "from-prepare" } },
|
||||
execute: { outcome: "success", contextPatch: { executeKey: "from-execute" } },
|
||||
},
|
||||
observed,
|
||||
),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
@@ -75,11 +143,38 @@ describe("WorkflowTaskRuntime", () => {
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["custom:prepare", "execute"]);
|
||||
expect(calls).toEqual(["custom:prepare", "prepare-worktree", "execute"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "prepare", "execute"]);
|
||||
expect(observed.prepared).toEqual({ worktreePath: "/tmp/fusion-worktree" });
|
||||
expect(result.context.preparedKey).toBe("from-prepare");
|
||||
expect(result.context.executeKey).toBe("from-execute");
|
||||
expect(workflowSelectionReads).toBe(1);
|
||||
});
|
||||
|
||||
it("fails execute instead of skipping coding when prepare succeeds without worktree data", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
primitives: recordingPrimitives(calls, {
|
||||
prepare: { outcome: "success", value: "prepared-without-data" },
|
||||
prepareData: null,
|
||||
}),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(calls).toEqual(["custom:prepare", "prepare-worktree"]);
|
||||
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({
|
||||
@@ -87,7 +182,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
primitives: recordingPrimitives(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
@@ -97,60 +192,67 @@ describe("WorkflowTaskRuntime", () => {
|
||||
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"]);
|
||||
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "workflow-step", "review", "merge"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "workflow-step", "review", "merge"]);
|
||||
});
|
||||
|
||||
it("turns selected workflow lookup failures into the built-in workflow target", async () => {
|
||||
it("stops the built-in workflow before review when workflow-step remediation is scheduled", async () => {
|
||||
const calls: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
primitives: recordingPrimitives(calls, {
|
||||
workflowStep: { outcome: "success", value: "remediation-scheduled" },
|
||||
}),
|
||||
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(["planning", "prepare-worktree", "execute", "workflow-step"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "workflow-step"]);
|
||||
});
|
||||
|
||||
it("fails selected workflow lookup misses instead of running the built-in workflow", async () => {
|
||||
const calls: string[] = [];
|
||||
const observedRunIds: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-MISSING", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
primitives: recordingPrimitives(calls),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
observedRunIds.push(runId);
|
||||
return [];
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
expect(observedRunIds).toContain("FN-9002:builtin:coding");
|
||||
expect(observedRunIds).not.toContain("FN-9002:WF-MISSING");
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toContain("workflow-resolution-error: workflow-missing: WF-MISSING");
|
||||
expect(calls).toEqual([]);
|
||||
});
|
||||
|
||||
it("turns corrupt selected workflow definitions into the built-in workflow target", async () => {
|
||||
it("fails corrupt selected workflow definitions instead of running the built-in workflow", async () => {
|
||||
const calls: string[] = [];
|
||||
const observedRunIds: string[] = [];
|
||||
const runtime = new WorkflowTaskRuntime({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-CORRUPT", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: "not a workflow ir" }),
|
||||
},
|
||||
seams: recordingSeams(calls),
|
||||
primitives: recordingPrimitives(calls),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
observedRunIds.push(runId);
|
||||
return [];
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runtime.run(task, flagOff);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["execute", "review", "merge"]);
|
||||
expect(observedRunIds).toContain("FN-9002:builtin:coding");
|
||||
expect(observedRunIds).not.toContain("FN-9002:WF-CORRUPT");
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toContain("workflow-resolution-error:");
|
||||
expect(calls).toEqual([]);
|
||||
});
|
||||
|
||||
it("forces only the graph executor flag while preserving other settings", async () => {
|
||||
@@ -160,7 +262,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
handlers: {
|
||||
prompt: async (_node, context) => {
|
||||
@@ -189,7 +291,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
@@ -211,7 +313,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => undefined,
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
branchPersistence: {
|
||||
loadBranchStates: (_taskId, runId) => {
|
||||
@@ -233,7 +335,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams(calls, { execute: { outcome: "failure", value: "implementation-incomplete" } }),
|
||||
primitives: recordingPrimitives(calls, { execute: { outcome: "failure", value: "implementation-incomplete" } }),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
@@ -244,7 +346,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(calls).toEqual(["custom:prepare", "execute"]);
|
||||
expect(calls).toEqual(["custom:prepare", "prepare-worktree", "execute"]);
|
||||
});
|
||||
|
||||
it("converts interpreter throws into workflow-engine failures", async () => {
|
||||
@@ -262,7 +364,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: badIr }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
@@ -292,7 +394,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: cyclicIr }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
|
||||
@@ -309,7 +411,7 @@ describe("WorkflowTaskRuntime", () => {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
|
||||
},
|
||||
seams: recordingSeams([]),
|
||||
primitives: recordingPrimitives([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
onEvent: () => {
|
||||
throw new Error("diagnostics failed");
|
||||
|
||||
@@ -9,7 +9,7 @@ import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "n
|
||||
import { existsSync, realpathSync } from "node:fs";
|
||||
import { readFile, rm, writeFile } from "node:fs/promises";
|
||||
import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core";
|
||||
import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, isWorkflowColumnsEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry } from "@fusion/core";
|
||||
import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, isWorkflowColumnsEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow } from "@fusion/core";
|
||||
import { mergeEffectiveSettings } from "./effective-settings.js";
|
||||
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult } from "@fusion/core";
|
||||
import {
|
||||
@@ -30,10 +30,17 @@ import { observeWorkflowParity, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from ".
|
||||
import {
|
||||
FOREACH_ACTIVE_CONTEXT_KEY,
|
||||
SEAM_GOVERNING_NODE_CONTEXT_KEY,
|
||||
SPLIT_ACTIVE_CONTEXT_KEY,
|
||||
type ForeachActiveContext,
|
||||
type WorkflowLegacySeams,
|
||||
} from "./workflow-node-handlers.js";
|
||||
import type { WorkflowNodeResult } from "./workflow-graph-executor.js";
|
||||
import type {
|
||||
AuditPrimitiveInput,
|
||||
PreparedWorktree,
|
||||
WorkflowPrimitiveContext,
|
||||
WorkflowRuntimePrimitives,
|
||||
} from "./runtime-primitives.js";
|
||||
import {
|
||||
ApprovalRequestStore,
|
||||
buildExecutionMemoryInstructions,
|
||||
@@ -3630,9 +3637,9 @@ export class TaskExecutor {
|
||||
private static processWideGraphRouting = new Set<string>();
|
||||
|
||||
/** Wired by the runtime to ProjectEngine.onMerge — resolves with the merge outcome. */
|
||||
private mergeRequester?: (taskId: string) => Promise<MergeResult>;
|
||||
private mergeRequester?: (taskId: string, options?: { signal?: AbortSignal }) => Promise<MergeResult>;
|
||||
|
||||
setMergeRequester(requestMerge: (taskId: string) => Promise<MergeResult>): void {
|
||||
setMergeRequester(requestMerge: (taskId: string, options?: { signal?: AbortSignal }) => Promise<MergeResult>): void {
|
||||
this.mergeRequester = requestMerge;
|
||||
}
|
||||
|
||||
@@ -3649,19 +3656,38 @@ export class TaskExecutor {
|
||||
let settings: Settings;
|
||||
try {
|
||||
settings = await this.store.getSettings();
|
||||
} catch {
|
||||
return false;
|
||||
} catch (err) {
|
||||
await this.handleGraphFailure(task, {
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
reason: `settings-load-failed: ${err instanceof Error ? err.message : String(err)}`,
|
||||
visitedNodeIds: [],
|
||||
});
|
||||
return true;
|
||||
}
|
||||
if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) return false;
|
||||
if (typeof this.store.getTaskWorkflowSelection !== "function") return false;
|
||||
|
||||
const hasWorkflowResolver = typeof this.store.getTaskWorkflowSelection === "function";
|
||||
const explicitlyEnabled = isExperimentalFeatureEnabled(settings, "workflowGraphExecutor");
|
||||
if (!hasWorkflowResolver && !explicitlyEnabled) return false;
|
||||
settings = {
|
||||
...settings,
|
||||
experimentalFeatures: {
|
||||
...(settings.experimentalFeatures ?? {}),
|
||||
workflowGraphExecutor: true,
|
||||
},
|
||||
};
|
||||
let selection: { workflowId: string; stepIds: string[] } | undefined;
|
||||
try {
|
||||
selection = this.store.getTaskWorkflowSelection(task.id);
|
||||
} catch {
|
||||
return false;
|
||||
selection = this.store.getTaskWorkflowSelection?.(task.id);
|
||||
} catch (err) {
|
||||
await this.handleGraphFailure(task, {
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
reason: `workflow-selection-failed: ${err instanceof Error ? err.message : String(err)}`,
|
||||
visitedNodeIds: [],
|
||||
});
|
||||
return true;
|
||||
}
|
||||
if (!selection) return false;
|
||||
selection ??= { workflowId: "builtin:coding", stepIds: [] };
|
||||
|
||||
// Resolve the production run id ONCE, here, so it is the single source of
|
||||
// truth shared by the runner AND the executor-side persistence deps
|
||||
@@ -3673,7 +3699,9 @@ export class TaskExecutor {
|
||||
// prior behavior — so this never strands a task.
|
||||
let resolvedRunId: string | undefined;
|
||||
try {
|
||||
const definition = await this.store.getWorkflowDefinition?.(selection.workflowId);
|
||||
const definition = selection.workflowId === "builtin:coding"
|
||||
? { id: "builtin:coding" }
|
||||
: await this.store.getWorkflowDefinition?.(selection.workflowId);
|
||||
if (definition) resolvedRunId = `${task.id}:${definition.id}`;
|
||||
} catch {
|
||||
// Definition load failure — leave undefined; deps/runner use fallbacks.
|
||||
@@ -3710,8 +3738,16 @@ export class TaskExecutor {
|
||||
}
|
||||
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: this.store,
|
||||
store: {
|
||||
...this.store,
|
||||
getTaskWorkflowSelection: (taskId: string) =>
|
||||
this.store.getTaskWorkflowSelection?.(taskId) ?? { workflowId: "builtin:coding", stepIds: [] },
|
||||
getWorkflowDefinition: async (id: string) =>
|
||||
(await this.store.getWorkflowDefinition?.(id))
|
||||
?? (id === "builtin:coding" ? getBuiltinWorkflow("builtin:coding") : undefined),
|
||||
},
|
||||
runId: resolvedRunId,
|
||||
primitives: this.createAuthoritativeWorkflowPrimitives(settings),
|
||||
seams: this.createAuthoritativeWorkflowSeams(settings),
|
||||
runCustomNode: (node, nodeTask) =>
|
||||
this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)),
|
||||
@@ -3758,16 +3794,26 @@ export class TaskExecutor {
|
||||
const detail = await this.store.getTask(task.id);
|
||||
result = await runner.run(detail, settings);
|
||||
} catch (err) {
|
||||
// A thrown interpreter error must not strand the task in-progress: fall
|
||||
// back to the legacy pipeline so the normal executor lock + flow runs.
|
||||
executorLog.error(
|
||||
`[workflow-graph] ${task.id} interpreter threw — falling back to legacy pipeline: ${err instanceof Error ? err.message : String(err)}`,
|
||||
`[workflow-graph] ${task.id} interpreter threw — parking task as workflow failure: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
return false;
|
||||
await this.handleGraphFailure(task, {
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
reason: `interpreter-error: ${err instanceof Error ? err.message : String(err)}`,
|
||||
visitedNodeIds: [],
|
||||
});
|
||||
return true;
|
||||
}
|
||||
if (result.disposition === "fell-back") {
|
||||
executorLog.log(`[workflow-graph] ${task.id} fell back to legacy pipeline: ${result.reason}`);
|
||||
return false;
|
||||
executorLog.warn(`[workflow-graph] ${task.id} could not resolve workflow — parking task instead of legacy fallback: ${result.reason}`);
|
||||
await this.handleGraphFailure(task, {
|
||||
...result,
|
||||
disposition: "failed",
|
||||
outcome: "failure",
|
||||
reason: result.reason ?? "workflow-resolution-failed",
|
||||
});
|
||||
return true;
|
||||
}
|
||||
if (result.disposition === "failed") {
|
||||
await this.handleGraphFailure(task, result);
|
||||
@@ -4415,13 +4461,23 @@ export class TaskExecutor {
|
||||
* interceptor makes execute() stop at the completion boundary instead of
|
||||
* running workflow steps and the review handoff.
|
||||
*/
|
||||
private async runImplementationPhase(task: Task): Promise<{ taskDone: boolean; modifiedFiles: string[] }> {
|
||||
private async runImplementationPhase(
|
||||
task: Task,
|
||||
prepared?: PreparedWorktree,
|
||||
): Promise<{ taskDone: boolean; modifiedFiles: string[] }> {
|
||||
let captured: { taskDone: boolean; modifiedFiles: string[] } = { taskDone: false, modifiedFiles: [] };
|
||||
this.graphCompletionInterceptors.set(task.id, (info) => {
|
||||
captured = { taskDone: true, modifiedFiles: info.modifiedFiles };
|
||||
});
|
||||
const executionTask = prepared
|
||||
? {
|
||||
...task,
|
||||
worktree: prepared.worktreePath || task.worktree,
|
||||
branch: prepared.branchName || task.branch,
|
||||
}
|
||||
: task;
|
||||
try {
|
||||
await this.execute(task);
|
||||
await this.execute(executionTask);
|
||||
} finally {
|
||||
this.graphCompletionInterceptors.delete(task.id);
|
||||
}
|
||||
@@ -4553,6 +4609,304 @@ export class TaskExecutor {
|
||||
|
||||
/** Public authoritative-driver seam factory: exposes the same real lifecycle
|
||||
* seams the internal graph runner uses, without changing legacy behavior. */
|
||||
public createAuthoritativeWorkflowPrimitives(settings: Settings): WorkflowRuntimePrimitives {
|
||||
const logAudit = async (taskId: string | undefined, input: AuditPrimitiveInput): Promise<void> => {
|
||||
if (!taskId) return;
|
||||
try {
|
||||
await this.store.logEntry(taskId, input.message, input.metadata ? JSON.stringify(input.metadata) : undefined);
|
||||
} catch {
|
||||
// Audit is diagnostic-only and must not affect workflow execution.
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
prepareWorktree: async (_ctx, task) => {
|
||||
const live = await this.store.getTask(task.id);
|
||||
const prepared: PreparedWorktree = {
|
||||
worktreePath: live.worktree || this.rootDir,
|
||||
branchName: live.branch,
|
||||
};
|
||||
return { outcome: "success", value: "worktree-ready", data: prepared };
|
||||
},
|
||||
readArtifact: async (_ctx, task, key) => {
|
||||
const deps = this.buildParseStepsDeps(`${task.id}:artifact-read`);
|
||||
return deps.readArtifact(task, key);
|
||||
},
|
||||
writeArtifact: async (ctx, task, key, content) => {
|
||||
const writer = (this.store as unknown as {
|
||||
writeTaskDocument?: (taskId: string, key: string, content: string) => Promise<void>;
|
||||
}).writeTaskDocument;
|
||||
if (!writer) {
|
||||
await logAudit(task.id, {
|
||||
type: "artifact-write-unavailable",
|
||||
message: `Workflow node ${ctx.node.node.id} could not write artifact ${key}: store writer unavailable`,
|
||||
});
|
||||
return { outcome: "failure", value: "artifact-write-unavailable" };
|
||||
}
|
||||
await writer.call(this.store, task.id, key, content);
|
||||
return { outcome: "success", value: "artifact-written", data: { key } };
|
||||
},
|
||||
runPlanningSession: async () => ({ outcome: "success", value: "pre-specified", data: {
|
||||
approved: true,
|
||||
artifactKeys: [],
|
||||
} }),
|
||||
runCodingSession: async (ctx, task, prepared) => {
|
||||
const governingNodeId = ctx.node.context?.[SEAM_GOVERNING_NODE_CONTEXT_KEY];
|
||||
if (typeof governingNodeId === "string") {
|
||||
this.graphSeamGoverningNodeId.set(task.id, governingNodeId);
|
||||
}
|
||||
let result: { taskDone: boolean; modifiedFiles: string[] };
|
||||
try {
|
||||
result = await this.runImplementationPhase(task, prepared);
|
||||
} finally {
|
||||
this.graphSeamGoverningNodeId.delete(task.id);
|
||||
}
|
||||
if (result.taskDone) {
|
||||
return { outcome: "success", value: "implemented", data: result };
|
||||
}
|
||||
let paused = this.pausedAborted.has(task.id);
|
||||
if (!paused) {
|
||||
try {
|
||||
paused = Boolean((await this.store.getTask(task.id)).paused);
|
||||
} catch {
|
||||
// Best-effort pause probe; fall through to the failure value.
|
||||
}
|
||||
}
|
||||
return {
|
||||
outcome: "failure",
|
||||
value: paused ? "implementation-paused" : "implementation-incomplete",
|
||||
data: result,
|
||||
};
|
||||
},
|
||||
runTaskStep: async (ctx, task, stepIndex) => {
|
||||
const context = ctx.node.context ?? {};
|
||||
const active = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined;
|
||||
if (!active || typeof active.stepIndex !== "number") {
|
||||
return { outcome: "failure" };
|
||||
}
|
||||
const live = await this.store.getTask(task.id);
|
||||
const worktreePath = active.worktreePath || live.worktree || this.rootDir;
|
||||
this.graphStepActiveContext.set(this.graphActiveContextKey(task.id, active.instanceId), active);
|
||||
const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY];
|
||||
return await runTaskStep(
|
||||
{
|
||||
store: this.store,
|
||||
worktreePath,
|
||||
runStep: (idx) =>
|
||||
this.runGraphTaskStep(
|
||||
task,
|
||||
idx,
|
||||
active.instanceId,
|
||||
typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined,
|
||||
),
|
||||
},
|
||||
{ id: task.id, steps: live.steps },
|
||||
stepIndex,
|
||||
{ markDoneOnSuccess: active.deferDoneToReview !== true },
|
||||
);
|
||||
},
|
||||
resetTaskStep: async (ctx, task, stepIndex, baselineSha, checkpointId) => {
|
||||
const active = ctx.node.context?.[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined;
|
||||
const branchScoped = typeof active?.worktreePath === "string" && active.worktreePath.length > 0;
|
||||
let worktreePath = active?.worktreePath ?? this.rootDir;
|
||||
if (!branchScoped) {
|
||||
try {
|
||||
worktreePath = (await this.store.getTask(task.id)).worktree || this.rootDir;
|
||||
} catch {
|
||||
// Best-effort worktree resolution; fall back to rootDir.
|
||||
}
|
||||
}
|
||||
const liveSteps = await this.store.getTask(task.id).then((t) => t.steps).catch(() => []);
|
||||
return await resetStepToBaseline(
|
||||
{
|
||||
store: this.store,
|
||||
worktreePath,
|
||||
sessionRef: { current: null },
|
||||
reviewType: "code",
|
||||
blastRadiusGuard: branchScoped
|
||||
? undefined
|
||||
: makeAncestryBlastRadiusGuard({
|
||||
worktreePath,
|
||||
task: { id: task.id, steps: liveSteps },
|
||||
stepIndex,
|
||||
}),
|
||||
},
|
||||
{ id: task.id, steps: liveSteps },
|
||||
stepIndex,
|
||||
baselineSha,
|
||||
checkpointId,
|
||||
);
|
||||
},
|
||||
runReview: async (ctx, task, input) => {
|
||||
if (typeof input.stepIndex === "number") {
|
||||
const context = ctx.node.context ?? {};
|
||||
const active = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined;
|
||||
if (!active || typeof active.stepIndex !== "number") {
|
||||
return {
|
||||
outcome: "success",
|
||||
value: "unavailable",
|
||||
data: { verdict: "UNAVAILABLE", review: "no active step instance" },
|
||||
};
|
||||
}
|
||||
const config = {
|
||||
type: input.type,
|
||||
advisory: context[SPLIT_ACTIVE_CONTEXT_KEY] === true,
|
||||
} as const;
|
||||
const seamResult = await this.createAuthoritativeWorkflowSeams(settings).stepReview?.(
|
||||
task,
|
||||
context,
|
||||
config,
|
||||
);
|
||||
return {
|
||||
outcome: "success",
|
||||
value: seamResult?.verdict === "APPROVE" ? "approve" : seamResult?.verdict === "REVISE" ? "revise" : seamResult?.verdict === "RETHINK" ? "rethink" : "unavailable",
|
||||
data: seamResult ?? { verdict: "UNAVAILABLE", review: "step review unavailable" },
|
||||
};
|
||||
}
|
||||
const live = await this.store.getTask(task.id);
|
||||
await this.persistTokenUsage(task.id);
|
||||
await this.handoffTaskToReview(live, "workflow-graph-review");
|
||||
return {
|
||||
outcome: "success",
|
||||
value: "in-review",
|
||||
data: { verdict: "APPROVE", summary: "Task handed off for merge review" },
|
||||
};
|
||||
},
|
||||
runVerification: async () => ({ outcome: "success", value: "verification-skipped", data: {
|
||||
verdict: "skipped",
|
||||
} }),
|
||||
runWorkflowStep: async (_ctx, task, input) => {
|
||||
if (input.phase !== "pre-merge") {
|
||||
return { outcome: "success", value: "workflow-step-skipped", data: { allPassed: true } };
|
||||
}
|
||||
const live = await this.store.getTask(task.id);
|
||||
if (live.executionMode === "fast") {
|
||||
executorLog.log(`${task.id}: fast mode — skipping pre-merge workflow steps`);
|
||||
await this.store.logEntry(task.id, "Fast mode — pre-merge workflow steps skipped", undefined, this.getRunContextFor(task.id));
|
||||
return { outcome: "success", value: "workflow-step-skipped", data: { allPassed: true } };
|
||||
}
|
||||
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after task completion")) {
|
||||
return { outcome: "success", value: "deferred-paused", data: { allPassed: false } };
|
||||
}
|
||||
const worktreePath = input.worktreePath || live.worktree || this.rootDir;
|
||||
const workflowResult = await this.runWorkflowSteps(live, worktreePath, settings, undefined);
|
||||
if (workflowResult === "deferred-paused") {
|
||||
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
|
||||
this.pausedAborted.delete(task.id);
|
||||
} else if (this.pausedAborted.has(task.id)) {
|
||||
this.pausedAborted.delete(task.id);
|
||||
}
|
||||
return { outcome: "success", value: "deferred-paused", data: { allPassed: false } };
|
||||
}
|
||||
if (!workflowResult.allPassed) {
|
||||
const feedback = workflowResult.feedback || "Workflow step failed";
|
||||
const stepName = workflowResult.stepName || "Unknown";
|
||||
if (workflowResult.revisionRequested) {
|
||||
const rerunScheduled = await this.handleWorkflowRevisionRequest(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
settings,
|
||||
);
|
||||
if (!rerunScheduled) {
|
||||
return {
|
||||
outcome: "failure",
|
||||
value: "workflow-step-revision-unhandled",
|
||||
data: workflowResult,
|
||||
};
|
||||
}
|
||||
} else {
|
||||
const retried = await this.handleWorkflowStepFailure(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
);
|
||||
if (!retried) {
|
||||
await this.sendTaskBackForFix(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
"Workflow step failed",
|
||||
);
|
||||
}
|
||||
}
|
||||
return { outcome: "success", value: "remediation-scheduled", data: workflowResult };
|
||||
}
|
||||
await this.store.updateTask(task.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
|
||||
return { outcome: "success", value: "workflow-steps-passed", data: workflowResult };
|
||||
},
|
||||
updateSteps: async (_ctx, task, steps) => {
|
||||
await this.store.updateTask(task.id, { steps });
|
||||
return { outcome: "success", value: "steps-updated", data: { count: steps.length } };
|
||||
},
|
||||
transitionTask: async (_ctx, task, input) => {
|
||||
const patch: Partial<TaskDetail> = {};
|
||||
if (input.column !== undefined) patch.column = input.column;
|
||||
if (input.status !== undefined && input.status !== null) patch.status = input.status;
|
||||
if (Object.keys(patch).length > 0) {
|
||||
await this.store.updateTask(task.id, patch);
|
||||
}
|
||||
return { outcome: "success", value: input.reason };
|
||||
},
|
||||
requestMerge: async (ctx, task) => {
|
||||
if (!this.mergeRequester) {
|
||||
return { outcome: "failure", value: "merge-unavailable", data: { status: "failed", reason: "merge-unavailable" } };
|
||||
}
|
||||
const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000;
|
||||
const controller = new AbortController();
|
||||
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeout = new Promise<"timeout">((resolve) => {
|
||||
timeoutHandle = setTimeout(() => {
|
||||
controller.abort();
|
||||
resolve("timeout");
|
||||
}, GRAPH_MERGE_TIMEOUT_MS);
|
||||
timeoutHandle.unref?.();
|
||||
});
|
||||
try {
|
||||
const result = await Promise.race([this.mergeRequester(task.id, { signal: controller.signal }), timeout]);
|
||||
if (result === "timeout") {
|
||||
executorLog.warn(`${task.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`);
|
||||
return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } };
|
||||
}
|
||||
if (result.merged || result.noOp) {
|
||||
return {
|
||||
outcome: "success",
|
||||
value: result.noOp ? "merge-noop" : "merged",
|
||||
data: { status: "merged", noOp: result.noOp },
|
||||
};
|
||||
}
|
||||
return {
|
||||
outcome: "failure",
|
||||
value: result.reason ?? result.error ?? "merge-failed",
|
||||
data: { status: "failed", reason: result.reason ?? result.error ?? "merge-failed" },
|
||||
};
|
||||
} finally {
|
||||
if (timeoutHandle) clearTimeout(timeoutHandle);
|
||||
await logAudit(task.id, {
|
||||
type: "merge-requested",
|
||||
message: `Workflow node ${ctx.node.node.id} requested merge`,
|
||||
});
|
||||
}
|
||||
},
|
||||
abortRun: async (_ctx, task, input) => {
|
||||
if (input.hardCancel) {
|
||||
this.pausedAborted.add(task.id);
|
||||
}
|
||||
await this.store.updateTask(task.id, {
|
||||
paused: true,
|
||||
pausedReason: input.reason,
|
||||
} as Partial<TaskDetail>);
|
||||
return { outcome: "success", value: "aborted" };
|
||||
},
|
||||
audit: async (ctx: WorkflowPrimitiveContext, input) => {
|
||||
await logAudit(ctx.run.taskId, input);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
public createAuthoritativeWorkflowSeams(_settings: Settings): WorkflowLegacySeams {
|
||||
return {
|
||||
// Built-in triage/spec generation runs upstream of the interpreter today,
|
||||
@@ -4593,6 +4947,58 @@ export class TaskExecutor {
|
||||
value: paused ? "implementation-paused" : "implementation-incomplete",
|
||||
};
|
||||
},
|
||||
workflowStep: async (seamTask) => {
|
||||
const live = await this.store.getTask(seamTask.id);
|
||||
if (live.executionMode === "fast") {
|
||||
executorLog.log(`${seamTask.id}: fast mode — skipping pre-merge workflow steps`);
|
||||
await this.store.logEntry(seamTask.id, "Fast mode — pre-merge workflow steps skipped", undefined, this.getRunContextFor(seamTask.id));
|
||||
return { outcome: "success", value: "workflow-step-skipped" };
|
||||
}
|
||||
const worktreePath = live.worktree || this.rootDir;
|
||||
const settings = await this.store.getSettings();
|
||||
const workflowResult = await this.runWorkflowSteps(live, worktreePath, settings, undefined);
|
||||
if (workflowResult === "deferred-paused") {
|
||||
if (await this.parkTaskAfterWorkflowStepPause(seamTask.id)) {
|
||||
this.pausedAborted.delete(seamTask.id);
|
||||
} else if (this.pausedAborted.has(seamTask.id)) {
|
||||
this.pausedAborted.delete(seamTask.id);
|
||||
}
|
||||
return { outcome: "success", value: "deferred-paused" };
|
||||
}
|
||||
if (!workflowResult.allPassed) {
|
||||
const feedback = workflowResult.feedback || "Workflow step failed";
|
||||
const stepName = workflowResult.stepName || "Unknown";
|
||||
if (workflowResult.revisionRequested) {
|
||||
const rerunScheduled = await this.handleWorkflowRevisionRequest(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
settings,
|
||||
);
|
||||
if (!rerunScheduled) return { outcome: "failure", value: "workflow-step-revision-unhandled" };
|
||||
} else {
|
||||
const retried = await this.handleWorkflowStepFailure(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
);
|
||||
if (!retried) {
|
||||
await this.sendTaskBackForFix(
|
||||
live,
|
||||
worktreePath,
|
||||
feedback,
|
||||
stepName,
|
||||
"Workflow step failed",
|
||||
);
|
||||
}
|
||||
}
|
||||
return { outcome: "success", value: "remediation-scheduled" };
|
||||
}
|
||||
await this.store.updateTask(seamTask.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
|
||||
return { outcome: "success", value: "workflow-steps-passed" };
|
||||
},
|
||||
review: async (seamTask) => {
|
||||
// The legacy "review" stage is the in-review handoff: per-step AI review
|
||||
// already ran during implementation (fn_review_step), and the in-review
|
||||
@@ -5710,10 +6116,10 @@ export class TaskExecutor {
|
||||
executorLog.log(`execute() called for ${task.id} while graph routing is active — skipping duplicate`);
|
||||
return;
|
||||
}
|
||||
const authoritativeOwned = await this.options.workflowAuthoritativeDispatch?.(task);
|
||||
if (authoritativeOwned) return;
|
||||
const graphOwned = await this.maybeExecuteWorkflowGraph(task);
|
||||
if (graphOwned) return;
|
||||
const authoritativeOwned = await this.options.workflowAuthoritativeDispatch?.(task);
|
||||
if (authoritativeOwned) return;
|
||||
}
|
||||
|
||||
// FN-4811 follow-up (FN-4814/FN-4809/FN-4811 production failure): claim a
|
||||
|
||||
@@ -40,10 +40,14 @@ export {
|
||||
} from "./workflow-graph-branches.js";
|
||||
export {
|
||||
createDefaultNodeHandlers,
|
||||
createPrimitivePromptLikeHandler,
|
||||
createPrimitiveStepReviewHandler,
|
||||
createNoopLegacySeams,
|
||||
createParseStepsHandler,
|
||||
createCodeNodeHandler,
|
||||
PARSE_STEPS_DEFAULT_ARTIFACT,
|
||||
WORKFLOW_ID_CONTEXT_KEY,
|
||||
WORKFLOW_RUN_ID_CONTEXT_KEY,
|
||||
type WorkflowCustomNodeRunner,
|
||||
type WorkflowLegacySeams,
|
||||
type WorkflowSeamName,
|
||||
@@ -51,6 +55,28 @@ export {
|
||||
type CodeNodeRunner,
|
||||
type DefaultNodeHandlerDeps,
|
||||
} from "./workflow-node-handlers.js";
|
||||
export {
|
||||
markSideEffectsStarted,
|
||||
primitiveNodeContext,
|
||||
type RuntimePrimitiveName,
|
||||
type WorkflowRuntimeRunContext,
|
||||
type WorkflowRuntimeNodeContext,
|
||||
type WorkflowPrimitiveContext,
|
||||
type RuntimePrimitiveResult,
|
||||
type PreparedWorktree,
|
||||
type PlanningSessionResult,
|
||||
type CodingSessionResult,
|
||||
type ReviewPrimitiveResult,
|
||||
type VerificationPrimitiveResult,
|
||||
type WorkflowStepPrimitiveInput,
|
||||
type WorkflowStepPrimitiveResult,
|
||||
type TransitionPrimitiveInput,
|
||||
type MergePrimitiveInput,
|
||||
type MergePrimitiveResult,
|
||||
type AbortPrimitiveInput,
|
||||
type AuditPrimitiveInput,
|
||||
type WorkflowRuntimePrimitives,
|
||||
} from "./runtime-primitives.js";
|
||||
export {
|
||||
createPrNodeHandlers,
|
||||
createAutoMergeGateHandler,
|
||||
|
||||
@@ -331,6 +331,14 @@ export class ProjectEngine {
|
||||
else this.manualMergeResolvers.set(taskId, [r]);
|
||||
}
|
||||
|
||||
private removeMergeResolver(taskId: string, resolver: MergeResolver): void {
|
||||
const list = this.manualMergeResolvers.get(taskId);
|
||||
if (!list) return;
|
||||
const next = list.filter((candidate) => candidate !== resolver);
|
||||
if (next.length > 0) this.manualMergeResolvers.set(taskId, next);
|
||||
else this.manualMergeResolvers.delete(taskId);
|
||||
}
|
||||
|
||||
/** Remove and return all waiters for a task (empty array if none). */
|
||||
private takeMergeResolvers(taskId: string): MergeResolver[] {
|
||||
const list = this.manualMergeResolvers.get(taskId);
|
||||
@@ -426,7 +434,7 @@ export class ProjectEngine {
|
||||
// Workflow-graph interpreter merge seam: routes through the auto-merge
|
||||
// eligibility gate (requestInterpreterMerge), NOT the human "merge now"
|
||||
// bypass, so a graph merge node can't override an autoMerge-off project.
|
||||
this.runtime.setMergeRequester?.((taskId) => this.requestInterpreterMerge(taskId));
|
||||
this.runtime.setMergeRequester?.((taskId, options) => this.requestInterpreterMerge(taskId, options));
|
||||
}
|
||||
|
||||
getActiveMergeTaskId(): string | null {
|
||||
@@ -1072,21 +1080,56 @@ export class ProjectEngine {
|
||||
* Returns the full MergeResult so it can be used as the `onMerge` callback
|
||||
* in createServer().
|
||||
*/
|
||||
async onMerge(taskId: string): Promise<MergeResult> {
|
||||
// If this task is already queued or actively merging, wait for the
|
||||
// existing merge to finish rather than starting a second one.
|
||||
if (this.mergeActive.has(taskId)) {
|
||||
return new Promise<MergeResult>((resolve, reject) => {
|
||||
this.addMergeResolver(taskId, { resolve, reject });
|
||||
// Don't re-enqueue — the task is already in the queue/active
|
||||
});
|
||||
async onMerge(taskId: string, options: { signal?: AbortSignal } = {}): Promise<MergeResult> {
|
||||
const signal = options.signal;
|
||||
if (signal?.aborted) {
|
||||
throw new Error(`Merge request for ${taskId} aborted`);
|
||||
}
|
||||
|
||||
return new Promise<MergeResult>((resolve, reject) => {
|
||||
this.addMergeResolver(taskId, { resolve, reject });
|
||||
let settled = false;
|
||||
let abort: () => void = () => undefined;
|
||||
const cleanup = () => {
|
||||
signal?.removeEventListener("abort", abort);
|
||||
};
|
||||
const resolver: MergeResolver = {
|
||||
resolve: (result) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
resolve(result);
|
||||
},
|
||||
reject: (err) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
reject(err);
|
||||
},
|
||||
};
|
||||
abort = () => {
|
||||
this.removeMergeResolver(taskId, resolver);
|
||||
if (this.activeMergeTaskId === taskId) {
|
||||
this.mergeAbortController?.abort();
|
||||
this.mergeAbortController = null;
|
||||
this.activeMergeSession?.dispose();
|
||||
this.activeMergeSession = null;
|
||||
} else if (!this.hasMergeResolvers(taskId)) {
|
||||
this.mergeQueue = this.mergeQueue.filter((queuedTaskId) => queuedTaskId !== taskId);
|
||||
this.mergeActive.delete(taskId);
|
||||
}
|
||||
resolver.reject(new Error(`Merge request for ${taskId} aborted`));
|
||||
};
|
||||
|
||||
signal?.addEventListener("abort", abort, { once: true });
|
||||
this.addMergeResolver(taskId, resolver);
|
||||
|
||||
// If this task is already queued or actively merging, wait for the
|
||||
// existing merge to finish rather than starting a second one.
|
||||
if (this.mergeActive.has(taskId)) return;
|
||||
|
||||
if (!this.internalEnqueueMerge(taskId)) {
|
||||
// Drop just-added waiter(s) for this task and fail them.
|
||||
this.rejectMergeResolvers(taskId, new Error(`Merge enqueue rejected for ${taskId}`));
|
||||
this.removeMergeResolver(taskId, resolver);
|
||||
resolver.reject(new Error(`Merge enqueue rejected for ${taskId}`));
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -1099,7 +1142,7 @@ export class ProjectEngine {
|
||||
* it as "manual merge required" and parks the task in review — preserving the
|
||||
* contract that autoMerge-off leaves in-review terminal until a human merges.
|
||||
*/
|
||||
async requestInterpreterMerge(taskId: string): Promise<MergeResult> {
|
||||
async requestInterpreterMerge(taskId: string, options: { signal?: AbortSignal } = {}): Promise<MergeResult> {
|
||||
let task: Task | null = null;
|
||||
let settings: Settings | undefined;
|
||||
try {
|
||||
@@ -1135,7 +1178,7 @@ export class ProjectEngine {
|
||||
} as MergeResult;
|
||||
}
|
||||
// Eligible: route through the normal serialized merge path.
|
||||
return this.onMerge(taskId);
|
||||
return this.onMerge(taskId, options);
|
||||
}
|
||||
|
||||
private setRestoreDiagnostics(
|
||||
|
||||
238
packages/engine/src/runtime-primitives.ts
Normal file
238
packages/engine/src/runtime-primitives.ts
Normal file
@@ -0,0 +1,238 @@
|
||||
import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core";
|
||||
|
||||
import type { PrMergeCallResult } from "./pr-nodes.js";
|
||||
import type { RunTaskStepResult, ResetStepResult } from "./step-runner.js";
|
||||
import type { WorkflowNodeOutcome } from "./workflow-graph-executor.js";
|
||||
|
||||
export type RuntimePrimitiveName =
|
||||
| "prepare-worktree"
|
||||
| "read-artifact"
|
||||
| "write-artifact"
|
||||
| "planning-session"
|
||||
| "coding-session"
|
||||
| "step-session"
|
||||
| "reset-step"
|
||||
| "review"
|
||||
| "verification"
|
||||
| "workflow-step"
|
||||
| "transition"
|
||||
| "merge"
|
||||
| "abort"
|
||||
| "audit";
|
||||
|
||||
export interface WorkflowRuntimeRunContext {
|
||||
runId: string;
|
||||
taskId: string;
|
||||
workflowId: string;
|
||||
/** True after any primitive with task/git/session side effects starts. */
|
||||
sideEffectsStarted?: boolean;
|
||||
recoveryEventId?: string;
|
||||
}
|
||||
|
||||
export interface WorkflowRuntimeNodeContext {
|
||||
node: Pick<WorkflowIrNode, "id" | "kind" | "column" | "config">;
|
||||
effectivePrincipalId?: string;
|
||||
attempt?: number;
|
||||
context?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface WorkflowPrimitiveContext {
|
||||
run: WorkflowRuntimeRunContext;
|
||||
node: WorkflowRuntimeNodeContext;
|
||||
}
|
||||
|
||||
export interface RuntimePrimitiveResult<TValue = unknown> {
|
||||
outcome: WorkflowNodeOutcome;
|
||||
value?: string;
|
||||
data?: TValue;
|
||||
contextPatch?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface PreparedWorktree {
|
||||
worktreePath: string;
|
||||
branchName?: string;
|
||||
baseCommitSha?: string;
|
||||
modifiedFiles?: string[];
|
||||
}
|
||||
|
||||
export interface PlanningSessionResult {
|
||||
approved: boolean;
|
||||
artifactKeys: string[];
|
||||
createdTaskIds?: string[];
|
||||
feedback?: string;
|
||||
}
|
||||
|
||||
export interface CodingSessionResult {
|
||||
taskDone: boolean;
|
||||
modifiedFiles: string[];
|
||||
summary?: string;
|
||||
}
|
||||
|
||||
export interface ReviewPrimitiveResult {
|
||||
verdict: "APPROVE" | "REVISE" | "RETHINK" | "UNAVAILABLE";
|
||||
review?: string;
|
||||
summary?: string;
|
||||
}
|
||||
|
||||
export interface VerificationPrimitiveResult {
|
||||
verdict: "approve" | "revise" | "failed" | "advisory-failed" | "skipped";
|
||||
feedback?: string;
|
||||
stepName?: string;
|
||||
}
|
||||
|
||||
export interface WorkflowStepPrimitiveInput {
|
||||
phase: "pre-merge" | "post-merge";
|
||||
stepId?: string;
|
||||
worktreePath?: string;
|
||||
}
|
||||
|
||||
export interface WorkflowStepPrimitiveResult {
|
||||
allPassed: boolean;
|
||||
revisionRequested?: boolean;
|
||||
feedback?: string;
|
||||
stepName?: string;
|
||||
}
|
||||
|
||||
export interface TransitionPrimitiveInput {
|
||||
column?: string;
|
||||
status?: string | null;
|
||||
reason: string;
|
||||
preserveProgress?: boolean;
|
||||
}
|
||||
|
||||
export interface MergePrimitiveInput {
|
||||
expectedHeadOid?: string;
|
||||
manualAllowed?: boolean;
|
||||
}
|
||||
|
||||
export type MergePrimitiveResult =
|
||||
| { status: "merged"; noOp?: boolean }
|
||||
| { status: "manual-required"; reason?: string }
|
||||
| { status: "failed"; reason: string }
|
||||
| { status: "timeout" }
|
||||
| PrMergeCallResult;
|
||||
|
||||
export interface AbortPrimitiveInput {
|
||||
reason: string;
|
||||
hardCancel?: boolean;
|
||||
}
|
||||
|
||||
export interface AuditPrimitiveInput {
|
||||
type: string;
|
||||
message: string;
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface WorkflowRuntimePrimitives {
|
||||
prepareWorktree(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
): Promise<RuntimePrimitiveResult<PreparedWorktree>>;
|
||||
|
||||
readArtifact(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
key: string,
|
||||
): Promise<string | undefined>;
|
||||
|
||||
writeArtifact(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
key: string,
|
||||
content: string,
|
||||
): Promise<RuntimePrimitiveResult<{ key: string }>>;
|
||||
|
||||
runPlanningSession(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
): Promise<RuntimePrimitiveResult<PlanningSessionResult>>;
|
||||
|
||||
runCodingSession(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
prepared: PreparedWorktree,
|
||||
): Promise<RuntimePrimitiveResult<CodingSessionResult>>;
|
||||
|
||||
runTaskStep(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
stepIndex: number,
|
||||
): Promise<RunTaskStepResult>;
|
||||
|
||||
resetTaskStep(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
stepIndex: number,
|
||||
baselineSha?: string,
|
||||
checkpointId?: string,
|
||||
): Promise<ResetStepResult>;
|
||||
|
||||
runReview(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
input: { type: "plan" | "code"; stepIndex?: number; baselineSha?: string },
|
||||
): Promise<RuntimePrimitiveResult<ReviewPrimitiveResult>>;
|
||||
|
||||
runVerification(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
prepared: PreparedWorktree,
|
||||
): Promise<RuntimePrimitiveResult<VerificationPrimitiveResult>>;
|
||||
|
||||
runWorkflowStep(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
input: WorkflowStepPrimitiveInput,
|
||||
): Promise<RuntimePrimitiveResult<WorkflowStepPrimitiveResult>>;
|
||||
|
||||
updateSteps(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
steps: TaskStep[],
|
||||
): Promise<RuntimePrimitiveResult<{ count: number }>>;
|
||||
|
||||
transitionTask(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
input: TransitionPrimitiveInput,
|
||||
): Promise<RuntimePrimitiveResult>;
|
||||
|
||||
requestMerge(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
input?: MergePrimitiveInput,
|
||||
): Promise<RuntimePrimitiveResult<MergePrimitiveResult>>;
|
||||
|
||||
abortRun(
|
||||
ctx: WorkflowPrimitiveContext,
|
||||
task: TaskDetail,
|
||||
input: AbortPrimitiveInput,
|
||||
): Promise<RuntimePrimitiveResult>;
|
||||
|
||||
audit(ctx: WorkflowPrimitiveContext, input: AuditPrimitiveInput): Promise<void> | void;
|
||||
}
|
||||
|
||||
export function markSideEffectsStarted(ctx: WorkflowPrimitiveContext): WorkflowPrimitiveContext {
|
||||
return {
|
||||
...ctx,
|
||||
run: {
|
||||
...ctx.run,
|
||||
sideEffectsStarted: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function primitiveNodeContext(
|
||||
run: WorkflowRuntimeRunContext,
|
||||
node: WorkflowRuntimeNodeContext["node"],
|
||||
extras: Omit<WorkflowRuntimeNodeContext, "node"> = {},
|
||||
): WorkflowPrimitiveContext {
|
||||
return {
|
||||
run,
|
||||
node: {
|
||||
...extras,
|
||||
node,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -142,7 +142,10 @@ export class InProcessRuntime
|
||||
* before `start()` via `setMergeEnqueuer`.
|
||||
*/
|
||||
private mergeEnqueuer?: (taskId: string) => boolean;
|
||||
private mergeRequester?: (taskId: string) => Promise<import("@fusion/core").MergeResult>;
|
||||
private mergeRequester?: (
|
||||
taskId: string,
|
||||
options?: { signal?: AbortSignal },
|
||||
) => Promise<import("@fusion/core").MergeResult>;
|
||||
private clearMergeActive?: (taskId: string) => void;
|
||||
private activeMergeTaskIdProvider?: () => string | null;
|
||||
/** Tracks whether startup recovery was intentionally deferred due to pause state. */
|
||||
@@ -1140,7 +1143,12 @@ export class InProcessRuntime
|
||||
* forwards immediately when the executor already exists, and is re-applied at
|
||||
* executor construction during start().
|
||||
*/
|
||||
setMergeRequester(requestMerge: (taskId: string) => Promise<import("@fusion/core").MergeResult>): void {
|
||||
setMergeRequester(
|
||||
requestMerge: (
|
||||
taskId: string,
|
||||
options?: { signal?: AbortSignal },
|
||||
) => Promise<import("@fusion/core").MergeResult>,
|
||||
): void {
|
||||
this.mergeRequester = requestMerge;
|
||||
this.executor?.setMergeRequester(requestMerge);
|
||||
}
|
||||
|
||||
@@ -14,6 +14,8 @@ import type { TaskExecutor } from "./executor.js";
|
||||
import { executorLog } from "./logger.js";
|
||||
import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js";
|
||||
import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js";
|
||||
import type { StepReviewSeamResult, WorkflowLegacySeams } from "./workflow-node-handlers.js";
|
||||
import type { PreparedWorktree, WorkflowRuntimePrimitives } from "./runtime-primitives.js";
|
||||
|
||||
const AUTHORITATIVE_WORKFLOW_ID = "workflow-interpreter-authoritative";
|
||||
|
||||
@@ -26,7 +28,7 @@ export interface WorkflowAuthoritativeDriverStore {
|
||||
|
||||
export interface WorkflowAuthoritativeDriverDeps {
|
||||
store: WorkflowAuthoritativeDriverStore;
|
||||
executor: Pick<TaskExecutor, "createAuthoritativeWorkflowSeams">;
|
||||
executor: Pick<TaskExecutor, "createAuthoritativeWorkflowSeams"> & Partial<Pick<TaskExecutor, "createAuthoritativeWorkflowPrimitives">>;
|
||||
minimumObservedRuns?: number;
|
||||
}
|
||||
|
||||
@@ -49,6 +51,81 @@ function buildAuthoritativeSettings(settings: Settings): Settings {
|
||||
};
|
||||
}
|
||||
|
||||
function primitivesFromLegacySeams(seams: WorkflowLegacySeams): WorkflowRuntimePrimitives {
|
||||
const mapStepReviewValue = (verdict: StepReviewSeamResult["verdict"] | undefined): string => {
|
||||
switch (verdict) {
|
||||
case "APPROVE":
|
||||
return "approve";
|
||||
case "REVISE":
|
||||
return "revise";
|
||||
case "RETHINK":
|
||||
return "rethink";
|
||||
default:
|
||||
return "unavailable";
|
||||
}
|
||||
};
|
||||
|
||||
// Legacy seams do not consume PreparedWorktree; filesystem/session state is
|
||||
// still owned inside the seam implementation they delegate to.
|
||||
const prepared: PreparedWorktree = { worktreePath: "" };
|
||||
return {
|
||||
prepareWorktree: async () => ({ outcome: "success", data: prepared }),
|
||||
readArtifact: async () => undefined,
|
||||
writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }),
|
||||
runPlanningSession: async (ctx, task) => {
|
||||
const result = await seams.planning(task, ctx.node.context ?? {});
|
||||
return { ...result, data: { approved: result.outcome === "success", artifactKeys: [] } };
|
||||
},
|
||||
runCodingSession: async (ctx, task) => {
|
||||
const result = await seams.execute(task, ctx.node.context ?? {});
|
||||
return {
|
||||
...result,
|
||||
data: { taskDone: result.outcome === "success", modifiedFiles: [] },
|
||||
};
|
||||
},
|
||||
runTaskStep: async (ctx, task) => {
|
||||
const result = await seams.stepExecute?.(task, ctx.node.context ?? {});
|
||||
return { outcome: result?.outcome ?? "failure" };
|
||||
},
|
||||
resetTaskStep: async () => ({ ok: true }),
|
||||
runReview: async (ctx, task, input) => {
|
||||
if (typeof input.stepIndex === "number") {
|
||||
const result = await seams.stepReview?.(task, ctx.node.context ?? {}, { type: input.type });
|
||||
return {
|
||||
outcome: "success",
|
||||
value: mapStepReviewValue(result?.verdict),
|
||||
data: result ?? { verdict: "UNAVAILABLE" },
|
||||
};
|
||||
}
|
||||
const result = await seams.review(task, ctx.node.context ?? {});
|
||||
return { ...result, data: { verdict: result.outcome === "success" ? "APPROVE" : "REVISE" } };
|
||||
},
|
||||
runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }),
|
||||
runWorkflowStep: async (ctx, task) => {
|
||||
const result = await seams.workflowStep?.(task, ctx.node.context ?? {});
|
||||
return {
|
||||
outcome: result?.outcome ?? "success",
|
||||
value: result?.value ?? "workflow-step-skipped",
|
||||
contextPatch: result?.contextPatch,
|
||||
data: { allPassed: result?.outcome !== "failure" },
|
||||
};
|
||||
},
|
||||
updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }),
|
||||
transitionTask: async (ctx, task) => seams.schedule(task, ctx.node.context ?? {}),
|
||||
requestMerge: async (ctx, task) => {
|
||||
const result = await seams.merge(task, ctx.node.context ?? {});
|
||||
return {
|
||||
...result,
|
||||
data: result.outcome === "success"
|
||||
? { status: "merged" as const }
|
||||
: { status: "failed" as const, reason: result.value ?? "merge-failed" },
|
||||
};
|
||||
},
|
||||
abortRun: async () => ({ outcome: "success" }),
|
||||
audit: () => undefined,
|
||||
};
|
||||
}
|
||||
|
||||
export class WorkflowAuthoritativeDriver {
|
||||
public constructor(private readonly deps: WorkflowAuthoritativeDriverDeps) {}
|
||||
|
||||
@@ -114,6 +191,7 @@ export class WorkflowAuthoritativeDriver {
|
||||
readinessReasons: [],
|
||||
};
|
||||
}
|
||||
const seams = this.deps.executor.createAuthoritativeWorkflowSeams(settings);
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: AUTHORITATIVE_WORKFLOW_ID, stepIds: [] }),
|
||||
@@ -123,7 +201,8 @@ export class WorkflowAuthoritativeDriver {
|
||||
ir: BUILTIN_CODING_WORKFLOW_IR,
|
||||
} satisfies Pick<WorkflowDefinition, "id" | "name" | "ir"> as WorkflowDefinition),
|
||||
},
|
||||
seams: this.deps.executor.createAuthoritativeWorkflowSeams(settings),
|
||||
primitives: this.deps.executor.createAuthoritativeWorkflowPrimitives?.(settings) ?? primitivesFromLegacySeams(seams),
|
||||
seams,
|
||||
runCustomNode: async (node) => {
|
||||
throw new Error(`unexpected custom node in builtin authoritative workflow: ${node.id}`);
|
||||
},
|
||||
|
||||
@@ -5,12 +5,15 @@ import {
|
||||
createDefaultNodeHandlers,
|
||||
createNoopLegacySeams,
|
||||
SPLIT_ACTIVE_CONTEXT_KEY,
|
||||
WORKFLOW_ID_CONTEXT_KEY,
|
||||
WORKFLOW_RUN_ID_CONTEXT_KEY,
|
||||
type CodeNodeRunner,
|
||||
type ForeachActiveContext,
|
||||
type ParseStepsHandlerDeps,
|
||||
type WorkflowCustomNodeRunner,
|
||||
type WorkflowLegacySeams,
|
||||
} from "./workflow-node-handlers.js";
|
||||
import type { WorkflowRuntimePrimitives } from "./runtime-primitives.js";
|
||||
import type { PrNodeDeps } from "./pr-nodes.js";
|
||||
import {
|
||||
runSplitJoin,
|
||||
@@ -48,6 +51,9 @@ export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeEx
|
||||
|
||||
export interface WorkflowGraphExecutorDeps {
|
||||
handlers?: Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>>;
|
||||
/** Workflow-native runtime primitives. When present, default nodes call these
|
||||
* directly instead of legacy executor/reviewer/merge seams. */
|
||||
primitives?: WorkflowRuntimePrimitives;
|
||||
seams?: WorkflowLegacySeams;
|
||||
/** Executes custom (non-seam) prompt/script/gate nodes. */
|
||||
runCustomNode?: WorkflowCustomNodeRunner;
|
||||
@@ -150,6 +156,7 @@ export class WorkflowGraphExecutor {
|
||||
this.maxRetriesPerNode = Math.max(1, Math.floor(deps.maxRetriesPerNode ?? 2));
|
||||
this.handlers = {
|
||||
...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams(), deps.runCustomNode, {
|
||||
primitives: deps.primitives,
|
||||
parseSteps: deps.parseStepsDeps,
|
||||
runCode: deps.runCode,
|
||||
prNodes: deps.prNodes,
|
||||
@@ -181,10 +188,13 @@ export class WorkflowGraphExecutor {
|
||||
outgoingMap.set(edge.from, list);
|
||||
}
|
||||
|
||||
const context: Record<string, unknown> = {};
|
||||
const runId = this.deps.runId ?? `${task.id}:run`;
|
||||
const context: Record<string, unknown> = {
|
||||
[WORKFLOW_RUN_ID_CONTEXT_KEY]: runId,
|
||||
[WORKFLOW_ID_CONTEXT_KEY]: ir.name || "unknown",
|
||||
};
|
||||
const visitedNodeIds: string[] = [];
|
||||
const inStack = new Set<string>();
|
||||
const runId = this.deps.runId ?? `${task.id}:run`;
|
||||
|
||||
// Bounded-rework generalization (U6). A `kind: "rework"` edge is the only
|
||||
// legal cycle: it loops back to a "rework region head" (the edge's `to` node).
|
||||
@@ -420,7 +430,12 @@ export class WorkflowGraphExecutor {
|
||||
return sourceResult;
|
||||
}
|
||||
|
||||
const matching = edges.filter((edge) => this.shouldTraverseEdge(edge, sourceResult));
|
||||
const outcomeMatching = edges.filter((edge) =>
|
||||
edge.condition?.startsWith("outcome:") && this.shouldTraverseEdge(edge, sourceResult)
|
||||
);
|
||||
const matching = outcomeMatching.length > 0
|
||||
? outcomeMatching
|
||||
: edges.filter((edge) => this.shouldTraverseEdge(edge, sourceResult));
|
||||
if (matching.length === 0) {
|
||||
return sourceResult;
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import type {
|
||||
} from "./workflow-graph-branches.js";
|
||||
import type { ForeachEnvironment, WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js";
|
||||
import type { PrNodeDeps } from "./pr-nodes.js";
|
||||
import type { WorkflowPrimitiveContext, WorkflowRuntimePrimitives } from "./runtime-primitives.js";
|
||||
// (Both types are also used as values in the side-effect tracking wrappers below.)
|
||||
|
||||
/**
|
||||
@@ -46,6 +47,7 @@ export interface WorkflowGraphRunnerStore {
|
||||
export interface WorkflowGraphTaskRunnerDeps {
|
||||
store: WorkflowGraphRunnerStore;
|
||||
seams: WorkflowLegacySeams;
|
||||
primitives?: WorkflowRuntimePrimitives;
|
||||
runCustomNode: WorkflowCustomNodeRunner;
|
||||
maxRetriesPerNode?: number;
|
||||
/** Optional diagnostics hook (audit/log emission). Never throws into the run. */
|
||||
@@ -169,6 +171,11 @@ export class WorkflowGraphTaskRunner {
|
||||
const wrappedSeams: WorkflowLegacySeams = {
|
||||
planning: (t, c) => ((sideEffectsRan = true), invoked.push("planning"), seams.planning(t, c)),
|
||||
execute: (t, c) => ((sideEffectsRan = true), invoked.push("execute"), seams.execute(t, c)),
|
||||
workflowStep: (t, c) => {
|
||||
sideEffectsRan = true;
|
||||
invoked.push("workflow-step");
|
||||
return seams.workflowStep?.(t, c) ?? Promise.resolve({ outcome: "success", value: "workflow-step-skipped" });
|
||||
},
|
||||
review: (t, c) => ((sideEffectsRan = true), invoked.push("review"), seams.review(t, c)),
|
||||
merge: (t, c) => ((sideEffectsRan = true), invoked.push("merge"), seams.merge(t, c)),
|
||||
schedule: (t, c) => ((sideEffectsRan = true), invoked.push("schedule"), seams.schedule(t, c)),
|
||||
@@ -186,10 +193,25 @@ export class WorkflowGraphTaskRunner {
|
||||
invoked.push(node.id);
|
||||
return this.deps.runCustomNode(node, t, c);
|
||||
};
|
||||
const wrappedPrimitives = this.deps.primitives
|
||||
? new Proxy(this.deps.primitives, {
|
||||
get: (target, prop, receiver) => {
|
||||
const value = Reflect.get(target, prop, receiver);
|
||||
if (typeof value !== "function") return value;
|
||||
return (...args: unknown[]) => {
|
||||
sideEffectsRan = true;
|
||||
const ctx = args[0] as WorkflowPrimitiveContext | undefined;
|
||||
invoked.push(ctx?.node?.node?.id ?? String(prop));
|
||||
return value.apply(target, args);
|
||||
};
|
||||
},
|
||||
}) as WorkflowRuntimePrimitives
|
||||
: undefined;
|
||||
|
||||
try {
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
seams: wrappedSeams,
|
||||
primitives: wrappedPrimitives,
|
||||
runCustomNode: wrappedRunCustomNode,
|
||||
maxRetriesPerNode: this.deps.maxRetriesPerNode,
|
||||
branchPersistence: this.deps.branchPersistence,
|
||||
|
||||
@@ -3,8 +3,20 @@ 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";
|
||||
export type WorkflowSeamName =
|
||||
| "planning"
|
||||
| "execute"
|
||||
| "workflow-step"
|
||||
| "review"
|
||||
| "merge"
|
||||
| "schedule"
|
||||
| "step-execute";
|
||||
|
||||
export interface WorkflowLegacySeams {
|
||||
/** Planning/spec stage. Built-in triage runs upstream of the interpreter
|
||||
@@ -12,6 +24,7 @@ export interface WorkflowLegacySeams {
|
||||
* custom planning behavior is expressed as a custom prompt node. */
|
||||
planning: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
execute: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
workflowStep?: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
review: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
merge: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
schedule: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
|
||||
@@ -104,6 +117,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 +168,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;
|
||||
@@ -156,6 +203,7 @@ export function resolveSeamName(node: { config?: Record<string, unknown> }): Wor
|
||||
if (
|
||||
seam === "planning" ||
|
||||
seam === "execute" ||
|
||||
seam === "workflow-step" ||
|
||||
seam === "review" ||
|
||||
seam === "merge" ||
|
||||
seam === "schedule" ||
|
||||
@@ -211,6 +259,11 @@ export function createPromptLikeHandler(
|
||||
// IS the seam node, so its declared column drives the binding. (Other seams
|
||||
// — planning/review/merge/schedule — stamp it too; only execute reads it.)
|
||||
context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = node.id;
|
||||
if (seam === "workflow-step") {
|
||||
return seams.workflowStep
|
||||
? seams.workflowStep(context.task, context.context)
|
||||
: { outcome: "success", value: "workflow-step-skipped" };
|
||||
}
|
||||
return seams[seam]!(context.task, context.context);
|
||||
}
|
||||
if (!runCustomNode) {
|
||||
@@ -220,6 +273,106 @@ 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 === "success" ? "failure" : prepared.outcome,
|
||||
value: prepared.value ?? "prepare-worktree-failed",
|
||||
contextPatch: prepared.contextPatch,
|
||||
};
|
||||
}
|
||||
const result = await primitives.runCodingSession(primitiveCtx, context.task, prepared.data);
|
||||
const contextPatch = prepared.contextPatch || result.contextPatch
|
||||
? {
|
||||
...(prepared.contextPatch ?? {}),
|
||||
...(result.contextPatch ?? {}),
|
||||
}
|
||||
: undefined;
|
||||
return {
|
||||
outcome: result.outcome,
|
||||
value: result.value,
|
||||
contextPatch: {
|
||||
...(contextPatch ?? {}),
|
||||
"workflow:worktree-path": prepared.data.worktreePath,
|
||||
},
|
||||
};
|
||||
}
|
||||
if (seam === "workflow-step") {
|
||||
const worktreePath = typeof context.context["workflow:worktree-path"] === "string"
|
||||
? context.context["workflow:worktree-path"]
|
||||
: undefined;
|
||||
const result = await primitives.runWorkflowStep(primitiveCtx, context.task, {
|
||||
phase: "pre-merge",
|
||||
worktreePath,
|
||||
});
|
||||
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 +490,65 @@ 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",
|
||||
};
|
||||
let primitivePatch: Record<string, unknown> | undefined;
|
||||
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,
|
||||
},
|
||||
);
|
||||
if (primitiveResult.outcome !== "success") {
|
||||
return {
|
||||
outcome: primitiveResult.outcome,
|
||||
value: primitiveResult.value,
|
||||
contextPatch: primitiveResult.contextPatch,
|
||||
};
|
||||
}
|
||||
primitivePatch = primitiveResult.contextPatch;
|
||||
result = primitiveResult.data ?? { verdict: "UNAVAILABLE" as const };
|
||||
if (result.verdict !== "UNAVAILABLE") break;
|
||||
}
|
||||
|
||||
if (!advisory) {
|
||||
active.verdict = result.verdict;
|
||||
}
|
||||
const patch: Record<string, unknown> = {
|
||||
...(primitivePatch ?? {}),
|
||||
[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 +754,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 +780,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 +812,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,
|
||||
@@ -611,6 +829,7 @@ export function createNoopLegacySeams(): WorkflowLegacySeams {
|
||||
return {
|
||||
planning: success,
|
||||
execute: success,
|
||||
workflowStep: success,
|
||||
review: success,
|
||||
merge: success,
|
||||
schedule: success,
|
||||
|
||||
@@ -15,9 +15,10 @@ import {
|
||||
} from "./workflow-graph-executor.js";
|
||||
import {
|
||||
createDefaultNodeHandlers,
|
||||
createNoopLegacySeams,
|
||||
type WorkflowCustomNodeRunner,
|
||||
type WorkflowLegacySeams,
|
||||
} from "./workflow-node-handlers.js";
|
||||
import type { WorkflowRuntimePrimitives } from "./runtime-primitives.js";
|
||||
|
||||
export type WorkflowTaskRuntimeDisposition = "completed" | "failed";
|
||||
|
||||
@@ -31,7 +32,7 @@ export interface WorkflowTaskRuntimeResult {
|
||||
|
||||
export interface WorkflowTaskRuntimeDeps extends Omit<WorkflowGraphExecutorDeps, "seams" | "runCustomNode"> {
|
||||
store: WorkflowIrResolverStore;
|
||||
seams: WorkflowLegacySeams;
|
||||
primitives: WorkflowRuntimePrimitives;
|
||||
runCustomNode: WorkflowCustomNodeRunner;
|
||||
onEvent?: (event: { type: "start" | "terminal"; taskId: string; detail: string }) => void;
|
||||
}
|
||||
@@ -39,11 +40,11 @@ export interface WorkflowTaskRuntimeDeps extends Omit<WorkflowGraphExecutorDeps,
|
||||
/**
|
||||
* 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. This is intentionally
|
||||
* different from `WorkflowGraphTaskRunner`, whose current contract still models
|
||||
* "no selection" as legacy fallback.
|
||||
* It always resolves a task to a workflow IR: explicit selections resolve only
|
||||
* to their selected workflow, and tasks without a selection resolve to the
|
||||
* built-in coding workflow. 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) {}
|
||||
@@ -80,6 +81,7 @@ export class WorkflowTaskRuntime {
|
||||
const invoked: string[] = [];
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
...this.deps,
|
||||
primitives: this.deps.primitives,
|
||||
handlers: this.recordingHandlers(invoked),
|
||||
// WorkflowTaskRuntime is the execution engine, so internally the graph
|
||||
// executor is authoritative even before the old feature flag plumbing is
|
||||
@@ -116,31 +118,28 @@ export class WorkflowTaskRuntime {
|
||||
let workflowId: string | undefined;
|
||||
try {
|
||||
workflowId = this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId;
|
||||
} catch {
|
||||
return builtinCodingTarget();
|
||||
} catch (err) {
|
||||
throw new Error(`workflow-selection-failed: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
|
||||
if (!workflowId) return builtinCodingTarget();
|
||||
|
||||
if (isBuiltinWorkflowId(workflowId)) {
|
||||
const builtin = getBuiltinWorkflow(workflowId);
|
||||
if (!builtin) return builtinCodingTarget();
|
||||
if (!builtin) throw new Error(`workflow-missing: ${workflowId}`);
|
||||
const ir = typeof builtin.ir === "string" ? parseWorkflowIr(builtin.ir) : builtin.ir;
|
||||
return { workflowId, ir };
|
||||
}
|
||||
|
||||
try {
|
||||
const def = await this.deps.store.getWorkflowDefinition(workflowId);
|
||||
if (!def) return builtinCodingTarget();
|
||||
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
|
||||
return { workflowId, ir };
|
||||
} catch {
|
||||
return builtinCodingTarget();
|
||||
}
|
||||
const def = await this.deps.store.getWorkflowDefinition(workflowId);
|
||||
if (!def) throw new Error(`workflow-missing: ${workflowId}`);
|
||||
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
|
||||
return { workflowId, ir };
|
||||
}
|
||||
|
||||
private recordingHandlers(invoked: string[]): Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>> {
|
||||
const defaultHandlers = createDefaultNodeHandlers(this.deps.seams, this.deps.runCustomNode, {
|
||||
const defaultHandlers = createDefaultNodeHandlers(createNoopLegacySeams(), this.deps.runCustomNode, {
|
||||
primitives: this.deps.primitives,
|
||||
parseSteps: this.deps.parseStepsDeps,
|
||||
runCode: this.deps.runCode,
|
||||
prNodes: this.deps.prNodes,
|
||||
|
||||
Reference in New Issue
Block a user