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

Merge pull request #1498 from Runfusion/feature/engine-plugin

Browse Source 
[codex] Add workflow extension plugin seams
This commit is contained in:
gsxdsm
2026-06-07 22:57:04 -07:00
committed by GitHub
parent cf80622875 6d47d1abd0
commit de254858f2
32 changed files with 2525 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
packages/engine/src/__tests__/auto-merge-fact-providers.test.ts Normal file
View File

@@ -0,0 +1,59 @@
// @vitest-environment node
import { afterEach, describe, expect, it, vi } from "vitest";
import {
WORKFLOW_EXTENSION_SCHEMA_VERSION,
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
type TaskDetail,
type WorkflowIr,
} from "@fusion/core";
import { evaluateAutoMergeFactProviders } from "../auto-merge-fact-providers.js";
describe("auto-merge fact providers", () => {
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
});
it("collects facts and chooses the strictest route", async () => {
const workflow: WorkflowIr = { version: "v2", name: "w", columns: [], nodes: [], edges: [] };
const task = { id: "FN-MERGE" } as TaskDetail;
const store = {
getTaskWorkflowSelection: vi.fn().mockReturnValue({ workflowId: "custom-workflow", stepIds: [] }),
getWorkflowDefinition: vi.fn().mockResolvedValue({ ir: workflow }),
};
getWorkflowExtensionRegistry().register("merge-plugin", {
extensionId: "facts",
name: "Facts",
kind: "merge-fact-provider",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
collect: vi.fn().mockResolvedValue({
route: "manual-required",
facts: { needsOwner: true },
reason: "owner approval required",
}),
});
getWorkflowExtensionRegistry().register("merge-plugin", {
extensionId: "blocker",
name: "Blocker",
kind: "merge-fact-provider",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
collect: vi.fn().mockResolvedValue({
route: "blocked",
facts: { risk: "high" },
reason: "risk gate blocked",
}),
});
await expect(evaluateAutoMergeFactProviders(store, task)).resolves.toEqual({
route: "blocked",
facts: {
"plugin:merge-plugin:facts": { needsOwner: true },
"plugin:merge-plugin:blocker": { risk: "high" },
},
reasons: ["owner approval required", "risk gate blocked"],
});
});
});

33
packages/engine/src/__tests__/merger-merge-lifecycle.test.ts
View File

@@ -512,6 +512,38 @@ describe("aiMergeTask pre-merge fetch + fast-forward (smart strategies)", () =>
expect(store.upsertMergeRequestRecord).toHaveBeenCalledWith("FN-5741-MANUAL", { state: "manual-required" });
expect(store.transitionMergeRequestState).not.toHaveBeenCalledWith("FN-5741-MANUAL", "running");
});
it("keeps globally disabled auto-merge records in manual-required unless the task opts in", async () => {
const store = createMockStore({ id: "FN-5741-GLOBAL-MANUAL", worktree: "/tmp/root", autoMerge: undefined });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
...DEFAULT_SETTINGS,
mergeIntegrationWorktree: "cwd-main" as const,
mergeRequestContractShadowEnabled: true,
autoMerge: false,
});
setupSyncMock({ behind: 0, ahead: 0 });
await aiMergeTask(store, "/tmp/root", "FN-5741-GLOBAL-MANUAL");
expect(store.upsertMergeRequestRecord).toHaveBeenCalledWith("FN-5741-GLOBAL-MANUAL", { state: "manual-required" });
expect(store.transitionMergeRequestState).not.toHaveBeenCalledWith("FN-5741-GLOBAL-MANUAL", "running");
});
it("allows task-level autoMerge true to opt into shadow merge when global auto-merge is disabled", async () => {
const store = createMockStore({ id: "FN-5741-TASK-OPT-IN", worktree: "/tmp/root", autoMerge: true });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
...DEFAULT_SETTINGS,
mergeIntegrationWorktree: "cwd-main" as const,
mergeRequestContractShadowEnabled: true,
autoMerge: false,
});
setupSyncMock({ behind: 0, ahead: 0 });
await aiMergeTask(store, "/tmp/root", "FN-5741-TASK-OPT-IN");
expect(store.upsertMergeRequestRecord).toHaveBeenCalledWith("FN-5741-TASK-OPT-IN", { state: "queued" });
expect(store.transitionMergeRequestState).toHaveBeenCalledWith("FN-5741-TASK-OPT-IN", "running");
});
});
@@ -3219,4 +3251,3 @@ describe("aiMergeTask post-squash audit gate", () => {
});
});

48
packages/engine/src/__tests__/plugin-runner.test.ts
View File

@@ -7,7 +7,14 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { PluginRunner, type PluginRunnerOptions } from "../plugin-runner.js";
import type { PluginLoader, PluginStore, PluginInstallation } from "@fusion/core";
import {
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
workflowExtensionRegistryId,
type PluginLoader,
type PluginStore,
type PluginInstallation,
} from "@fusion/core";
import type { FusionPlugin, PluginToolDefinition } from "@fusion/core";
import { createLogger } from "../logger.js";
@@ -39,6 +46,7 @@ describe("PluginRunner", () => {
getCliProviderContributions: ReturnType<typeof vi.fn>;
getPluginSkills: ReturnType<typeof vi.fn>;
getPluginWorkflowSteps: ReturnType<typeof vi.fn>;
getPluginWorkflowExtensions: ReturnType<typeof vi.fn>;
getPluginWorkflowStepTemplates: ReturnType<typeof vi.fn>;
getPluginPromptContributions: ReturnType<typeof vi.fn>;
getPluginSetupInfo: ReturnType<typeof vi.fn>;
@@ -63,6 +71,7 @@ describe("PluginRunner", () => {
off: ReturnType<typeof vi.fn>;
getTask: ReturnType<typeof vi.fn>;
getDatabase: ReturnType<typeof vi.fn>;
recordRunAuditEvent: ReturnType<typeof vi.fn>;
};
let pluginRunner: PluginRunner;
@@ -104,6 +113,7 @@ describe("PluginRunner", () => {
getCliProviderContributions: vi.fn().mockReturnValue([]),
getPluginSkills: vi.fn().mockReturnValue([]),
getPluginWorkflowSteps: vi.fn().mockReturnValue([]),
getPluginWorkflowExtensions: vi.fn().mockReturnValue([]),
getPluginWorkflowStepTemplates: vi.fn().mockReturnValue([]),
getPluginPromptContributions: vi.fn().mockReturnValue([]),
getPluginSetupInfo: vi.fn().mockReturnValue([]),
@@ -127,6 +137,7 @@ describe("PluginRunner", () => {
off: mockOff,
getTask: vi.fn(),
getDatabase: vi.fn().mockReturnValue({ runPluginSchemaInits: mockRunPluginSchemaInits }),
recordRunAuditEvent: vi.fn(),
};
mockPluginStore = {
@@ -150,6 +161,7 @@ describe("PluginRunner", () => {
});
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
vi.clearAllMocks();
});
@@ -878,22 +890,50 @@ describe("PluginRunner", () => {
expect(second).toBe(first);
});
it("returns workflow steps, workflow step templates, prompt contributions, and setup info", async () => {
it("returns workflow steps, workflow extensions, workflow step templates, prompt contributions, and setup info", async () => {
const steps = [{ pluginId: "test-plugin", step: { stepId: "ws1", name: "Step", description: "d", mode: "prompt", prompt: "Run checks" } }];
const extensions = [{ pluginId: "test-plugin", extension: { extensionId: "move-policy", name: "Move Policy", kind: "move-policy", schemaVersion: 1, fallback: "degradeToDefault" } }];
const templates = [{ pluginId: "test-plugin", template: { id: "plugin:test-plugin:ws1", name: "Step", description: "d", prompt: "Run checks", category: "Plugin", icon: "puzzle" } }];
const prompts = [{ pluginId: "test-plugin", contribution: { surface: "executor-system", content: "extra" }, config: { enabledByDefault: true, contributions: [] } }];
const setups = [{ pluginId: "test-plugin", manifest: { binaryName: "agent-browser", description: "Do it" }, hooks: { checkSetup: vi.fn().mockResolvedValue({ status: "installed" }) } }];
mockPluginLoader.getPluginWorkflowSteps.mockReturnValue(steps);
mockPluginLoader.getPluginWorkflowExtensions.mockReturnValue(extensions);
mockPluginLoader.getPluginWorkflowStepTemplates.mockReturnValue(templates);
mockPluginLoader.getPluginPromptContributions.mockReturnValue(prompts);
mockPluginLoader.getPluginSetupInfo.mockReturnValue(setups);
await pluginRunner.init();
expect(pluginRunner.getPluginWorkflowSteps()).toEqual(steps);
expect(pluginRunner.getPluginWorkflowExtensions()).toEqual(extensions);
expect(pluginRunner.getPluginWorkflowStepTemplates()).toEqual(templates);
expect(pluginRunner.getPluginPromptContributions()).toEqual(prompts);
expect(pluginRunner.getPluginSetupInfo()).toEqual(setups);
});
it("syncPluginWorkflowExtensions preserves force-degraded definitions across cache invalidation", async () => {
const extensions = [{
pluginId: "test-plugin",
extension: {
extensionId: "move-policy",
name: "Move Policy",
kind: "move-policy",
schemaVersion: 1,
fallback: "degradeToDefault",
},
}];
mockPluginLoader.getPluginWorkflowExtensions.mockReturnValue(extensions);
const id = workflowExtensionRegistryId("test-plugin", "move-policy");
pluginRunner.syncPluginWorkflowExtensions();
const degraded = pluginRunner.disablePluginWorkflowExtensions("test-plugin", { force: true });
(pluginRunner as unknown as { invalidateWorkflowExtensionsCache: () => void }).invalidateWorkflowExtensionsCache();
pluginRunner.syncPluginWorkflowExtensions();
expect(degraded.degraded).toEqual([id]);
expect(getWorkflowExtensionRegistry().get(id)?.degraded).toMatchObject({
reason: "force-disabled",
});
});
it("getPromptContributionsForSurface filters by surface", async () => {
mockPluginLoader.getPluginPromptContributions.mockReturnValue([
{ pluginId: "test-plugin", contribution: { surface: "executor-system", content: "ok" }, config: { enabledByDefault: true, contributions: [] } },
@@ -921,6 +961,7 @@ describe("PluginRunner", () => {
pluginRunner.getCliProviderContributions();
pluginRunner.getPluginSkills();
pluginRunner.getPluginWorkflowSteps();
pluginRunner.getPluginWorkflowExtensions();
pluginRunner.getPluginWorkflowStepTemplates();
pluginRunner.getPluginPromptContributions();
pluginRunner.getPluginSetupInfo();
@@ -930,6 +971,7 @@ describe("PluginRunner", () => {
pluginRunner.getCliProviderContributions();
pluginRunner.getPluginSkills();
pluginRunner.getPluginWorkflowSteps();
pluginRunner.getPluginWorkflowExtensions();
pluginRunner.getPluginWorkflowStepTemplates();
pluginRunner.getPluginPromptContributions();
pluginRunner.getPluginSetupInfo();
@@ -939,6 +981,7 @@ describe("PluginRunner", () => {
pluginRunner.getCliProviderContributions();
pluginRunner.getPluginSkills();
pluginRunner.getPluginWorkflowSteps();
pluginRunner.getPluginWorkflowExtensions();
pluginRunner.getPluginWorkflowStepTemplates();
pluginRunner.getPluginPromptContributions();
pluginRunner.getPluginSetupInfo();
@@ -946,6 +989,7 @@ describe("PluginRunner", () => {
expect(mockPluginLoader.getCliProviderContributions).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginSkills).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginWorkflowSteps).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginWorkflowExtensions).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginWorkflowStepTemplates).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginPromptContributions).toHaveBeenCalledTimes(3);
expect(mockPluginLoader.getPluginSetupInfo).toHaveBeenCalledTimes(3);

145
packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts Normal file
View File

@@ -0,0 +1,145 @@
// @vitest-environment node
import { afterEach, describe, expect, it, vi } from "vitest";
import {
WORKFLOW_EXTENSION_SCHEMA_VERSION,
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
workflowExtensionRegistryId,
type TaskDetail,
type WorkflowIr,
} from "@fusion/core";
import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
const settingsOn = { experimentalFeatures: { workflowGraphExecutor: true } };
describe("workflow node-handler extensions", () => {
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
});
it("executes an extension-marked node and routes custom outcomes", async () => {
const extensionKey = workflowExtensionRegistryId("node-plugin", "decision");
const handle = vi.fn().mockResolvedValue({
outcome: "outcome:needs-human",
contextPatch: { decidedBy: "plugin" },
});
getWorkflowExtensionRegistry().register("node-plugin", {
extensionId: "decision",
name: "Decision",
kind: "node-handler",
nodeKind: "prompt",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
handle,
});
const workflow: WorkflowIr = {
version: "v2",
name: "node-extension",
columns: [{ id: "work", name: "Work", traits: [] }],
nodes: [
{ id: "start", kind: "start" },
{ id: "decide", kind: "prompt", column: "work", extensions: { [extensionKey]: {} } },
{ id: "human", kind: "prompt", column: "work", config: { prompt: "human" } },
{ id: "default", kind: "end" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "decide" },
{ from: "decide", to: "human", condition: "outcome:needs-human" },
{ from: "decide", to: "default", condition: "success" },
{ from: "human", to: "end" },
],
};
const prompt = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
const result = await executor.run({ id: "FN-NODE" } as TaskDetail, settingsOn, workflow);
expect(result.outcome).toBe("success");
expect(result.visitedNodeIds).toEqual(["start", "decide", "human"]);
expect(result.context).toMatchObject({ decidedBy: "plugin" });
expect(handle).toHaveBeenCalledWith(expect.objectContaining({
node: expect.objectContaining({ id: "decide" }),
workflow,
}));
expect(prompt).toHaveBeenCalledWith(expect.objectContaining({ id: "human" }), expect.any(Object));
});
it("preserves plugin-provided values for custom outcomes", async () => {
const extensionKey = workflowExtensionRegistryId("node-plugin", "decision");
getWorkflowExtensionRegistry().register("node-plugin", {
extensionId: "decision",
name: "Decision",
kind: "node-handler",
nodeKind: "prompt",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
handle: vi.fn().mockResolvedValue({
outcome: "outcome:ignored-route",
value: "needs-human",
}),
});
const workflow: WorkflowIr = {
version: "v2",
name: "node-extension",
columns: [{ id: "work", name: "Work", traits: [] }],
nodes: [
{ id: "start", kind: "start" },
{ id: "decide", kind: "prompt", column: "work", extensions: { [extensionKey]: {} } },
{ id: "human", kind: "prompt", column: "work", config: { prompt: "human" } },
{ id: "default", kind: "end" },
],
edges: [
{ from: "start", to: "decide" },
{ from: "decide", to: "human", condition: "outcome:needs-human" },
{ from: "decide", to: "default", condition: "outcome:ignored-route" },
],
};
const prompt = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
const result = await executor.run({ id: "FN-NODE" } as TaskDetail, settingsOn, workflow);
expect(result.outcome).toBe("success");
expect(result.visitedNodeIds).toEqual(["start", "decide", "human"]);
});
it("degrades faulty node handlers before falling through to default handler", async () => {
const extensionKey = workflowExtensionRegistryId("node-plugin", "decision");
getWorkflowExtensionRegistry().register("node-plugin", {
extensionId: "decision",
name: "Decision",
kind: "node-handler",
nodeKind: "prompt",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "degradeToDefault",
handle: vi.fn().mockRejectedValue(new Error("handler failed")),
});
const workflow: WorkflowIr = {
version: "v2",
name: "node-extension",
columns: [{ id: "work", name: "Work", traits: [] }],
nodes: [
{ id: "start", kind: "start" },
{ id: "decide", kind: "prompt", column: "work", extensions: { [extensionKey]: {} } },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "decide" },
{ from: "decide", to: "end" },
],
};
const prompt = vi.fn(async () => ({ outcome: "success" as const }));
const executor = new WorkflowGraphExecutor({ handlers: { prompt } });
const result = await executor.run({ id: "FN-NODE" } as TaskDetail, settingsOn, workflow);
expect(result.outcome).toBe("success");
expect(prompt).toHaveBeenCalledWith(expect.objectContaining({ id: "decide" }), expect.any(Object));
expect(getWorkflowExtensionRegistry().get(extensionKey)?.degraded).toMatchObject({
reason: "runtime-fault",
message: "handler failed",
});
});
});

63
packages/engine/src/__tests__/workflow-verdict-provider-extensions.test.ts Normal file
View File

@@ -0,0 +1,63 @@
// @vitest-environment node
import { afterEach, describe, expect, it, vi } from "vitest";
import {
WORKFLOW_EXTENSION_SCHEMA_VERSION,
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
type TaskDetail,
type WorkflowIr,
} from "@fusion/core";
import { TaskExecutor } from "../executor.js";
describe("workflow verdict-provider extensions", () => {
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
});
function makeExecutor(workflow: WorkflowIr) {
const store = {
on: vi.fn(),
getTaskWorkflowSelection: vi.fn().mockReturnValue({ workflowId: "custom-workflow", stepIds: [] }),
getWorkflowDefinition: vi.fn().mockResolvedValue({ ir: workflow }),
};
return new TaskExecutor(store as any, "/tmp/fusion-verdict-provider-test");
}
it("allows task completion when all provider verdicts pass", async () => {
const workflow: WorkflowIr = { version: "v2", name: "w", columns: [], nodes: [], edges: [] };
const task = { id: "FN-PASS", steps: [] } as unknown as TaskDetail;
getWorkflowExtensionRegistry().register("verdict-plugin", {
extensionId: "quality",
name: "Quality",
kind: "verdict-provider",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
evaluate: vi.fn().mockResolvedValue({ status: "pass", summary: "ok" }),
});
await expect((makeExecutor(workflow) as any).evaluateTaskVerdictProviders(task)).resolves.toEqual({ ok: true });
});
it("blocks task completion when a provider returns a failing verdict", async () => {
const workflow: WorkflowIr = { version: "v2", name: "w", columns: [], nodes: [], edges: [] };
const task = { id: "FN-BLOCK", steps: [] } as unknown as TaskDetail;
getWorkflowExtensionRegistry().register("verdict-plugin", {
extensionId: "quality",
name: "Quality",
kind: "verdict-provider",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
evaluate: vi.fn().mockResolvedValue({
status: "fail",
summary: "quality gate failed",
failureReasons: [{ code: "missing-test", message: "missing regression test" }],
}),
});
await expect((makeExecutor(workflow) as any).evaluateTaskVerdictProviders(task)).resolves.toEqual({
ok: false,
message: "fn_task_done refused (verdict-provider): quality gate failed — missing regression test",
});
});
});

84
packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts Normal file
View File

@@ -0,0 +1,84 @@
// @vitest-environment node
import { afterEach, describe, expect, it, vi } from "vitest";
import {
WORKFLOW_EXTENSION_SCHEMA_VERSION,
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
workflowExtensionRegistryId,
type Task,
type TaskDetail,
type WorkflowIr,
} from "@fusion/core";
import { TaskExecutor } from "../executor.js";
describe("workflow work-engine dispatch", () => {
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
});
it("lets a plugin work engine claim a task from column extension metadata", async () => {
const extensionKey = workflowExtensionRegistryId("engine-plugin", "custom-dispatch");
const task = {
id: "FN-WORK",
column: "in-progress",
title: "plugin work",
description: "plugin work",
} as TaskDetail;
const workflow: WorkflowIr = {
version: "v2",
name: "custom",
columns: [
{ id: "todo", name: "Todo", traits: [] },
{
id: "in-progress",
name: "Running",
traits: [],
extensions: { [extensionKey]: { lane: "custom" } },
},
],
nodes: [],
edges: [],
};
const dispatch = vi.fn().mockResolvedValue({
kind: "claimed",
runId: "plugin-run-1",
message: "claimed by plugin",
});
getWorkflowExtensionRegistry().register("engine-plugin", {
extensionId: "custom-dispatch",
name: "Custom dispatch",
kind: "work-engine",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
dispatch,
});
const store = {
on: vi.fn(),
getTask: vi.fn().mockResolvedValue(task),
getTaskWorkflowSelection: vi.fn().mockReturnValue({ workflowId: "custom-workflow", stepIds: [] }),
getWorkflowDefinition: vi.fn().mockResolvedValue({ ir: workflow }),
logEntry: vi.fn().mockResolvedValue(undefined),
recordRunAuditEvent: vi.fn().mockResolvedValue(undefined),
updateTask: vi.fn().mockResolvedValue(undefined),
};
const executor = new TaskExecutor(store as any, "/tmp/fusion-work-engine-test");
const claimed = await (executor as any).maybeDispatchWorkflowWorkEngine(task as Task);
expect(claimed).toBe(true);
expect(dispatch).toHaveBeenCalledWith(expect.objectContaining({
task,
workflow,
columnId: "in-progress",
metadata: { lane: "custom" },
}));
expect(store.logEntry).toHaveBeenCalledWith("FN-WORK", "claimed by plugin");
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
mutationType: "workflow:work-engine:claimed",
metadata: expect.objectContaining({ extensionId: extensionKey, pluginId: "engine-plugin" }),
}));
expect(store.updateTask).not.toHaveBeenCalled();
});
});

64
packages/engine/src/auto-merge-fact-providers.ts Normal file
View File

@@ -0,0 +1,64 @@
import {
getWorkflowExtensionRegistry,
resolveWorkflowIrForTask,
type AutoMergeFactProviderResult,
type AutoMergeRoute,
type TaskDetail,
type WorkflowIrResolverStore,
} from "@fusion/core";
export interface AutoMergeFactProviderEvaluation {
route?: AutoMergeRoute;
facts: Record<string, unknown>;
reasons: string[];
}
export async function evaluateAutoMergeFactProviders(
store: WorkflowIrResolverStore,
task: TaskDetail,
): Promise<AutoMergeFactProviderEvaluation> {
const workflow = await resolveWorkflowIrForTask(store, task.id);
const evaluation: AutoMergeFactProviderEvaluation = { facts: {}, reasons: [] };
for (const definition of getWorkflowExtensionRegistry().list("merge-fact-provider")) {
const extension = definition.extension;
if (definition.degraded || extension.kind !== "merge-fact-provider" || !extension.collect) continue;
let result: AutoMergeFactProviderResult;
try {
result = await extension.collect({ task, workflow });
} catch (error) {
if (extension.fallback === "degradeToDefault") continue;
const message = error instanceof Error ? error.message : String(error);
return {
route: "blocked",
facts: evaluation.facts,
reasons: [...evaluation.reasons, `fact provider '${definition.id}' failed: ${message}`],
};
}
if (result.facts) {
evaluation.facts[definition.id] = result.facts;
}
if (result.reason) {
evaluation.reasons.push(result.reason);
}
if (result.route) {
evaluation.route = chooseStricterAutoMergeRoute(evaluation.route, result.route);
}
}
return evaluation;
}
function chooseStricterAutoMergeRoute(
current: AutoMergeRoute | undefined,
next: AutoMergeRoute,
): AutoMergeRoute {
const rank: Record<AutoMergeRoute, number> = {
"auto-enqueue": 0,
"workflow-subgraph": 1,
"manual-required": 2,
blocked: 3,
};
if (!current) return next;
return rank[next] > rank[current] ? next : current;
}

149
packages/engine/src/executor.ts
View File

@@ -9,9 +9,9 @@ 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 } from "@fusion/core";
import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, isWorkflowColumnsEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry } from "@fusion/core";
import { mergeEffectiveSettings } from "./effective-settings.js";
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput } from "@fusion/core";
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult } from "@fusion/core";
import {
buildWorkflowObservationFromTask,
buildWorkflowObservation,
@@ -5562,6 +5562,129 @@ export class TaskExecutor {
}
}
private async maybeDispatchWorkflowWorkEngine(task: Task): Promise<boolean> {
let detail: TaskDetail;
let workflow: WorkflowIr;
try {
detail = await this.store.getTask(task.id);
workflow = await resolveWorkflowIrForTask(this.store, task.id);
} catch (error) {
executorLog.warn(`${task.id}: failed to resolve workflow work-engine bindings: ${error instanceof Error ? error.message : String(error)}`);
return false;
}
if (workflow.version !== "v2") return false;
const column = workflow.columns.find((candidate) => candidate.id === detail.column);
const extensionEntries = Object.entries(column?.extensions ?? {});
if (extensionEntries.length === 0) return false;
const registry = getWorkflowExtensionRegistry();
for (const [extensionId, metadata] of extensionEntries) {
const definition = registry.get(extensionId);
const extension = definition?.extension;
if (!definition || definition.degraded || extension?.kind !== "work-engine" || !extension.dispatch) continue;
let result: WorkflowWorkEngineDispatchResult;
try {
result = await extension.dispatch({
task: detail,
workflow,
columnId: detail.column,
metadata,
});
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
executorLog.warn(`${task.id}: workflow work-engine ${extensionId} failed: ${message}`);
if (extension.fallback === "degradeToDefault") continue;
await this.store.logEntry(task.id, `Workflow work engine ${extensionId} failed`, message);
await this.store.updateTask(task.id, {
status: extension.fallback === "parkNeedsAttention" ? "queued" : "failed",
error: message,
});
return true;
}
if (result.kind === "not-claimed") continue;
if (result.kind === "degraded-to-default") {
executorLog.warn(`${task.id}: workflow work-engine ${extensionId} degraded to default: ${result.reason}`);
await this.store.logEntry(task.id, `Workflow work engine ${extensionId} degraded to default`, result.reason);
continue;
}
if (result.kind === "parked") {
await this.store.logEntry(task.id, result.message, result.reason);
await this.store.updateTask(task.id, { status: "queued", error: result.reason });
return true;
}
await this.store.logEntry(
task.id,
result.message ?? `Workflow work engine ${extensionId} claimed execution`,
);
try {
await this.store.recordRunAuditEvent?.({
taskId: task.id,
agentId: "workflow-work-engine",
runId: result.runId ?? generateSyntheticRunId("workflow-work-engine", task.id),
domain: "database",
mutationType: "workflow:work-engine:claimed",
target: task.id,
metadata: {
extensionId,
columnId: detail.column,
pluginId: definition.pluginId,
},
});
} catch (error) {
executorLog.warn(`${task.id}: failed to record workflow work-engine claim audit: ${error instanceof Error ? error.message : String(error)}`);
}
return true;
}
return false;
}
private async evaluateTaskVerdictProviders(
task: TaskDetail,
context: Record<string, unknown> = {},
): Promise<{ ok: true } | { ok: false; message: string }> {
let workflow: WorkflowIr;
try {
workflow = await resolveWorkflowIrForTask(this.store, task.id);
} catch (error) {
executorLog.warn(`${task.id}: failed to resolve workflow for verdict providers: ${error instanceof Error ? error.message : String(error)}`);
return { ok: true };
}
const providers = getWorkflowExtensionRegistry().list("verdict-provider");
for (const definition of providers) {
const extension = definition.extension;
if (definition.degraded || extension.kind !== "verdict-provider" || !extension.evaluate) continue;
try {
const verdict = await extension.evaluate({
task,
workflow,
reworkRound: 0,
metadata: context,
});
if (verdict.status === "pass") continue;
const reasons = verdict.failureReasons?.map((reason) => reason.message).filter(Boolean).join("; ");
return {
ok: false,
message: `fn_task_done refused (verdict-provider): ${verdict.summary}${reasons ? ` — ${reasons}` : ""}`,
};
} catch (error) {
if (extension.fallback === "degradeToDefault") continue;
const message = error instanceof Error ? error.message : String(error);
return {
ok: false,
message: `fn_task_done refused (verdict-provider): provider '${definition.id}' failed — ${message}`,
};
}
}
return { ok: true };
}
async execute(task: Task): Promise<void> {
// Workflow graph interpreter routing (cutover M-C): graph-selected tasks
// are orchestrated by the interpreter. The execute seam re-enters this
@@ -5607,6 +5730,13 @@ export class TaskExecutor {
return;
}
if (await this.maybeDispatchWorkflowWorkEngine(task)) {
executorLog.log(`${task.id}: workflow work engine claimed execution`);
this.executing.delete(task.id);
executingTaskLock.release(task.id);
return;
}
// Column-agent principal alignment (plan U5, R6): the heartbeat-deferral gate
// must consult the EFFECTIVE principal, not blindly `assignedAgentId`. For a
// graph-routed seam the binding context (governing node id + per-run resolver)
@@ -9032,6 +9162,21 @@ export class TaskExecutor {
};
}
const providerVerdict = await this.evaluateTaskVerdictProviders(task, {
summary: params.summary,
source: "fn_task_done",
});
if (!providerVerdict.ok) {
await store.logEntry(taskId, providerVerdict.message, undefined, this.getRunContextFor(task.id));
executorLog.error(`${taskId}: ${providerVerdict.message}`);
return {
content: [{ type: "text" as const, text: providerVerdict.message }],
details: {
error: providerVerdict.message,
},
};
}
const invariantCheck = await this.verifyWorktreeInvariants(task, worktreePath);
if (!invariantCheck.ok) {
const refusalMessage = `fn_task_done refused: ${invariantCheck.reason} — observed=${invariantCheck.observed}, expected=${invariantCheck.expected}`;

21
packages/engine/src/merger.ts
View File

@@ -89,10 +89,12 @@ import {
type PostMergeAuditMode,
type TaskSourceIssue,
type Task,
type TaskDetail,
type AutostashOrphanRecord,
normalizeMergeAdvanceAutoSyncMode,
isMergeRequestContractShadowEnabled,
} from "@fusion/core";
import { evaluateAutoMergeFactProviders } from "./auto-merge-fact-providers.js";
import { resolveMergePolicy, type MergeFileScopeMode } from "./merge-trait.js";
import { describeModel, promptWithFallback } from "./pi.js";
import { accumulateSessionTokenUsage } from "./session-token-usage.js";
@@ -8149,14 +8151,26 @@ export async function aiMergeTask(
}
if (isMergeRequestContractShadowEnabled(settings)) {
const initialState = task.autoMerge === false ? "manual-required" : "queued";
const autoMergeFacts = await evaluateAutoMergeFactProviders(store, task as TaskDetail).catch((error) => ({
route: "blocked" as const,
facts: {},
reasons: [`auto-merge fact provider evaluation failed: ${error instanceof Error ? error.message : String(error)}`],
}));
const providerManualRoute =
autoMergeFacts.route === "manual-required" ||
autoMergeFacts.route === "blocked";
const autoMergeManuallyGated =
task.autoMerge === false ||
(settings.autoMerge === false && task.autoMerge !== true) ||
providerManualRoute;
const initialState = autoMergeManuallyGated ? "manual-required" : "queued";
const existingRecord = store.getMergeRequestRecord(task.id);
const currentState = existingRecord?.state ?? initialState;
if (!existingRecord) {
store.upsertMergeRequestRecord(task.id, { state: initialState });
}
if (task.autoMerge !== false) {
if (!autoMergeManuallyGated) {
if (currentState === "retrying") {
store.transitionMergeRequestState(task.id, "queued");
store.transitionMergeRequestState(task.id, "running");
@@ -8171,6 +8185,9 @@ export async function aiMergeTask(
metadata: {
taskId: task.id,
state: initialState,
autoMergeProviderRoute: autoMergeFacts.route ?? null,
autoMergeProviderReasons: autoMergeFacts.reasons,
autoMergeProviderFacts: autoMergeFacts.facts,
integrationMode: integrationRoot.mode === "reuse-task-worktree" ? "reuse-task-worktree" : "cwd-integration",
},
});

113
packages/engine/src/plugin-runner.ts
View File

@@ -21,6 +21,7 @@ import type {
PluginContext,
PluginSkillContribution,
PluginWorkflowStepContribution,
WorkflowExtensionContribution,
PluginTraitContribution,
WorkflowIr,
PluginPromptContribution,
@@ -36,7 +37,9 @@ import { Type } from "@earendil-works/pi-ai";
import { isAbsolute } from "node:path";
import {
getTraitRegistry,
getWorkflowExtensionRegistry,
resolveWorkflowIrForTask,
workflowExtensionRegistryId,
} from "@fusion/core";
import { createLogger, executorLog } from "./logger.js";
import type { WorkflowCustomNodeRunner } from "./workflow-node-handlers.js";
@@ -54,6 +57,11 @@ import {
unregisterPluginStepParsers,
type PluginStepParserContribution,
} from "./plugin-parser-adapter.js";
import {
degradePluginWorkflowExtensions,
registerPluginWorkflowExtensions,
unregisterPluginWorkflowExtensions,
} from "./plugin-workflow-extension-adapter.js";
// Type for the task store's event data
interface TaskMovedEvent {
@@ -122,6 +130,11 @@ interface CachedWorkflowSteps {
version: number;
}
interface CachedWorkflowExtensions {
extensions: Array<{ pluginId: string; extension: WorkflowExtensionContribution }>;
version: number;
}
interface CachedWorkflowStepTemplates {
templates: Array<{ pluginId: string; template: WorkflowStepTemplate }>;
version: number;
@@ -158,6 +171,7 @@ export class PluginRunner {
private cachedCliProviderContributions: CachedCliProviderContributions | null = null;
private cachedSkills: CachedSkills | null = null;
private cachedWorkflowSteps: CachedWorkflowSteps | null = null;
private cachedWorkflowExtensions: CachedWorkflowExtensions | null = null;
private cachedWorkflowStepTemplates: CachedWorkflowStepTemplates | null = null;
private cachedTraits: CachedTraits | null = null;
private cachedPromptContributions: CachedPromptContributions | null = null;
@@ -170,11 +184,14 @@ export class PluginRunner {
private cliProviderContributionsCacheVersion = 0;
private skillsCacheVersion = 0;
private workflowStepsCacheVersion = 0;
private workflowExtensionsCacheVersion = 0;
private workflowStepTemplatesCacheVersion = 0;
private traitsCacheVersion = 0;
private promptContributionsCacheVersion = 0;
/** Map of pluginId → the registry trait ids it currently has registered. */
private registeredPluginTraitIds = new Map<string, string[]>();
/** Map of pluginId → the workflow extension ids it currently has registered. */
private registeredPluginWorkflowExtensionIds = new Map<string, string[]>();
/** Map of pluginId → the step-parser registry ids it currently has registered
* (U12, KTD-12; mirrors registeredPluginTraitIds). */
private registeredPluginParserIds = new Map<string, string[]>();
@@ -256,6 +273,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -396,6 +414,16 @@ export class PluginRunner {
return this.cachedWorkflowSteps.steps;
}
getPluginWorkflowExtensions(): Array<{ pluginId: string; extension: WorkflowExtensionContribution }> {
if (!this.cachedWorkflowExtensions || this.cachedWorkflowExtensions.version !== this.workflowExtensionsCacheVersion) {
this.cachedWorkflowExtensions = {
extensions: this.options.pluginLoader.getPluginWorkflowExtensions(),
version: this.workflowExtensionsCacheVersion,
};
}
return this.cachedWorkflowExtensions.extensions;
}
/**
* Get all plugin trait contributions with their plugin ids (U8). Aggregated /
* cached / invalidated exactly like workflow steps.
@@ -479,6 +507,76 @@ export class PluginRunner {
}
}
syncPluginWorkflowExtensions(): void {
const registry = getWorkflowExtensionRegistry();
const current = this.getPluginWorkflowExtensions();
const byPlugin = new Map<string, WorkflowExtensionContribution[]>();
for (const { pluginId, extension } of current) {
const list = byPlugin.get(pluginId) ?? [];
list.push(extension);
byPlugin.set(pluginId, list);
}
for (const [pluginId, ids] of [...this.registeredPluginWorkflowExtensionIds.entries()]) {
if (!byPlugin.has(pluginId)) {
unregisterPluginWorkflowExtensions(registry, ids);
this.registeredPluginWorkflowExtensionIds.delete(pluginId);
}
}
for (const [pluginId, contributions] of byPlugin) {
try {
const ids = registerPluginWorkflowExtensions({ registry, pluginId, contributions });
this.registeredPluginWorkflowExtensionIds.set(pluginId, ids);
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
this.log.warn(`Failed to register workflow extensions for plugin '${pluginId}': ${msg}`);
}
}
}
disablePluginWorkflowExtensions(pluginId: string, opts?: { force?: boolean }): {
degraded: string[];
dependents: [];
} {
const registry = getWorkflowExtensionRegistry();
const ids = this.collectPluginWorkflowExtensionIds(pluginId);
if (!opts?.force) {
unregisterPluginWorkflowExtensions(registry, ids);
this.registeredPluginWorkflowExtensionIds.delete(pluginId);
return { degraded: [], dependents: [] };
}
const degraded = degradePluginWorkflowExtensions(registry, ids);
if (degraded.length > 0) {
try {
this.options.taskStore.recordRunAuditEvent({
agentId: "system",
runId: `plugin-workflow-extension-degrade-${pluginId}-${Date.now()}`,
domain: "database",
mutationType: "plugin:workflow-extension-degraded",
target: pluginId,
metadata: {
pluginId,
degradedExtensionIds: degraded,
note: "workflow extension handlers are degraded by fallback policy",
},
});
} catch {
// Audit is best-effort; degradation already applied.
}
}
return { degraded, dependents: [] };
}
private collectPluginWorkflowExtensionIds(pluginId: string): string[] {
const tracked = this.registeredPluginWorkflowExtensionIds.get(pluginId);
if (tracked && tracked.length > 0) return tracked;
return this.getPluginWorkflowExtensions()
.filter((entry) => entry.pluginId === pluginId)
.map((entry) => workflowExtensionRegistryId(pluginId, entry.extension.extensionId));
}
/**
* Register all currently-loaded plugins' step-parser contributions into the
* core StepParserRegistry (plugin-namespaced ids, U12/KTD-12). Mirrors
@@ -807,6 +905,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -829,6 +928,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -856,6 +956,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -883,6 +984,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -909,6 +1011,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -927,6 +1030,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -945,6 +1049,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -963,6 +1068,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -981,6 +1087,7 @@ export class PluginRunner {
this.invalidateCliProviderContributionsCache();
this.invalidateSkillsCache();
this.invalidateWorkflowStepsCache();
this.invalidateWorkflowExtensionsCache();
this.invalidateWorkflowStepTemplatesCache();
this.invalidateTraitsCache();
this.invalidatePromptContributionsCache();
@@ -1210,6 +1317,12 @@ export class PluginRunner {
this.log.log(`Workflow steps cache invalidated (version: ${this.workflowStepsCacheVersion})`);
}
private invalidateWorkflowExtensionsCache(): void {
this.workflowExtensionsCacheVersion++;
this.log.log(`Workflow extensions cache invalidated (version: ${this.workflowExtensionsCacheVersion})`);
this.syncPluginWorkflowExtensions();
}
private invalidateWorkflowStepTemplatesCache(): void {
this.workflowStepTemplatesCacheVersion++;
this.log.log(`Workflow step templates cache invalidated (version: ${this.workflowStepTemplatesCacheVersion})`);

38
packages/engine/src/plugin-workflow-extension-adapter.ts Normal file
View File

@@ -0,0 +1,38 @@
import {
type WorkflowExtensionContribution,
type WorkflowExtensionRegistry,
workflowExtensionRegistryId,
} from "@fusion/core";
export function registerPluginWorkflowExtensions(params: {
registry: WorkflowExtensionRegistry;
pluginId: string;
contributions: WorkflowExtensionContribution[];
}): string[] {
const registered: string[] = [];
for (const contribution of params.contributions) {
const id = workflowExtensionRegistryId(params.pluginId, contribution.extensionId);
params.registry.upsert(params.pluginId, contribution);
registered.push(id);
}
return registered;
}
export function unregisterPluginWorkflowExtensions(
registry: WorkflowExtensionRegistry,
ids: readonly string[],
): string[] {
const removed: string[] = [];
for (const id of ids) {
if (registry.unregister(id)) removed.push(id);
}
return removed;
}
export function degradePluginWorkflowExtensions(
registry: WorkflowExtensionRegistry,
ids: readonly string[],
message = "workflow extension plugin force-disabled",
): string[] {
return registry.degrade(ids, "force-disabled", message);
}

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

@@ -1,5 +1,5 @@
import type { Settings, TaskDetail, TaskStep, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core";
import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled, resolveMaxReworkCycles } from "@fusion/core";
import type { Settings, TaskDetail, TaskStep, WorkflowIr, WorkflowIrEdge, WorkflowIrNode, WorkflowNodeExtensionResult } from "@fusion/core";
import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, getWorkflowExtensionRegistry, isExperimentalFeatureEnabled, resolveMaxReworkCycles } from "@fusion/core";
import {
createDefaultNodeHandlers,
@@ -249,7 +249,7 @@ export class WorkflowGraphExecutor {
runId,
nodeMap,
outgoingMap,
runBranchNode: (node, signal) => this.executeNodeWithRetries(node, task, settings, context, signal),
runBranchNode: (node, signal) => this.executeNodeWithRetries(node, task, settings, context, ir, signal),
shouldTraverseEdge: (edge, source) => this.shouldTraverseEdge(edge, source),
persistence: this.deps.branchPersistence,
semaphore: this.deps.branchSemaphore,
@@ -314,7 +314,7 @@ export class WorkflowGraphExecutor {
steps,
context,
runTemplateNode: (tNode, sig, contextOverride) =>
this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, sig),
this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, ir, sig),
shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src),
persistence: this.deps.stepInstancePersistence,
onReworkReset: this.deps.onReworkReset,
@@ -338,7 +338,7 @@ export class WorkflowGraphExecutor {
return await traverseChildren(node, result);
}
const result = await this.executeNodeWithRetries(node, task, settings, context);
const result = await this.executeNodeWithRetries(node, task, settings, context, ir);
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;
@@ -488,17 +488,79 @@ export class WorkflowGraphExecutor {
throw new WorkflowIrError(`Unsupported edge condition: ${edge.condition}`);
}
private normalizePluginNodeResult(result: WorkflowNodeExtensionResult): WorkflowNodeResult {
if (result.outcome === "success" || result.outcome === "failure") {
return result;
}
return {
outcome: "success",
value: result.value ?? result.outcome.slice("outcome:".length),
contextPatch: result.contextPatch,
};
}
private async executePluginNodeHandler(
node: WorkflowIrNode,
task: TaskDetail,
workflow: WorkflowIr,
context: Record<string, unknown>,
signal?: AbortSignal,
): Promise<WorkflowNodeResult | undefined> {
const extensionIds = Object.keys(node.extensions ?? {});
if (extensionIds.length === 0) return undefined;
const registry = getWorkflowExtensionRegistry();
for (const extensionId of extensionIds) {
const definition = registry.get(extensionId);
const extension = definition?.extension;
if (!definition || definition.degraded || extension?.kind !== "node-handler" || !extension.handle) continue;
if (extension.nodeKind && extension.nodeKind !== node.kind) continue;
try {
const result = await extension.handle({
task,
workflow,
node,
context,
signal,
});
return this.normalizePluginNodeResult(result);
} catch (error) {
if (extension.fallback === "degradeToDefault") {
try {
registry.degrade(
[definition.id],
"runtime-fault",
error instanceof Error ? error.message : String(error),
);
} catch {
// Degradation is best-effort; falling through to the default node
// handler is still the correct fallback for this invocation.
}
continue;
}
return {
outcome: "failure",
value: "plugin-node-handler-error",
contextPatch: {
[`node:${node.id}:error`]: error instanceof Error ? error.message : String(error),
[`node:${node.id}:extensionId`]: extensionId,
},
};
}
}
return undefined;
}
private async executeNodeWithRetries(
node: WorkflowIrNode,
task: TaskDetail,
settings: Pick<Settings, "experimentalFeatures"> | undefined,
context: Record<string, unknown>,
workflow: WorkflowIr,
signal?: AbortSignal,
): Promise<WorkflowNodeResult> {
const handler = this.handlers[node.kind];
if (!handler) {
throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`);
}
// Per-node override: config.maxRetries beats the executor-wide default.
const configured = Number(node.config?.maxRetries);
@@ -511,6 +573,11 @@ export class WorkflowGraphExecutor {
// Fail-fast cancellation: a branch aborted mid-retry stops re-trying.
if (signal?.aborted) return { outcome: "failure", value: "aborted" };
try {
const pluginResult = await this.executePluginNodeHandler(node, task, workflow, context, signal);
if (pluginResult) return pluginResult;
if (!handler) {
throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`);
}
return await handler(node, { task, settings, context, signal });
} catch (error) {
lastError = error;