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

FN-5767: wire workflow graph executor to legacy seams

Browse Source 
Complete Phase 3 interpreter parity by routing workflow graph execution through legacy seam handlers.

- Simplify workflow IR schema and built-in coding workflow to seam-based v1 nodes/edges.
- Add workflow node handler layer with default legacy seam adapters and dedicated handler/parity tests.
- Update engine exports/executor behavior for conditional traversal, retries, outcome context, and parity assertions.
- Refresh core workflow IR tests/coverage for the new shape (including seam-stage expectations without triage) and remove obsolete schema fixture test.

Files changed:
 .changeset/fn-5767-interpreter-parity.md           |   5 +
 .github/actions/setup-node-pnpm/action.yml         |  41 +---
 docs/workflow-steps.md                             |  17 ++
 .../__tests__/builtin-coding-workflow-ir.test.ts   |  25 +--
 packages/core/src/__tests__/workflow-ir.test.ts    |  70 ------
 packages/core/src/builtin-coding-workflow-ir.ts    |  71 ++----
 packages/core/src/index.ts                         |  31 +--
 packages/core/src/workflow-ir-types.ts             |  91 +-------
 packages/core/src/workflow-ir.ts                   | 246 ++-------------------
 .../workflow-graph-executor-handlers.test.ts       | 201 +++++++++++++++++
 .../workflow-graph-executor-parity.test.ts         | 126 ++++++++---
 .../src/__tests__/workflow-node-handlers.test.ts   |  50 +++++
 packages/engine/src/index.ts                       |  16 +-
 packages/engine/src/workflow-graph-executor.ts     | 205 ++++++++++++-----
 packages/engine/src/workflow-node-handlers.ts      |  56 +++++
 15 files changed, 661 insertions(+), 590 deletions(-)

Fusion-Task-Id: FN-5767

Fusion-Task-Lineage: 0c68586e-2709-4521-a2ce-938ba1006ae0
This commit is contained in:
gsxdsm
2026-05-31 05:56:20 -07:00
parent 1edbb54fce
commit ba81d1f0ac
15 changed files with 672 additions and 601 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-5767-interpreter-parity.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": minor
---
Add workflow graph interpreter node handlers and traversal semantics behind the default-off `workflowGraphExecutor` experimental flag. The interpreter now supports prompt/script/gate dispatch through legacy seam DI, edge-condition routing (`success`/`failure`/`outcome:<value>`), bounded retries, and parity-oriented tests for no-op flag behavior and lifecycle routing.

41
.github/actions/setup-node-pnpm/action.yml vendored
View File

@@ -30,37 +30,16 @@ runs:
cache: pnpm
registry-url: ${{ inputs.registry-url }}
# Cache node_modules only for the exact pnpm-lock.yaml hash and Node/OS/arch tuple.
# runner.arch is essential: pnpm only installs the current platform's optional
# native deps (e.g. @rollup/rollup-linux-arm64-gnu), so an x64 cache restored on
# an arm64 runner (same runner.os) would be missing native binaries and break builds.
# github.job scopes the key per job: jobs sharing an OS/arch (e.g. the windows-x64
# CLI build and the Windows desktop build) would otherwise race on the same key and
# one would fail the post-job cache save ("unable to reserve cache ... another job
# may be creating this cache"). Per-job keys trade a little reuse for reliability.
# Intentionally no restore-keys fallback: partial restores can create inconsistent trees.
- name: Cache node_modules
id: node-modules-cache
if: ${{ inputs.skip-install != 'true' }}
uses: actions/cache@v4
with:
path: |
node_modules
**/node_modules
!**/.cache
key: node-modules-${{ github.job }}-${{ runner.os }}-${{ runner.arch }}-node${{ inputs.node-version }}-${{ hashFiles('pnpm-lock.yaml') }}
# Do NOT cache the linked node_modules tree. pnpm builds node_modules from
# symlinks/junctions into the content-addressable store, and actions/cache does
# not preserve Windows junctions across its tar/restore — a restored tree has
# broken peer links (e.g. @vitejs/plugin-react can't resolve its `vite` peer →
# ERR_MODULE_NOT_FOUND), which silently broke the Windows release build. The
# pnpm *store* is already cached by actions/setup-node above (`cache: pnpm`), so
# `pnpm install --frozen-lockfile` is fast with a warm store and relinks
# correctly per-OS/arch every time (also fixing the prior arch-key and per-job
# cache-race issues this node_modules cache was patched for).
- name: Install dependencies
if: ${{ inputs.skip-install != 'true' && steps.node-modules-cache.outputs.cache-hit != 'true' }}
if: ${{ inputs.skip-install != 'true' }}
shell: bash
run: pnpm install ${{ inputs.install-args }}
- name: Verify restored node_modules cache integrity
if: ${{ inputs.skip-install != 'true' && steps.node-modules-cache.outputs.cache-hit == 'true' }}
shell: bash
run: |
test -f node_modules/.modules.yaml || {
echo "ERROR: node_modules cache hit but node_modules/.modules.yaml is missing"
exit 1
}
pnpm -v

17
docs/workflow-steps.md
View File

@@ -361,6 +361,23 @@ Prompt-mode workflow agents should emit a trailing JSON object:
- Backward compatibility remains for legacy prose-only responses via heuristic fallback (`REQUEST REVISION` and approval keywords).
- If neither structured JSON nor fallback prose can be interpreted, output is recorded as `malformed` (no inferable verdict) instead of hard-failing the task.
## Workflow Graph Executor (interpreter)
The experimental `workflowGraphExecutor` path remains **default OFF** and only runs when `settings.experimentalFeatures.workflowGraphExecutor = true`.
When enabled, interpreter nodes dispatch through DI-backed legacy seams:
- `prompt` / `script` nodes with `config.seam` dispatch to `execute`, `review`, `merge`, or `schedule`
- `gate` nodes evaluate context-key expectations and return success/failure outcomes
Traversal semantics:
- edge with no condition or `success` routes on success
- `failure` routes on failure
- `outcome:<value>` routes when the node result value matches exactly
- unsupported conditions throw `WorkflowIrError`
- per-node retries are bounded and deterministic
Parity coverage includes flag-OFF no-op behavior, lifecycle ordering parity vs legacy seams, merge/file-scope-like failure routing, and downstream halt behavior for hard-cancel/self-healing style failures.
## Workflow Step APIs
| Endpoint | Purpose |

25
packages/core/src/__tests__/builtin-coding-workflow-ir.test.ts
View File

@@ -1,18 +1,12 @@
import { describe, expect, it } from "vitest";
import {
BUILTIN_CODING_WORKFLOW_IR,
WORKFLOW_IR_SCHEMA_VERSION,
buildBuiltinCodingWorkflowIr,
parseWorkflowIr,
serializeWorkflowIr,
} from "../index.js";
import { BUILTIN_CODING_WORKFLOW_IR, parseWorkflowIr, serializeWorkflowIr } from "../index.js";
describe("builtin coding workflow ir", () => {
it("parses and round-trips", () => {
const parsed = parseWorkflowIr(BUILTIN_CODING_WORKFLOW_IR);
const reparsed = parseWorkflowIr(serializeWorkflowIr(parsed));
expect(reparsed).toEqual(parsed);
expect(parsed.schemaVersion).toBe(WORKFLOW_IR_SCHEMA_VERSION);
expect(parsed.version).toBe("v1");
});
it("contains exactly one start and one end node", () => {
@@ -21,14 +15,11 @@ describe("builtin coding workflow ir", () => {
expect(nodes.filter((node) => node.kind === "end")).toHaveLength(1);
});
it("exposes coding lifecycle stages", () => {
const stageNodes = BUILTIN_CODING_WORKFLOW_IR.nodes.filter((node) => node.config?.stage);
const stages = stageNodes.map((node) => String(node.config?.stage));
expect(stages).toEqual(expect.arrayContaining(["triage", "execute", "review", "merge"]));
});
it("builder returns parser-validated ir", () => {
const built = buildBuiltinCodingWorkflowIr();
expect(built.metadata.name).toContain("Coding Lifecycle");
it("exposes coding lifecycle seams", () => {
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).not.toContain("triage");
});
});

70
packages/core/src/__tests__/workflow-ir.test.ts
View File

@@ -1,70 +0,0 @@
import { describe, expect, it } from "vitest";
import {
BUILTIN_WORKFLOW_IR_FIXTURE,
WORKFLOW_IR_SCHEMA_VERSION,
parseWorkflowIr,
serializeWorkflowIr,
WorkflowIrError,
} from "../index.js";
describe("workflow ir", () => {
it("round-trips fixture with no data loss", () => {
const serialized = serializeWorkflowIr(BUILTIN_WORKFLOW_IR_FIXTURE);
const parsed = parseWorkflowIr(serialized);
expect(parsed).toEqual(BUILTIN_WORKFLOW_IR_FIXTURE);
});
it("rejects missing or mismatched schemaVersion", () => {
expect(() =>
parseWorkflowIr({
metadata: { name: "missing-version" },
nodes: [],
edges: [],
}),
).toThrowError(expect.objectContaining({ code: "unsupported_version" }));
expect(() =>
parseWorkflowIr({
schemaVersion: "2.0.0",
metadata: { name: "wrong-version" },
nodes: [],
edges: [],
}),
).toThrowError(expect.objectContaining({ code: "unsupported_version" }));
});
it("rejects unknown node kinds and dangling edges", () => {
expect(() =>
parseWorkflowIr({
schemaVersion: WORKFLOW_IR_SCHEMA_VERSION,
metadata: { name: "unknown-kind" },
nodes: [{ id: "n1", kind: "custom" }],
edges: [],
}),
).toThrowError(WorkflowIrError);
expect(() =>
parseWorkflowIr({
schemaVersion: WORKFLOW_IR_SCHEMA_VERSION,
metadata: { name: "dangling-edge" },
nodes: [{ id: "start", kind: "start" }],
edges: [{ id: "e1", from: "start", to: "missing" }],
}),
).toThrowError(expect.objectContaining({ code: "dangling_edge" }));
});
it("parses fixture JSON string for interpreter parity", () => {
const json = JSON.stringify(BUILTIN_WORKFLOW_IR_FIXTURE);
const parsed = parseWorkflowIr(json);
expect(parsed.schemaVersion).toBe(WORKFLOW_IR_SCHEMA_VERSION);
expect(parsed.nodes.length).toBeGreaterThanOrEqual(1);
expect(parsed.edges.length).toBeGreaterThanOrEqual(1);
});
it("exposes workflow ir surface from package entry", () => {
expect(typeof parseWorkflowIr).toBe("function");
expect(typeof serializeWorkflowIr).toBe("function");
expect(BUILTIN_WORKFLOW_IR_FIXTURE.schemaVersion).toBe(WORKFLOW_IR_SCHEMA_VERSION);
});
});

71
packages/core/src/builtin-coding-workflow-ir.ts
View File

@@ -1,60 +1,25 @@
import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js";
import { WORKFLOW_IR_SCHEMA_VERSION, type WorkflowIr } from "./workflow-ir-types.js";
import type { WorkflowIr } from "./workflow-ir-types.js";
import { parseWorkflowIr } from "./workflow-ir.js";
/**
* Built-in coding lifecycle workflow encoded in v1 Workflow IR.
*
* Mapping notes:
* - Legacy "agent-call" semantics are represented by `prompt` nodes with `config.agentRole`.
* - Typed edge semantics are represented via `edge.condition` tokens.
*/
export const BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = {
schemaVersion: WORKFLOW_IR_SCHEMA_VERSION,
metadata: {
name: "Built-in Coding Lifecycle Workflow",
description: "Legacy authoritative coding lifecycle encoded as v1 IR scaffold.",
createdAt: "2026-05-31T00:00:00.000Z",
templateId: "builtin-coding-lifecycle-v1",
},
const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = {
version: "v1",
name: "builtin-coding-workflow",
nodes: [
{ id: "node-start", kind: "start", label: "Start" },
{
id: "node-triage",
kind: "prompt",
label: "Triage",
config: { stage: "triage", agentRole: "triage", legacySeam: "triage" },
},
{
id: "node-execute",
kind: "prompt",
label: "Execute",
config: { stage: "execute", agentRole: "executor", legacySeam: "executor" },
},
{
id: "node-review",
kind: "gate",
label: "Review",
config: { stage: "review", gateMode: "approval", legacySeam: "reviewer" },
},
{
id: "node-merge",
kind: "script",
label: "Merge",
config: { stage: "merge", script: "legacy-merger", legacySeam: "merger" },
},
{ id: "node-end", kind: "end", label: "End" },
{ id: "start", kind: "start" },
{ id: "execute", kind: "prompt", config: { seam: "execute" } },
{ id: "review", kind: "prompt", config: { seam: "review" } },
{ id: "merge", kind: "prompt", config: { seam: "merge" } },
{ id: "end", kind: "end" },
],
edges: [
{ id: "edge-start-triage", from: "node-start", to: "node-triage" },
{ id: "edge-triage-execute", from: "node-triage", to: "node-execute", condition: "success" },
{ id: "edge-execute-review", from: "node-execute", to: "node-review", condition: "success" },
{ id: "edge-review-merge", from: "node-review", to: "node-merge", condition: "approved" },
{ id: "edge-review-execute", from: "node-review", to: "node-execute", condition: "revise" },
{ id: "edge-merge-end", from: "node-merge", to: "node-end", condition: "success" },
{ from: "start", to: "execute" },
{ from: "execute", to: "review", condition: "success" },
{ from: "review", to: "merge", condition: "success" },
{ from: "merge", to: "end", condition: "success" },
{ from: "execute", to: "end", condition: "failure" },
{ from: "review", to: "end", condition: "failure" },
{ from: "merge", to: "end", condition: "failure" },
],
};
/** Ensure built-in IR remains parser-valid and serializable. */
export function buildBuiltinCodingWorkflowIr(): WorkflowIr {
return parseWorkflowIr(serializeWorkflowIr(BUILTIN_CODING_WORKFLOW_IR));
}
export const BUILTIN_CODING_WORKFLOW_IR = parseWorkflowIr(RAW_BUILTIN_CODING_WORKFLOW_IR);

31
packages/core/src/index.ts
View File

@@ -1,25 +1,6 @@
export { COLUMNS, DEFAULT_COLUMN, isColumn, normalizeColumn, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, isMergeRequestContractShadowEnabled, resolvePersistAgentThinkingLog, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, PERMANENT_AGENT_ACTION_CATEGORIES, AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, AGENT_PROVISIONING_APPROVAL_MODES, SANDBOX_PROVISIONING_APPROVAL_MODES, AGENT_PERMISSION_POLICY_PRESET_IDS, LEGACY_AGENT_PERMISSION_POLICY_ACTION_CATEGORY_ALIASES, APPROVAL_REQUEST_STATUSES, APPROVAL_REQUEST_AUDIT_EVENT_TYPES, normalizeApprovalRequestActionCategory, isValidApprovalRequestTransition, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, HIGH_FANOUT_BLOCKER_TODO_THRESHOLD, STALE_HIGH_FANOUT_BLOCKER_AGE_THRESHOLD_MS, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeIntegrationWorktreeMode, normalizeMergeAdvanceAutoSyncMode, MERGE_ADVANCE_AUTO_SYNC_MODES, normalizeMergeConflictStrategy, normalizeMergeStrategyOverlapBehavior, normalizePostMergeAuditMode, POST_MERGE_AUDIT_MODES, normalizeMergeAuditAutoRecovery, MERGE_AUDIT_AUTO_RECOVERY_MODES, normalizeMergerMode, MERGER_MODES, normalizeAutoRecovery, AUTO_RECOVERY_MODES, buildResearchDocumentKey, REPO_OVERRIDE_RE, SHARED_STATE_SNAPSHOT_VERSION } from "./types.js";
export type { Column, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeIntegrationWorktreeMode, MergeAdvanceAutoSyncMode, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, MergerMode, MergerSettings, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, ExecutionMode, TaskPriority, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, MergeRequestState, MergeRequestRecord, CompletionHandoffMarker, HandoffEvidence, HandoffToReviewOptions, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure } from "./types.js";
export { AGENT_VALID_TRANSITIONS, DUPLICATE_OF_METADATA_KEY } from "./types.js";
export { WORKFLOW_IR_SCHEMA_VERSION, WORKFLOW_IR_NODE_KINDS } from "./workflow-ir-types.js";
export type {
WorkflowIr,
WorkflowIrNode,
WorkflowIrEdge,
WorkflowIrMetadata,
WorkflowIrNodeKind,
WorkflowIrSchemaVersion,
} from "./workflow-ir-types.js";
export {
parseWorkflowIr,
serializeWorkflowIr,
WorkflowIrError,
BUILTIN_WORKFLOW_IR_FIXTURE,
} from "./workflow-ir.js";
export {
BUILTIN_CODING_WORKFLOW_IR,
buildBuiltinCodingWorkflowIr,
} from "./builtin-coding-workflow-ir.js";
export { customProviderRegistryKey } from "./custom-provider-key.js";
export { MOCK_PROVIDER_ID } from "./mock-provider-constants.js";
export type { MockProviderId, MockSessionPurpose } from "./mock-provider-constants.js";
@@ -50,6 +31,18 @@ export {
getAvailableTemplates,
getTemplatesForRole,
} from "./agent-prompts.js";
export {
parseWorkflowIr,
serializeWorkflowIr,
WorkflowIrError,
} from "./workflow-ir.js";
export type {
WorkflowIr,
WorkflowIrNode,
WorkflowIrEdge,
WorkflowIrNodeKind,
} from "./workflow-ir-types.js";
export { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js";
// ── Engine wiring (set by @fusion/engine at module load) ────────────
export {

91
packages/core/src/workflow-ir-types.ts
View File

@@ -1,103 +1,20 @@
/**
* Workflow IR schema version supported by this runtime.
*/
export const WORKFLOW_IR_SCHEMA_VERSION = "1.0.0" as const;
export type WorkflowIrNodeKind = "start" | "prompt" | "script" | "gate" | "end";
/**
* Supported workflow IR schema version literal.
*/
export type WorkflowIrSchemaVersion = typeof WORKFLOW_IR_SCHEMA_VERSION;
/**
* Built-in workflow IR node kinds available in v1.
*/
export const WORKFLOW_IR_NODE_KINDS = [
"start",
"prompt",
"script",
"gate",
"end",
] as const;
/**
* Union of allowed built-in workflow IR node kinds.
*/
export type WorkflowIrNodeKind = typeof WORKFLOW_IR_NODE_KINDS[number];
/**
* JSON-serializable primitive value.
*/
export type JsonPrimitive = string | number | boolean | null;
/**
* JSON-serializable value.
*/
export type JsonValue = JsonPrimitive | JsonObject | JsonArray;
/**
* JSON object map for serializable structures.
*/
export interface JsonObject {
[key: string]: JsonValue;
}
/**
* JSON array for serializable structures.
*/
export type JsonArray = JsonValue[];
/**
* Workflow IR node.
*/
export interface WorkflowIrNode {
/** Stable node identifier unique within a workflow document. */
id: string;
/** Built-in workflow node kind. */
kind: WorkflowIrNodeKind;
/** Optional human-readable label for editor/interpreter diagnostics. */
label?: string;
/** JSON-safe node configuration payload. */
config?: Record<string, JsonValue>;
config?: Record<string, unknown>;
}
/**
* Workflow IR edge connecting two nodes.
*/
export interface WorkflowIrEdge {
/** Stable edge identifier unique within a workflow document. */
id: string;
/** Source node id. */
from: string;
/** Target node id. */
to: string;
/** Optional condition expression controlling edge traversal. */
condition?: string;
}
/**
* Workflow IR metadata.
*/
export interface WorkflowIrMetadata {
/** Workflow name. */
name: string;
/** Optional workflow description. */
description?: string;
/** Optional ISO-8601 creation timestamp string. */
createdAt?: string;
/** Additional JSON-safe metadata fields. */
[k: string]: JsonValue | undefined;
}
/**
* Versioned, serializable workflow intermediate representation.
*/
export interface WorkflowIr {
/** Schema version tag for compatibility checks. */
schemaVersion: WorkflowIrSchemaVersion;
/** Workflow metadata payload. */
metadata: WorkflowIrMetadata;
/** Workflow node list. */
version: "v1";
name: string;
nodes: WorkflowIrNode[];
/** Workflow edge list. */
edges: WorkflowIrEdge[];
}

252
packages/core/src/workflow-ir.ts
View File

@@ -1,244 +1,30 @@
import {
WORKFLOW_IR_NODE_KINDS,
WORKFLOW_IR_SCHEMA_VERSION,
type JsonValue,
type WorkflowIr,
type WorkflowIrEdge,
type WorkflowIrMetadata,
type WorkflowIrNode,
type WorkflowIrNodeKind,
} from "./workflow-ir-types.js";
import type { WorkflowIr } from "./workflow-ir-types.js";
const WORKFLOW_IR_NODE_KIND_SET = new Set<WorkflowIrNodeKind>(WORKFLOW_IR_NODE_KINDS);
type WorkflowIrErrorCode = "unsupported_version" | "invalid_shape" | "dangling_edge";
/**
* Error thrown when workflow IR parsing or validation fails.
*/
export class WorkflowIrError extends Error {
constructor(
message: string,
readonly code: WorkflowIrErrorCode,
) {
constructor(message: string) {
super(message);
this.name = "WorkflowIrError";
}
}
function isObjectRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
export function parseWorkflowIr(input: string | WorkflowIr): WorkflowIr {
const value: unknown = typeof input === "string" ? JSON.parse(input) : input;
if (!value || typeof value !== "object") {
throw new WorkflowIrError("Workflow IR must be an object");
}
const ir = value as WorkflowIr;
if (ir.version !== "v1") throw new WorkflowIrError("Workflow IR version must be v1");
if (!Array.isArray(ir.nodes) || !Array.isArray(ir.edges)) {
throw new WorkflowIrError("Workflow IR nodes/edges must be arrays");
}
const startCount = ir.nodes.filter((n) => n.kind === "start").length;
const endCount = ir.nodes.filter((n) => n.kind === "end").length;
if (startCount !== 1 || endCount !== 1) {
throw new WorkflowIrError("Workflow IR must contain exactly one start and one end node");
}
return ir;
}
function isJsonValue(value: unknown): value is JsonValue {
if (value === null) return true;
const valueType = typeof value;
if (valueType === "string" || valueType === "number" || valueType === "boolean") {
return true;
}
if (Array.isArray(value)) {
return value.every((entry) => isJsonValue(entry));
}
if (!isObjectRecord(value)) {
return false;
}
return Object.values(value).every((entry) => isJsonValue(entry));
}
function parseNode(input: unknown, index: number): WorkflowIrNode {
if (!isObjectRecord(input)) {
throw new WorkflowIrError(`Node at index ${index} must be an object.`, "invalid_shape");
}
const { id, kind, label, config } = input;
if (typeof id !== "string" || id.length === 0) {
throw new WorkflowIrError(`Node at index ${index} has an invalid id.`, "invalid_shape");
}
if (typeof kind !== "string" || !WORKFLOW_IR_NODE_KIND_SET.has(kind as WorkflowIrNodeKind)) {
throw new WorkflowIrError(`Node ${id} has unknown kind: ${String(kind)}.`, "invalid_shape");
}
const parsedKind = kind as WorkflowIrNodeKind;
if (label !== undefined && typeof label !== "string") {
throw new WorkflowIrError(`Node ${id} label must be a string when provided.`, "invalid_shape");
}
if (config !== undefined) {
if (!isObjectRecord(config) || !isJsonValue(config)) {
throw new WorkflowIrError(`Node ${id} config must be a JSON-serializable object.`, "invalid_shape");
}
}
return {
id,
kind: parsedKind,
...(label !== undefined ? { label } : {}),
...(config !== undefined ? { config: config as Record<string, JsonValue> } : {}),
};
}
function parseEdge(input: unknown, index: number): WorkflowIrEdge {
if (!isObjectRecord(input)) {
throw new WorkflowIrError(`Edge at index ${index} must be an object.`, "invalid_shape");
}
const { id, from, to, condition } = input;
if (typeof id !== "string" || id.length === 0) {
throw new WorkflowIrError(`Edge at index ${index} has an invalid id.`, "invalid_shape");
}
if (typeof from !== "string" || from.length === 0 || typeof to !== "string" || to.length === 0) {
throw new WorkflowIrError(`Edge ${id} must include non-empty from/to node ids.`, "invalid_shape");
}
if (condition !== undefined && typeof condition !== "string") {
throw new WorkflowIrError(`Edge ${id} condition must be a string when provided.`, "invalid_shape");
}
return {
id,
from,
to,
...(condition !== undefined ? { condition } : {}),
};
}
function parseMetadata(input: unknown): WorkflowIrMetadata {
if (!isObjectRecord(input)) {
throw new WorkflowIrError("Workflow metadata must be an object.", "invalid_shape");
}
if (typeof input.name !== "string" || input.name.length === 0) {
throw new WorkflowIrError("Workflow metadata.name must be a non-empty string.", "invalid_shape");
}
if (input.description !== undefined && typeof input.description !== "string") {
throw new WorkflowIrError("Workflow metadata.description must be a string when provided.", "invalid_shape");
}
if (input.createdAt !== undefined && typeof input.createdAt !== "string") {
throw new WorkflowIrError("Workflow metadata.createdAt must be a string when provided.", "invalid_shape");
}
const metadata: WorkflowIrMetadata = {
name: input.name,
};
for (const [key, value] of Object.entries(input)) {
if (value === undefined) continue;
if (!isJsonValue(value)) {
throw new WorkflowIrError(`Workflow metadata.${key} must be JSON-serializable.`, "invalid_shape");
}
metadata[key] = value;
}
return metadata;
}
/**
* Parse and validate workflow IR from a JSON string or object.
*/
export function parseWorkflowIr(input: unknown): WorkflowIr {
const parsedInput =
typeof input === "string"
? (() => {
try {
return JSON.parse(input) as unknown;
} catch {
throw new WorkflowIrError("Workflow IR JSON could not be parsed.", "invalid_shape");
}
})()
: input;
if (!isObjectRecord(parsedInput)) {
throw new WorkflowIrError("Workflow IR must be an object.", "invalid_shape");
}
if (parsedInput.schemaVersion !== WORKFLOW_IR_SCHEMA_VERSION) {
throw new WorkflowIrError(
`Unsupported workflow IR schemaVersion: ${String(parsedInput.schemaVersion)}.`,
"unsupported_version",
);
}
if (!Array.isArray(parsedInput.nodes) || !Array.isArray(parsedInput.edges)) {
throw new WorkflowIrError("Workflow IR nodes and edges must be arrays.", "invalid_shape");
}
const metadata = parseMetadata(parsedInput.metadata);
const nodes = parsedInput.nodes.map((node, index) => parseNode(node, index));
const edges = parsedInput.edges.map((edge, index) => parseEdge(edge, index));
const nodeIds = new Set(nodes.map((node) => node.id));
for (const edge of edges) {
if (!nodeIds.has(edge.from) || !nodeIds.has(edge.to)) {
throw new WorkflowIrError(
`Edge ${edge.id} references missing node ids: ${edge.from} -> ${edge.to}.`,
"dangling_edge",
);
}
}
return {
schemaVersion: WORKFLOW_IR_SCHEMA_VERSION,
metadata,
nodes,
edges,
};
}
/**
* Serialize workflow IR as JSON.
*/
export function serializeWorkflowIr(ir: WorkflowIr): string {
return JSON.stringify(ir);
return JSON.stringify(ir, null, 2);
}
/**
* Canonical built-in workflow IR fixture for v1 interpreter parity tests.
*/
export const BUILTIN_WORKFLOW_IR_FIXTURE: WorkflowIr = {
schemaVersion: WORKFLOW_IR_SCHEMA_VERSION,
metadata: {
name: "Documentation Review Workflow",
description: "Built-in documentation review path using prompt and gate nodes.",
createdAt: "2026-05-30T00:00:00.000Z",
templateId: "documentation-review",
},
nodes: [
{
id: "node-start",
kind: "start",
label: "Start",
},
{
id: "node-prompt-review",
kind: "prompt",
label: "Run documentation review prompt",
config: {
promptTemplate: "documentation-review",
reviewLevel: 1,
},
},
{
id: "node-gate-approval",
kind: "gate",
label: "Review approved?",
config: {
mode: "approval",
},
},
{
id: "node-end",
kind: "end",
label: "Finish",
},
],
edges: [
{
id: "edge-start-to-prompt",
from: "node-start",
to: "node-prompt-review",
},
{
id: "edge-prompt-to-gate",
from: "node-prompt-review",
to: "node-gate-approval",
},
{
id: "edge-gate-to-end",
from: "node-gate-approval",
to: "node-end",
condition: "approved",
},
],
};

201
packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts Normal file
View File

@@ -0,0 +1,201 @@
import { describe, expect, it, vi } from "vitest";
import { BUILTIN_CODING_WORKFLOW_IR } from "@fusion/core";
import type { TaskDetail, WorkflowIr } from "@fusion/core";
import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
const task = { id: "FN-5767" } as TaskDetail;
function settingsOn() {
return { experimentalFeatures: { workflowGraphExecutor: true } };
}
describe("WorkflowGraphExecutor traversal", () => {
it("walks linear graph", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "linear",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "end", condition: "success" },
],
};
const handler = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({ handlers: { prompt: handler } });
const result = await executor.run(task, settingsOn(), ir);
expect(result.outcome).toBe("success");
expect(handler).toHaveBeenCalledTimes(1);
});
it("routes failure edges", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "failure-route",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "b", kind: "script" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "b", condition: "failure" },
{ from: "b", to: "end", condition: "success" },
],
};
const executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({ outcome: "failure" }),
script: async () => ({ outcome: "success" }),
},
});
const result = await executor.run(task, settingsOn(), ir);
expect(result.visitedNodeIds).toContain("b");
});
it("supports outcome:value conditions", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "outcome-value",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "left", kind: "script" },
{ id: "right", kind: "script" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "left", condition: "outcome:left" },
{ from: "a", to: "right", condition: "outcome:right" },
{ from: "left", to: "end" },
{ from: "right", to: "end" },
],
};
const script = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({ outcome: "success", value: "right" }),
script,
},
});
const result = await executor.run(task, settingsOn(), ir);
expect(result.visitedNodeIds).toContain("right");
expect(result.visitedNodeIds).not.toContain("left");
});
it("leaves outcome unchanged when outcome:value does not match any edge", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "outcome-miss",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "left", kind: "script" },
{ id: "right", kind: "script" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "left", condition: "outcome:left" },
{ from: "a", to: "right", condition: "outcome:right" },
],
};
const executor = new WorkflowGraphExecutor({ handlers: { prompt: async () => ({ outcome: "success", value: "miss" }) } });
const result = await executor.run(task, settingsOn(), ir);
expect(result.outcome).toBe("success");
expect(result.visitedNodeIds).not.toContain("left");
expect(result.visitedNodeIds).not.toContain("right");
});
it("caps retries and converts exceptions to failure", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "retry",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "end", condition: "failure" },
],
};
const handler = vi.fn(async () => {
throw new Error("boom");
});
const executor = new WorkflowGraphExecutor({ handlers: { prompt: handler }, maxRetriesPerNode: 3 });
const result = await executor.run(task, settingsOn(), ir);
expect(handler).toHaveBeenCalledTimes(3);
expect(result.outcome).toBe("failure");
});
it("fan-out executes deterministic sorted order", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "fanout",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "b", kind: "script" },
{ id: "c", kind: "script" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "c" },
{ from: "a", to: "b" },
{ from: "b", to: "end" },
{ from: "c", to: "end" },
],
};
const order: string[] = [];
const executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({ outcome: "success" }),
script: async (node) => {
order.push(node.id);
return { outcome: "success" };
},
},
});
await executor.run(task, settingsOn(), ir);
expect(order).toEqual(["b", "c"]);
});
it("builtin coding workflow ir exposes expected lifecycle nodes", () => {
expect(BUILTIN_CODING_WORKFLOW_IR.nodes.map((node) => node.id)).toEqual(
expect.arrayContaining(["start", "execute", "review", "merge", "end"]),
);
});
it("rejects malformed cyclic graphs", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "cycle",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "a" },
],
};
const executor = new WorkflowGraphExecutor({ handlers: { prompt: async () => ({ outcome: "success" }) } });
await expect(executor.run(task, settingsOn(), ir)).rejects.toThrow("Cycle detected");
});
});

120
packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts
View File

@@ -1,32 +1,108 @@
import { describe, expect, it, vi } from "vitest";
import { BUILTIN_CODING_WORKFLOW_IR, parseWorkflowIr } from "@fusion/core";
import { WorkflowGraphExecutor, WORKFLOW_GRAPH_EXECUTOR_FLAG } from "../workflow-graph-executor.js";
import type { TaskDetail } from "@fusion/core";
describe("workflow graph executor parity scaffold", () => {
it("is strict no-op when flag is absent or false", async () => {
const onNode = vi.fn();
const executor = new WorkflowGraphExecutor({ onNode });
import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
import type { WorkflowLegacySeams } from "../workflow-node-handlers.js";
const absent = await executor.run({ workflow: BUILTIN_CODING_WORKFLOW_IR, settings: undefined });
expect(absent).toEqual({ executed: false, visitedNodeIds: [], reason: "flag-disabled" });
const task = { id: "FN-5767" } as TaskDetail;
const disabled = await executor.run({
workflow: BUILTIN_CODING_WORKFLOW_IR,
settings: { experimentalFeatures: { [WORKFLOW_GRAPH_EXECUTOR_FLAG]: false } },
});
expect(disabled).toEqual({ executed: false, visitedNodeIds: [], reason: "flag-disabled" });
expect(onNode).not.toHaveBeenCalled();
function runLegacy(seams: WorkflowLegacySeams) {
return async () => {
const events: string[] = [];
const execute = await seams.execute(task, {});
events.push(`execute:${execute.outcome}`);
if (execute.outcome !== "success") return events;
const review = await seams.review(task, {});
events.push(`review:${review.outcome}`);
if (review.outcome !== "success") return events;
const merge = await seams.merge(task, {});
events.push(`merge:${merge.outcome}`);
return events;
};
}
describe("WorkflowGraphExecutor interpreter-parity", () => {
it("is a strict no-op when workflowGraphExecutor flag is disabled", async () => {
const prompt = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({ handlers: { prompt, script: prompt, gate: prompt } });
const result = await executor.run(task, { experimentalFeatures: {} });
expect(result.executed).toBe(false);
expect(prompt).not.toHaveBeenCalled();
});
it("loads builtin coding workflow IR", () => {
const parsed = parseWorkflowIr(BUILTIN_CODING_WORKFLOW_IR);
const stages = parsed.nodes.map((node) => String(node.config?.stage ?? ""));
it("matches legacy execute-review-merge success path", async () => {
const events: string[] = [];
const seams: WorkflowLegacySeams = {
execute: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "success" }),
schedule: async () => ({ outcome: "success" }),
};
const legacyEvents = await runLegacy(seams)();
const executor = new WorkflowGraphExecutor({ seams, handlers: { prompt: async (node, ctx) => {
const seam = String(node.config?.seam);
const result = await seams[seam as keyof WorkflowLegacySeams](ctx.task, ctx.context);
events.push(`${seam}:${result.outcome}`);
return result;
} } });
expect(stages).toEqual(expect.arrayContaining(["triage", "execute", "review", "merge"]));
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
expect(result.outcome).toBe("success");
expect(events).toEqual(legacyEvents);
});
it.todo("parity invariant: file-scope violations match legacy FileScopeViolationError behavior");
it.todo("parity invariant: squash/merge contract outcomes match legacy merger");
it.todo("parity invariant: autoMerge=false keeps in-review terminal until human merge");
it.todo("parity invariant: moveTask in-progress->todo hard-cancels active execution");
it("routes file-scope-like merge failure parity", async () => {
const seams: WorkflowLegacySeams = {
execute: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "failure", value: "FileScopeViolationError" }),
schedule: async () => ({ outcome: "success" }),
};
const legacyEvents = await runLegacy(seams)();
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"]);
});
it("preserves autoMerge:false terminal in-review semantics via review failure", async () => {
const seams: WorkflowLegacySeams = {
execute: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "failure", value: "manual-merge-required" }),
merge: async () => ({ outcome: "success" }),
schedule: async () => ({ outcome: "success" }),
};
const executor = new WorkflowGraphExecutor({ seams });
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
expect(result.outcome).toBe("failure");
expect(result.visitedNodeIds).not.toContain("merge");
});
it("matches self-healing parity by routing deterministic failure outcomes", async () => {
const seams: WorkflowLegacySeams = {
execute: async () => ({ outcome: "failure", value: "recoverable" }),
review: vi.fn(async () => ({ outcome: "success" as const })),
merge: vi.fn(async () => ({ outcome: "success" as const })),
schedule: async () => ({ outcome: "success" }),
};
const executor = new WorkflowGraphExecutor({ seams });
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
expect(result.outcome).toBe("failure");
expect(result.context["node:execute:value"]).toBe("recoverable");
expect(seams.review).not.toHaveBeenCalled();
});
it("matches moveTask hard-cancel behavior by halting downstream seams", async () => {
const seams: WorkflowLegacySeams = {
execute: async () => ({ outcome: "failure", value: "hard-cancel" }),
review: vi.fn(async () => ({ outcome: "success" as const })),
merge: vi.fn(async () => ({ outcome: "success" as const })),
schedule: async () => ({ outcome: "success" }),
};
const executor = new WorkflowGraphExecutor({ seams });
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
expect(result.outcome).toBe("failure");
expect(seams.review).not.toHaveBeenCalled();
expect(seams.merge).not.toHaveBeenCalled();
});
});

50
packages/engine/src/__tests__/workflow-node-handlers.test.ts Normal file
View File

@@ -0,0 +1,50 @@
import { describe, expect, it, vi } from "vitest";
import type { TaskDetail, WorkflowIrNode } from "@fusion/core";
import { createDefaultNodeHandlers } from "../workflow-node-handlers.js";
const task = { id: "FN-5767" } as TaskDetail;
const node = (kind: WorkflowIrNode["kind"], seam?: string): WorkflowIrNode => ({ id: kind, kind, config: seam ? { seam } : {} });
describe("workflow node handlers", () => {
it("dispatches prompt node to matching seam", async () => {
const seams = {
execute: vi.fn(async () => ({ outcome: "success" as const })),
review: vi.fn(async () => ({ outcome: "success" as const })),
merge: vi.fn(async () => ({ outcome: "success" as const })),
schedule: vi.fn(async () => ({ outcome: "success" as const })),
};
const handlers = createDefaultNodeHandlers(seams);
await handlers.prompt(node("prompt", "review"), { task, settings: undefined, context: {} });
expect(seams.review).toHaveBeenCalledOnce();
expect(seams.execute).not.toHaveBeenCalled();
});
it("dispatches script node to matching seam", async () => {
const seams = {
execute: vi.fn(async () => ({ outcome: "success" as const })),
review: vi.fn(async () => ({ outcome: "success" as const })),
merge: vi.fn(async () => ({ outcome: "success" as const })),
schedule: vi.fn(async () => ({ outcome: "success" as const })),
};
const handlers = createDefaultNodeHandlers(seams);
await handlers.script(node("script", "execute"), { task, settings: undefined, context: {} });
expect(seams.execute).toHaveBeenCalledOnce();
});
it("gate returns failure when expected context value does not match", async () => {
const handlers = createDefaultNodeHandlers({
execute: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "success" }),
schedule: async () => ({ outcome: "success" }),
});
const result = await handlers.gate(
{ id: "g", kind: "gate", config: { contextKey: "phase", expect: "merge" } },
{ task, settings: undefined, context: { phase: "review" } },
);
expect(result).toEqual({ outcome: "failure", value: "gate-mismatch" });
});
});

16
packages/engine/src/index.ts
View File

@@ -16,15 +16,19 @@ export {
export { AgentSemaphore, PRIORITY_MERGE, PRIORITY_EXECUTE, PRIORITY_SPECIFY } from "./concurrency.js";
export { TriageProcessor, type TriageProcessorOptions } from "./triage.js";
export { TaskExecutor, type TaskExecutorOptions } from "./executor.js";
export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
export { Scheduler, type SchedulerOptions } from "./scheduler.js";
export {
WorkflowGraphExecutor,
WORKFLOW_GRAPH_EXECUTOR_FLAG,
type WorkflowGraphExecutorDependencies,
type WorkflowGraphExecutorRunInput,
type WorkflowGraphExecutorRunResult,
type WorkflowGraphExecutorDeps,
type WorkflowGraphExecutorResult,
} from "./workflow-graph-executor.js";
export {
createDefaultNodeHandlers,
createNoopLegacySeams,
type WorkflowLegacySeams,
type WorkflowSeamName,
} from "./workflow-node-handlers.js";
export { collectTaskEvaluationEvidence } from "./evaluator-evidence.js";
export { Scheduler, type SchedulerOptions } from "./scheduler.js";
export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js";
export { MissionAutopilot, type MissionAutopilotOptions } from "./mission-autopilot.js";
export { MissionExecutionLoop, type MissionExecutionLoopOptions, type ValidationResult, loopLog } from "./mission-execution-loop.js";

227
packages/engine/src/workflow-graph-executor.ts
View File

@@ -1,82 +1,183 @@
import { isExperimentalFeatureEnabled, type Settings, type Task, type WorkflowIr, type WorkflowIrEdge, type WorkflowIrNode } from "@fusion/core";
import type { Settings, TaskDetail, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core";
import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled } from "@fusion/core";
export const WORKFLOW_GRAPH_EXECUTOR_FLAG = "workflowGraphExecutor" as const;
import { createDefaultNodeHandlers, createNoopLegacySeams, type WorkflowLegacySeams } from "./workflow-node-handlers.js";
export interface WorkflowGraphExecutorDependencies {
onNode?: (node: WorkflowIrNode) => Promise<void> | void;
export type WorkflowNodeOutcome = "success" | "failure";
export interface WorkflowNodeResult {
outcome: WorkflowNodeOutcome;
value?: string;
contextPatch?: Record<string, unknown>;
}
export interface WorkflowGraphExecutorRunInput {
workflow: WorkflowIr;
settings?: Pick<Settings, "experimentalFeatures">;
task?: Pick<Task, "id">;
export interface WorkflowNodeExecutionContext {
task: TaskDetail;
settings: Pick<Settings, "experimentalFeatures"> | undefined;
context: Record<string, unknown>;
}
export interface WorkflowGraphExecutorRunResult {
export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeExecutionContext) => Promise<WorkflowNodeResult>;
export interface WorkflowGraphExecutorDeps {
handlers?: Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>>;
seams?: WorkflowLegacySeams;
maxRetriesPerNode?: number;
}
export interface WorkflowGraphExecutorResult {
executed: boolean;
outcome: WorkflowNodeOutcome;
context: Record<string, unknown>;
visitedNodeIds: string[];
reason?: "flag-disabled";
}
const TERMINAL_FAILURE: WorkflowGraphExecutorResult = {
executed: false,
outcome: "failure",
context: {},
visitedNodeIds: [],
};
export class WorkflowGraphExecutor {
constructor(private readonly deps: WorkflowGraphExecutorDependencies = {}) {}
private readonly maxRetriesPerNode: number;
async run(input: WorkflowGraphExecutorRunInput): Promise<WorkflowGraphExecutorRunResult> {
if (!isExperimentalFeatureEnabled(input.settings, WORKFLOW_GRAPH_EXECUTOR_FLAG)) {
return { executed: false, visitedNodeIds: [], reason: "flag-disabled" };
}
private readonly handlers: Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>>;
const nodesById = new Map(input.workflow.nodes.map((node) => [node.id, node]));
const outgoingByNode = new Map<string, WorkflowIrEdge[]>();
for (const edge of input.workflow.edges) {
const list = outgoingByNode.get(edge.from) ?? [];
list.push(edge);
outgoingByNode.set(edge.from, list);
}
const startNodes = input.workflow.nodes.filter((node) => node.kind === "start");
if (startNodes.length !== 1) {
throw new Error(`WorkflowGraphExecutor expected exactly one start node, received ${startNodes.length}.`);
}
const visitedNodeIds: string[] = [];
const queue: string[] = [startNodes[0].id];
const seen = new Set<string>();
while (queue.length > 0) {
const nodeId = queue.shift();
if (!nodeId || seen.has(nodeId)) continue;
seen.add(nodeId);
const node = nodesById.get(nodeId);
if (!node) {
throw new Error(`WorkflowGraphExecutor found unknown node id: ${nodeId}`);
}
visitedNodeIds.push(node.id);
await this.dispatchNode(node);
const nextEdges = outgoingByNode.get(node.id) ?? [];
for (const edge of nextEdges) {
queue.push(edge.to);
}
}
return { executed: true, visitedNodeIds };
public constructor(private readonly deps: WorkflowGraphExecutorDeps) {
this.maxRetriesPerNode = Math.max(1, Math.floor(deps.maxRetriesPerNode ?? 2));
this.handlers = {
...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams()),
...(deps.handlers ?? {}),
};
}
private async dispatchNode(node: WorkflowIrNode): Promise<void> {
await this.deps.onNode?.(node);
switch (node.kind) {
case "start":
case "prompt":
case "script":
case "gate":
case "end":
return;
default: {
const exhaustive: never = node.kind;
throw new Error(`Unsupported node kind: ${String(exhaustive)}`);
public async run(
task: TaskDetail,
settings: Pick<Settings, "experimentalFeatures"> | undefined,
ir: WorkflowIr = BUILTIN_CODING_WORKFLOW_IR,
): Promise<WorkflowGraphExecutorResult> {
if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) {
return TERMINAL_FAILURE;
}
const startNode = ir.nodes.find((node) => node.kind === "start");
if (!startNode) throw new WorkflowIrError("Workflow IR missing start node");
const nodeMap = new Map(ir.nodes.map((node) => [node.id, node]));
const outgoingMap = new Map<string, WorkflowIrEdge[]>();
for (const edge of ir.edges) {
if (!nodeMap.has(edge.from) || !nodeMap.has(edge.to)) {
throw new WorkflowIrError(`Workflow IR edge references unknown node: ${edge.from} -> ${edge.to}`);
}
const list = outgoingMap.get(edge.from) ?? [];
list.push(edge);
outgoingMap.set(edge.from, list);
}
const context: Record<string, unknown> = {};
const visitedNodeIds: string[] = [];
const inStack = new Set<string>();
const walk = async (nodeId: string): Promise<WorkflowNodeResult> => {
const node = nodeMap.get(nodeId);
if (!node) throw new WorkflowIrError(`Unknown workflow node: ${nodeId}`);
if (inStack.has(nodeId)) throw new WorkflowIrError(`Cycle detected at node: ${nodeId}`);
inStack.add(nodeId);
visitedNodeIds.push(nodeId);
try {
if (node.kind === "start") {
return await traverseChildren(node, { outcome: "success" });
}
if (node.kind === "end") {
return { outcome: "success" };
}
const result = await this.executeNodeWithRetries(node, task, settings, context);
if (result.contextPatch) Object.assign(context, result.contextPatch);
context[`node:${node.id}:outcome`] = result.outcome;
if (result.value !== undefined) context[`node:${node.id}:value`] = result.value;
return await traverseChildren(node, result);
} finally {
inStack.delete(nodeId);
}
};
const traverseChildren = async (node: WorkflowIrNode, sourceResult: WorkflowNodeResult): Promise<WorkflowNodeResult> => {
const edges = outgoingMap.get(node.id) ?? [];
if (edges.length === 0) {
return sourceResult;
}
const matching = edges.filter((edge) => this.shouldTraverseEdge(edge, sourceResult));
if (matching.length === 0) {
return sourceResult;
}
let aggregate: WorkflowNodeResult = sourceResult;
for (const edge of matching.sort((a, b) => a.to.localeCompare(b.to))) {
const target = nodeMap.get(edge.to);
if (target?.kind === "end") {
aggregate = sourceResult;
continue;
}
const child = await walk(edge.to);
if (child.outcome === "failure") {
aggregate = child;
break;
}
aggregate = child;
}
return aggregate;
};
const terminal = await walk(startNode.id);
return {
executed: true,
outcome: terminal.outcome,
context,
visitedNodeIds,
};
}
private shouldTraverseEdge(edge: WorkflowIrEdge, sourceResult: WorkflowNodeResult): boolean {
if (!edge.condition) return sourceResult.outcome === "success";
if (edge.condition === "success") return sourceResult.outcome === "success";
if (edge.condition === "failure") return sourceResult.outcome === "failure";
if (edge.condition.startsWith("outcome:")) {
return sourceResult.value === edge.condition.slice("outcome:".length);
}
throw new WorkflowIrError(`Unsupported edge condition: ${edge.condition}`);
}
private async executeNodeWithRetries(
node: WorkflowIrNode,
task: TaskDetail,
settings: Pick<Settings, "experimentalFeatures"> | undefined,
context: Record<string, unknown>,
): Promise<WorkflowNodeResult> {
const handler = this.handlers[node.kind];
if (!handler) {
throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`);
}
let lastError: unknown;
for (let attempt = 0; attempt < this.maxRetriesPerNode; attempt++) {
try {
return await handler(node, { task, settings, context });
} catch (error) {
lastError = error;
}
}
return {
outcome: "failure",
value: "exception",
contextPatch: {
[`node:${node.id}:error`]: lastError instanceof Error ? lastError.message : String(lastError),
},
};
}
}

56
packages/engine/src/workflow-node-handlers.ts Normal file
View File

@@ -0,0 +1,56 @@
import { WorkflowIrError } from "@fusion/core";
import type { TaskDetail } from "@fusion/core";
import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";
export type WorkflowSeamName = "execute" | "review" | "merge" | "schedule";
export interface WorkflowLegacySeams {
execute: (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>;
}
function resolveSeam(node: { config?: Record<string, unknown> }): WorkflowSeamName {
const seam = node.config?.seam;
if (seam === "execute" || seam === "review" || seam === "merge" || seam === "schedule") {
return seam;
}
throw new WorkflowIrError(`Unsupported workflow seam: ${String(seam)}`);
}
export function createPromptLikeHandler(seams: WorkflowLegacySeams): WorkflowNodeHandler {
return async (node, context) => {
const seam = resolveSeam(node);
return seams[seam](context.task, context.context);
};
}
export const gateNodeHandler: WorkflowNodeHandler = async (node, context) => {
const expected = node.config?.expect;
const actual = context.context[String(node.config?.contextKey ?? "outcome")];
if (typeof expected === "string" && actual !== expected) {
return { outcome: "failure", value: "gate-mismatch" };
}
return { outcome: "success" };
};
export function createDefaultNodeHandlers(seams: WorkflowLegacySeams): Record<"prompt" | "script" | "gate", WorkflowNodeHandler> {
const promptLike = createPromptLikeHandler(seams);
return {
prompt: promptLike,
script: promptLike,
gate: gateNodeHandler,
};
}
export function createNoopLegacySeams(): WorkflowLegacySeams {
const success = async (): Promise<WorkflowNodeResult> => ({ outcome: "success" });
return {
execute: success,
review: success,
merge: success,
schedule: success,
};
}