feat(engine): WorkflowGraphTaskRunner — interpreter drives a task's lifecycle (CU-U2)
Loads a task's selected workflow, runs the graph with injected legacy seams (execute/review/merge) and a custom-node runner, and maps the terminal outcome to completed/failed/fell-back. Any interpreter-level error falls back so the caller can run the legacy pipeline — a task is never stranded. Covered with fake seams: lifecycle ordering, failure routing, gate blocking, fallback reasons, diagnostics isolation. Includes the interpreter-cutover plan doc.
This commit is contained in:
gsxdsm
@@ -0,0 +1,82 @@
|
||||
---
|
||||
title: "feat: Workflow interpreter cutover — graph owns the full lifecycle"
|
||||
type: feat
|
||||
status: active
|
||||
date: 2026-06-03
|
||||
depth: deep
|
||||
origin: docs/plans/2026-06-03-001-feat-executable-custom-workflows-node-editor-plan.md (Deferred Track)
|
||||
---
|
||||
|
||||
# feat: Workflow interpreter cutover — graph owns the full lifecycle
|
||||
|
||||
## Summary
|
||||
|
||||
Promote `WorkflowGraphExecutor` from a flag-off no-op to the authoritative driver of a task's lifecycle, so a custom workflow graph can **replace** planning → execute → review → merge, not just inject steps around it. The user has explicitly waived the FN-4359 reliability freeze for this track; the parity invariants below remain the correctness bar regardless.
|
||||
|
||||
The existing scaffold already provides: graph walking with cycle detection, `success`/`failure`/`outcome:*` edge conditions, per-node retries, a `WorkflowLegacySeams` interface (`execute/review/merge/schedule`), the `workflowGraphExecutor` experimental flag, and the dual-observe parity machinery. The MVP (plan 001) provides: persisted workflow definitions, the node editor, and per-task selection. What's missing is **real seam implementations**, **custom-node handlers** (the default handlers throw for non-seam prompt/script nodes), and the **entry point** that routes a graph-selected task through the interpreter.
|
||||
|
||||
---
|
||||
|
||||
## Scope Boundaries
|
||||
|
||||
### In scope
|
||||
- M-A: Interpreter foundation — custom-node handlers + a `WorkflowGraphTaskRunner` with injected seams, fully tested with fakes.
|
||||
- M-B: Real seam wiring — delegate execute/review/merge to the legacy engine implementations via narrow injected callbacks from `TaskExecutor`/`ProjectEngine`.
|
||||
- M-C: Flag-gated entry point — graph-selected tasks route through the interpreter when `experimentalFeatures.workflowGraphExecutor` is on; legacy fallback on any interpreter error.
|
||||
- M-D: Parity + graduation — dual-observe on real runs, drive drift to zero, then default the flag on for graph-selected tasks.
|
||||
|
||||
### Deferred to Follow-Up Work
|
||||
- Removing the legacy hardcoded pipeline (FN-5719 Phase 4) — only after M-D proves parity in the field.
|
||||
- Planning/triage as a replaceable seam (the `schedule` seam exists but triage replacement needs the planning subsystem mapped first).
|
||||
- Parallel branches (fan-out) executing concurrently — the walker is sequential; concurrency within a graph is a later extension.
|
||||
|
||||
---
|
||||
|
||||
## Key Technical Decisions
|
||||
|
||||
- **KTD-1 — Seams delegate, never reimplement.** The `execute`/`review`/`merge` seam implementations call the same engine functions the legacy path uses (agent session machinery, `reviewStep`, the auto-merge queue). The interpreter owns *sequencing*; the engine keeps owning *mechanics* (worktrees, leases, file-scope guard, squash contract, self-healing). This is the enqueue-only posture from DAG ADR-0001 applied to seams.
|
||||
- **KTD-2 — Custom nodes run on the WorkflowStep machinery.** Non-seam prompt/script/gate nodes execute via the same prompt-session/script/verdict machinery as workflow steps (proven, readonly-tool-policy aware), invoked through an injected `runCustomNode` callback — the interpreter stays engine-agnostic and unit-testable with fakes.
|
||||
- **KTD-3 — Legacy fallback on interpreter error (M-C).** Any thrown error from the interpreter path (not a graph-routed `failure` edge) falls back to the legacy pipeline for that task and emits an audit event. No task is ever stranded by interpreter bugs.
|
||||
- **KTD-4 — Column transitions are seam side-effects.** `execute` seam entry → `in-progress`, review handoff → `in-review`, merge success → `done` — performed by the delegated engine code itself (KTD-1), so board invariants (FN-5147 terminal-until-merged, hard-cancel) are preserved by construction.
|
||||
- **KTD-5 — Invariant bar (from FN-5719 / workflow-steps.md):** `FileScopeViolationError` guard, squash/merge contract, `autoMerge:false` terminal-until-merged, `moveTask(in-progress→todo)` hard-cancel, resume-limbo non-oscillation. Parity is machine-checked via the existing `compareWorkflowRunObservations` / `compareWorkflowRunAudits`.
|
||||
|
||||
---
|
||||
|
||||
## Implementation Units
|
||||
|
||||
### U1. Custom-node handlers (unblock non-seam nodes)
|
||||
**Goal:** Default handlers route seam-configured prompt/script nodes to seams and **custom** prompt/script nodes to an injected runner instead of throwing.
|
||||
**Files:** `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/__tests__/workflow-node-handlers.test.ts`.
|
||||
**Approach:** `createDefaultNodeHandlers(seams, runCustomNode)` — `resolveSeam` returns undefined (not throw) for non-seam nodes; gate nodes keep the context-gate behavior but also support prompt/script-backed gates via `runCustomNode` with `gateMode` semantics.
|
||||
**Test scenarios:** seam node → seam called; custom prompt node → runner called with node config; custom script node → runner; gate node with context expectation → pass/fail; gate node with prompt config → runner verdict drives outcome; unknown seam string → error.
|
||||
|
||||
### U2. WorkflowGraphTaskRunner (engine-agnostic orchestration)
|
||||
**Goal:** A runner that loads a task's selected workflow IR, runs `WorkflowGraphExecutor` with injected seams + custom-node runner, and maps the terminal outcome to a lifecycle disposition (`completed` / `failed` / `fell-back`).
|
||||
**Files:** `packages/engine/src/workflow-graph-task-runner.ts` (new), `packages/engine/src/__tests__/workflow-graph-task-runner.test.ts` (new).
|
||||
**Approach:** Pure DI: `{ store, seams, runCustomNode, settings }`. Reads selection via `store.getTaskWorkflowSelection` + `getWorkflowDefinition`. Falls back (disposition `fell-back`) when flag off, no selection, or IR load fails. Audit events for start/terminal/fallback.
|
||||
**Test scenarios:** full graph run order with fake seams (execute→review→merge sequencing via the builtin IR); custom pre-merge node runs between start and execute when authored that way; failure edge routes to end with `failed`; thrown seam error → `fell-back`; flag off → `fell-back`; gate failure blocks merge seam.
|
||||
|
||||
### U3. Real seam implementations (M-B)
|
||||
**Goal:** `createEngineSeams(executor, projectEngine)` delegating to real engine entry points.
|
||||
**Files:** `packages/engine/src/workflow-engine-seams.ts` (new), executor/project-engine narrow accessor methods as needed, `packages/engine/src/__tests__/workflow-engine-seams.test.ts`.
|
||||
**Approach:** execute → the executor's implementation-phase entry for an already-claimed task; review → `reviewStep` path with verdict mapped to `outcome:*`; merge → enqueue on the auto-merge queue and await terminal merge outcome. Each seam returns `WorkflowNodeResult` with `value` carrying verdict/outcome tokens for edge conditions.
|
||||
**Execution note:** characterization-first — capture the legacy call sequence for one task end-to-end before extracting accessors.
|
||||
|
||||
### U4. Flag-gated entry point + fallback (M-C)
|
||||
**Goal:** Graph-selected tasks route through the runner from `TaskExecutor.execute`; interpreter errors fall back to legacy mid-flight where safe, else fail the task through existing recovery.
|
||||
**Files:** `packages/engine/src/executor.ts` (top-of-execute branch only), `packages/engine/src/__tests__/workflow-graph-entry.test.ts`.
|
||||
|
||||
### U5. Dual-observe parity on real runs + graduation (M-D)
|
||||
**Goal:** Enable `workflowInterpreterDualObserve` for graph-selected tasks, record drift, fix until `{agree:true}` is sustained, then flip `workflowGraphExecutor` default for selected-workflow tasks.
|
||||
**Files:** `packages/engine/src/workflow-parity-observer.ts` call-site wiring, settings default change, docs update.
|
||||
|
||||
---
|
||||
|
||||
## Risks
|
||||
|
||||
- **TaskExecutor.execute is ~3k lines of intertwined state.** Mitigation: U3 extracts *accessors*, never moves logic; U4 touches only a top-of-function branch; characterization tests first.
|
||||
- **Self-healing/recovery assume the legacy shape.** Mitigation: KTD-3 fallback + parity observation before authority; recovery paths treat interpreter tasks as legacy until M-D.
|
||||
- **Merge queue awaiting from inside a graph walk** could deadlock with the executor's own lifecycle. Mitigation: merge seam enqueues and resolves on the queue's completion callback (same contract the legacy handoff uses), never polls.
|
||||
|
||||
## Sources
|
||||
Plan 001 (Deferred Track), `workflow-graph-executor.ts`, `workflow-node-handlers.ts`, `workflow-parity-observer.ts`, `docs/rfcs/FN-5719-decouple-executor-merger.md`, `docs/dag/adr-0001-dag-orchestration.md`, `docs/workflow-steps.md`.
|
||||
211
packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
Normal file
211
packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
Normal file
@@ -0,0 +1,211 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr } from "@fusion/core";
|
||||
|
||||
import { WorkflowGraphTaskRunner, type WorkflowGraphRunnerStore } from "../workflow-graph-task-runner.js";
|
||||
import type { WorkflowNodeResult } from "../workflow-graph-executor.js";
|
||||
|
||||
const task = { id: "FN-9001" } as TaskDetail;
|
||||
const flagOn = { experimentalFeatures: { workflowGraphExecutor: true } } as unknown as Pick<
|
||||
Settings,
|
||||
"experimentalFeatures"
|
||||
>;
|
||||
const flagOff = { experimentalFeatures: {} } as unknown as Pick<Settings, "experimentalFeatures">;
|
||||
|
||||
/** start → lint(custom) → execute → review → merge → notify(custom) → end, with seam failure edges to end. */
|
||||
function fullLifecycleIr(): WorkflowIr {
|
||||
return {
|
||||
version: "v1",
|
||||
name: "full",
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "lint", kind: "prompt", config: { prompt: "lint it" } },
|
||||
{ id: "execute", kind: "prompt", config: { seam: "execute" } },
|
||||
{ id: "review", kind: "prompt", config: { seam: "review" } },
|
||||
{ id: "merge", kind: "prompt", config: { seam: "merge" } },
|
||||
{ id: "notify", kind: "script", config: { scriptName: "notify" } },
|
||||
{ id: "zend", kind: "end" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "lint" },
|
||||
{ from: "lint", to: "execute", condition: "success" },
|
||||
{ from: "execute", to: "review", condition: "success" },
|
||||
{ from: "review", to: "merge", condition: "success" },
|
||||
{ from: "merge", to: "notify", condition: "success" },
|
||||
{ from: "notify", to: "zend", condition: "success" },
|
||||
{ from: "execute", to: "zend", condition: "failure" },
|
||||
{ from: "review", to: "zend", condition: "failure" },
|
||||
{ from: "merge", to: "zend", condition: "failure" },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function definition(ir: WorkflowIr): WorkflowDefinition {
|
||||
return {
|
||||
id: "WF-001",
|
||||
name: "Full lifecycle",
|
||||
description: "",
|
||||
ir,
|
||||
layout: {},
|
||||
createdAt: "2026-06-03T00:00:00.000Z",
|
||||
updatedAt: "2026-06-03T00:00:00.000Z",
|
||||
};
|
||||
}
|
||||
|
||||
function storeWith(def: WorkflowDefinition | undefined, workflowId = "WF-001"): WorkflowGraphRunnerStore {
|
||||
return {
|
||||
getTaskWorkflowSelection: () => (def ? { workflowId, stepIds: [] } : undefined),
|
||||
getWorkflowDefinition: async () => def,
|
||||
};
|
||||
}
|
||||
|
||||
function recordingSeams(calls: string[], overrides: Partial<Record<string, WorkflowNodeResult>> = {}) {
|
||||
const seam = (name: string) => async (): Promise<WorkflowNodeResult> => {
|
||||
calls.push(name);
|
||||
return overrides[name] ?? { outcome: "success" };
|
||||
};
|
||||
return {
|
||||
execute: seam("execute"),
|
||||
review: seam("review"),
|
||||
merge: seam("merge"),
|
||||
schedule: seam("schedule"),
|
||||
};
|
||||
}
|
||||
|
||||
describe("WorkflowGraphTaskRunner (CU-U2)", () => {
|
||||
it("runs the full lifecycle in graph order: custom → execute → review → merge → custom", async () => {
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.run(task, flagOn);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["custom:lint", "execute", "review", "merge", "custom:notify"]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "lint", "execute", "review", "merge", "notify"]);
|
||||
});
|
||||
|
||||
it("a failing seam terminates the run as failed without running later nodes", async () => {
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams(calls, { review: { outcome: "failure", value: "REVISE" } }),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.run(task, flagOn);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(calls).toEqual(["custom:lint", "execute", "review"]);
|
||||
expect(calls).not.toContain("merge");
|
||||
});
|
||||
|
||||
it("a failing custom gate before execute blocks the whole pipeline", async () => {
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return node.id === "lint" ? { outcome: "failure", value: "lint-failed" } : { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.run(task, flagOn);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(calls).toEqual(["custom:lint"]);
|
||||
});
|
||||
|
||||
it("falls back when the flag is off", async () => {
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
const result = await runner.run(task, flagOff);
|
||||
expect(result.disposition).toBe("fell-back");
|
||||
expect(result.reason).toBe("flag-off");
|
||||
});
|
||||
|
||||
it("falls back when the task has no workflow selection", async () => {
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(undefined),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result).toMatchObject({ disposition: "fell-back", reason: "no-selection" });
|
||||
});
|
||||
|
||||
it("falls back when the selected workflow no longer exists", async () => {
|
||||
const store: WorkflowGraphRunnerStore = {
|
||||
getTaskWorkflowSelection: () => ({ workflowId: "WF-404", stepIds: [] }),
|
||||
getWorkflowDefinition: async () => undefined,
|
||||
};
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store,
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result.disposition).toBe("fell-back");
|
||||
expect(result.reason).toMatch(/workflow-missing/);
|
||||
});
|
||||
|
||||
it("falls back (never strands the task) when the interpreter throws", async () => {
|
||||
// Malformed graph: edge references unknown node → WorkflowIrError inside run().
|
||||
const badIr: WorkflowIr = {
|
||||
version: "v1",
|
||||
name: "bad",
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "end", kind: "end" },
|
||||
],
|
||||
edges: [{ from: "start", to: "ghost" }],
|
||||
};
|
||||
const events: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(badIr)),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
onEvent: (e) => events.push(e.type),
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result.disposition).toBe("fell-back");
|
||||
expect(result.reason).toMatch(/interpreter-error/);
|
||||
expect(events).toContain("fallback");
|
||||
});
|
||||
|
||||
it("exposes node outcomes in the shared context for downstream consumers", async () => {
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success", value: "APPROVE" }),
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result.context?.["node:lint:outcome"]).toBe("success");
|
||||
expect(result.context?.["node:lint:value"]).toBe("APPROVE");
|
||||
});
|
||||
|
||||
it("onEvent diagnostics failures never affect the run", async () => {
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
onEvent: () => {
|
||||
throw new Error("diagnostics boom");
|
||||
},
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result.disposition).toBe("completed");
|
||||
});
|
||||
});
|
||||
@@ -24,9 +24,17 @@ export {
|
||||
export {
|
||||
createDefaultNodeHandlers,
|
||||
createNoopLegacySeams,
|
||||
type WorkflowCustomNodeRunner,
|
||||
type WorkflowLegacySeams,
|
||||
type WorkflowSeamName,
|
||||
} from "./workflow-node-handlers.js";
|
||||
export {
|
||||
WorkflowGraphTaskRunner,
|
||||
type WorkflowGraphRunDisposition,
|
||||
type WorkflowGraphRunnerStore,
|
||||
type WorkflowGraphTaskRunResult,
|
||||
type WorkflowGraphTaskRunnerDeps,
|
||||
} from "./workflow-graph-task-runner.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";
|
||||
|
||||
117
packages/engine/src/workflow-graph-task-runner.ts
Normal file
117
packages/engine/src/workflow-graph-task-runner.ts
Normal file
@@ -0,0 +1,117 @@
|
||||
import type { Settings, TaskDetail, WorkflowDefinition } from "@fusion/core";
|
||||
import { isExperimentalFeatureEnabled } from "@fusion/core";
|
||||
|
||||
import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js";
|
||||
import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
|
||||
|
||||
/**
|
||||
* Terminal disposition of an interpreter-driven task run.
|
||||
* - "completed" — the graph ran to its end node successfully.
|
||||
* - "failed" — the graph ran and terminated on a failure outcome.
|
||||
* - "fell-back" — the interpreter did not (or could not) own this task;
|
||||
* the caller must run the legacy pipeline instead.
|
||||
*/
|
||||
export type WorkflowGraphRunDisposition = "completed" | "failed" | "fell-back";
|
||||
|
||||
export interface WorkflowGraphTaskRunResult {
|
||||
disposition: WorkflowGraphRunDisposition;
|
||||
outcome?: WorkflowNodeOutcome;
|
||||
visitedNodeIds: string[];
|
||||
/** Why the runner fell back (flag-off, no-selection, workflow-missing, interpreter-error). */
|
||||
reason?: string;
|
||||
/** Shared graph context after the run (node outcomes/values). */
|
||||
context?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** The minimal store surface the runner needs — keeps tests fake-friendly. */
|
||||
export interface WorkflowGraphRunnerStore {
|
||||
getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined;
|
||||
getWorkflowDefinition(id: string): Promise<WorkflowDefinition | undefined>;
|
||||
}
|
||||
|
||||
export interface WorkflowGraphTaskRunnerDeps {
|
||||
store: WorkflowGraphRunnerStore;
|
||||
seams: WorkflowLegacySeams;
|
||||
runCustomNode: WorkflowCustomNodeRunner;
|
||||
maxRetriesPerNode?: number;
|
||||
/** Optional diagnostics hook (audit/log emission). Never throws into the run. */
|
||||
onEvent?: (event: { type: "start" | "terminal" | "fallback"; taskId: string; detail: string }) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Drives a task's lifecycle from its selected workflow graph. The runner owns
|
||||
* SEQUENCING only — seam nodes delegate to the legacy engine implementations
|
||||
* (execute/review/merge), custom nodes run via the injected runner. Any
|
||||
* interpreter-level error yields a "fell-back" disposition so the caller can
|
||||
* run the legacy pipeline; a task is never stranded by interpreter bugs.
|
||||
*/
|
||||
export class WorkflowGraphTaskRunner {
|
||||
public constructor(private readonly deps: WorkflowGraphTaskRunnerDeps) {}
|
||||
|
||||
private emit(type: "start" | "terminal" | "fallback", taskId: string, detail: string): void {
|
||||
try {
|
||||
this.deps.onEvent?.({ type, taskId, detail });
|
||||
} catch {
|
||||
// Diagnostics must never affect the run.
|
||||
}
|
||||
}
|
||||
|
||||
private fallBack(taskId: string, reason: string): WorkflowGraphTaskRunResult {
|
||||
this.emit("fallback", taskId, reason);
|
||||
return { disposition: "fell-back", reason, visitedNodeIds: [] };
|
||||
}
|
||||
|
||||
public async run(
|
||||
task: TaskDetail,
|
||||
settings: Pick<Settings, "experimentalFeatures"> | undefined,
|
||||
): Promise<WorkflowGraphTaskRunResult> {
|
||||
if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) {
|
||||
return this.fallBack(task.id, "flag-off");
|
||||
}
|
||||
|
||||
let selection: { workflowId: string; stepIds: string[] } | undefined;
|
||||
try {
|
||||
selection = this.deps.store.getTaskWorkflowSelection(task.id);
|
||||
} catch (err) {
|
||||
return this.fallBack(task.id, `selection-error: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
if (!selection) {
|
||||
return this.fallBack(task.id, "no-selection");
|
||||
}
|
||||
|
||||
let definition: WorkflowDefinition | undefined;
|
||||
try {
|
||||
definition = await this.deps.store.getWorkflowDefinition(selection.workflowId);
|
||||
} catch (err) {
|
||||
return this.fallBack(task.id, `workflow-load-error: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
if (!definition) {
|
||||
return this.fallBack(task.id, `workflow-missing: ${selection.workflowId}`);
|
||||
}
|
||||
|
||||
this.emit("start", task.id, definition.id);
|
||||
try {
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
seams: this.deps.seams,
|
||||
runCustomNode: this.deps.runCustomNode,
|
||||
maxRetriesPerNode: this.deps.maxRetriesPerNode,
|
||||
});
|
||||
const result = await executor.run(task, settings, definition.ir);
|
||||
if (!result.executed) {
|
||||
return this.fallBack(task.id, "not-executed");
|
||||
}
|
||||
const disposition: WorkflowGraphRunDisposition = result.outcome === "success" ? "completed" : "failed";
|
||||
this.emit("terminal", task.id, `${definition.id}:${disposition}`);
|
||||
return {
|
||||
disposition,
|
||||
outcome: result.outcome,
|
||||
visitedNodeIds: result.visitedNodeIds,
|
||||
context: result.context,
|
||||
};
|
||||
} catch (err) {
|
||||
// Interpreter-level error (bad IR, handler wiring, etc.) — never strand
|
||||
// the task; the caller runs the legacy pipeline.
|
||||
return this.fallBack(task.id, `interpreter-error: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user