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

fix(engine): address workflow extension PR feedback

Browse Source
This commit is contained in:
gsxdsm
2026-06-07 21:19:10 -07:00
parent 71822f26db
commit 01523a6581
14 changed files with 417 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
packages/core/src/__tests__/transition-parity.test.ts
View File

@@ -185,6 +185,68 @@ describe("transition-parity — store flag-ON scenarios", () => {
expect(seen).toEqual([{ actorKind: "human", source: "board-drag" }]);
});
it("move-policy extensions run before the task lock is held", async () => {
getWorkflowExtensionRegistry().register("policy-plugin", {
extensionId: "preflight-update",
name: "Preflight update",
kind: "move-policy",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
evaluate: async ({ task }) => {
await store.updateTask(task.id, { summary: "policy evaluated outside lock" });
return { allowed: true };
},
});
const task = await seedInColumn("triage");
const moved = await store.moveTask(task.id, "todo", { moveSource: "user" });
expect(moved.column).toBe("todo");
expect((await store.getTask(task.id))?.summary).toBe("policy evaluated outside lock");
});
it("move-policy extensions cannot veto user hard-cancel moves", async () => {
const task = await seedInColumn("in-progress");
getWorkflowExtensionRegistry().register("policy-plugin", {
extensionId: "block-todo",
name: "Block todo",
kind: "move-policy",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
evaluate: ({ toColumn }) => {
if (toColumn === "todo") return { allowed: false, reason: "todo blocked", message: "Todo is blocked" };
return { allowed: true };
},
});
const moved = await store.moveTask(task.id, "todo", { moveSource: "user" });
expect(moved.column).toBe("todo");
expect(moved.userPaused).toBe(true);
});
it("degrades faulting move-policy extensions when fallback is degradeToDefault", async () => {
getWorkflowExtensionRegistry().register("policy-plugin", {
extensionId: "faulty",
name: "Faulty",
kind: "move-policy",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "degradeToDefault",
evaluate: () => {
throw new Error("boom");
},
});
const task = await seedInColumn("triage");
const moved = await store.moveTask(task.id, "todo", { moveSource: "user" });
expect(moved.column).toBe("todo");
expect(getWorkflowExtensionRegistry().get("plugin:policy-plugin:faulty")?.degraded).toMatchObject({
reason: "runtime-fault",
message: "boom",
});
});
it("handoffToReview maps skipMergeBlocker onto bypassGuards and enqueues exactly once", async () => {
const task = await seedInColumn("in-progress");
await store.handoffToReview(task.id, {

9
packages/core/src/__tests__/workflow-extension-registry.test.ts
View File

@@ -5,11 +5,14 @@ import {
} from "../workflow-extension-registry.js";
import type { WorkflowExtensionContribution } from "../workflow-extension-types.js";
function extension(extensionId = "move-policy"): WorkflowExtensionContribution {
function extension(
extensionId = "move-policy",
kind: WorkflowExtensionContribution["kind"] = "move-policy",
): WorkflowExtensionContribution {
return {
extensionId,
name: "Move Policy",
kind: "move-policy",
kind,
schemaVersion: 1,
fallback: "degradeToDefault",
};
@@ -37,7 +40,7 @@ describe("WorkflowExtensionRegistry", () => {
it("unregisters all extensions for a plugin", () => {
const registry = new WorkflowExtensionRegistry();
registry.register("plugin-a", extension("move-policy"));
registry.register("plugin-a", extension("work-engine"));
registry.register("plugin-a", extension("work-engine", "work-engine"));
registry.register("plugin-b", extension("move-policy"));
expect(registry.unregisterPlugin("plugin-a")).toEqual([

59
packages/core/src/__tests__/workflow-ir-extension-metadata.test.ts
View File

@@ -1,8 +1,13 @@
import { describe, expect, it } from "vitest";
import { afterEach, describe, expect, it } from "vitest";
import {
downgradeIrToV1IfPure,
parseWorkflowIr,
} from "../workflow-ir.js";
import {
__resetWorkflowExtensionRegistryForTests,
getWorkflowExtensionRegistry,
} from "../workflow-extension-registry.js";
import { WORKFLOW_EXTENSION_SCHEMA_VERSION } from "../workflow-extension-types.js";
import type { WorkflowIrV2 } from "../workflow-ir-types.js";
function ir(overrides: Partial<WorkflowIrV2> = {}): WorkflowIrV2 {
@@ -20,6 +25,10 @@ function ir(overrides: Partial<WorkflowIrV2> = {}): WorkflowIrV2 {
}
describe("workflow IR extension metadata", () => {
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
});
it("accepts plugin-namespaced column and node extension metadata", () => {
const parsed = parseWorkflowIr(ir({
columns: [
@@ -82,6 +91,54 @@ describe("workflow IR extension metadata", () => {
).toThrow(/metadata must be an object/);
});
it("requires enum extension fields to declare enumValues", () => {
getWorkflowExtensionRegistry().register("workflow-pack", {
extensionId: "role",
name: "Role",
kind: "column-metadata",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
configSchema: { fields: [{ key: "role", type: "enum" }] },
});
expect(() =>
parseWorkflowIr(ir({
columns: [
{
id: "todo",
name: "todo",
traits: [],
extensions: { "plugin:workflow-pack:role": { role: "lead" } },
},
],
})),
).toThrow(/field 'role' is enum but has no enumValues defined/);
});
it("rejects enum extension field values outside enumValues", () => {
getWorkflowExtensionRegistry().register("workflow-pack", {
extensionId: "role",
name: "Role",
kind: "column-metadata",
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
fallback: "failClosed",
configSchema: { fields: [{ key: "role", type: "enum", enumValues: ["lead", "executor"] }] },
});
expect(() =>
parseWorkflowIr(ir({
columns: [
{
id: "todo",
name: "todo",
traits: [],
extensions: { "plugin:workflow-pack:role": { role: "reviewer" } },
},
],
})),
).toThrow(/must be one of: lead, executor/);
});
it("keeps v2 when otherwise-pure workflows carry extension metadata", () => {
const parsed = parseWorkflowIr(ir({
columns: [

119
packages/core/src/store.ts
View File

@@ -1218,8 +1218,14 @@ interface MoveTaskInternalOptions {
ownerAgentId?: string | null;
evidence?: HandoffToReviewOptions["evidence"];
now?: string;
movePolicyPreflight?: {
fromColumn: string;
toColumn: string;
};
}
const WORKFLOW_MOVE_POLICY_TIMEOUT_MS = 5000;
export class TaskStore extends EventEmitter<TaskStoreEvents> {
private static readonly ACTIVE_TASKS_WHERE = '"deletedAt" IS NULL';
/** U6: sentinel effective-workflow id for default-workflow (null-selection)
@@ -6279,7 +6285,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
// ColumnId admits workflow-defined custom column ids (KTD-1). Both paths
// runtime-validate: flag-ON against the task's resolved workflow, flag-OFF
// via the VALID_TRANSITIONS lookup (non-legacy ids reject as before).
return this.withTaskLock(id, () => this.moveTaskInternal(id, toColumn, options, { fromHandoff: false }));
const movePolicyPreflight = await this.prepareWorkflowMovePolicyPreflight(id, toColumn, options, { fromHandoff: false });
return this.withTaskLock(id, () => this.moveTaskInternal(id, toColumn, options, { fromHandoff: false, movePolicyPreflight }));
}
async handoffToReview(taskId: string, opts: HandoffToReviewOptions): Promise<Task> {
@@ -6346,6 +6353,65 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return { kind: "engine" };
}
private resolveWorkflowBypassGuards(
moveSource: NonNullable<MoveTaskOptions["moveSource"]>,
options?: MoveTaskOptions,
): boolean {
return options?.recoveryRehome === true ||
(options?.bypassGuards ??
(moveSource === "engine" || moveSource === "scheduler" || options?.skipMergeBlocker === true));
}
private shouldSkipWorkflowMovePolicies(params: {
fromColumn: string;
toColumn: string;
moveSource: NonNullable<MoveTaskOptions["moveSource"]>;
bypassGuards: boolean;
options?: MoveTaskOptions;
}): boolean {
if (params.bypassGuards) return true;
if (params.options?.recoveryRehome === true) return true;
return params.moveSource === "user" && params.fromColumn === "in-progress" && params.toColumn === "todo";
}
private async prepareWorkflowMovePolicyPreflight(
id: string,
toColumn: ColumnId,
options: MoveTaskOptions | undefined,
internal: MoveTaskInternalOptions,
): Promise<MoveTaskInternalOptions["movePolicyPreflight"]> {
const task = await this.readTaskForMove(id);
const moveSource = options?.moveSource ?? "engine";
const mergedSettingsForMove = await this.getSettingsFast();
if (!isWorkflowColumnsEnabled(mergedSettingsForMove)) return undefined;
if (task.column === toColumn) return undefined;
const workflowIr = this.resolveTaskWorkflowIrSync(id);
const bypassGuards = this.resolveWorkflowBypassGuards(moveSource, options);
const fromColumn = task.column;
if (this.shouldSkipWorkflowMovePolicies({ fromColumn, toColumn, moveSource, bypassGuards, options })) {
return undefined;
}
const recoveryToLegacy =
options?.recoveryRehome === true && (COLUMNS as readonly string[]).includes(toColumn);
if (!workflowHasColumn(workflowIr, toColumn) && !recoveryToLegacy) return undefined;
const allowed = resolveAllowedColumns(workflowIr, fromColumn);
if (options?.recoveryRehome !== true && !allowed.includes(toColumn)) return undefined;
await this.evaluateWorkflowMovePolicies({
task,
workflow: workflowIr,
fromColumn,
toColumn,
actor: this.resolveWorkflowMoveActor(moveSource, internal, options),
source: options?.workflowMoveSource ?? moveSource,
metadata: options?.workflowMoveMetadata,
});
return { fromColumn, toColumn };
}
private async evaluateWorkflowMovePolicies(input: WorkflowMovePolicyInput): Promise<void> {
const policies = getWorkflowExtensionRegistry().list("move-policy");
for (const definition of policies) {
@@ -6354,7 +6420,20 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
let decision: Awaited<ReturnType<NonNullable<typeof extension.evaluate>>>;
try {
decision = await extension.evaluate(input);
decision = await new Promise<Awaited<ReturnType<NonNullable<typeof extension.evaluate>>>>((resolve, reject) => {
const timer = setTimeout(() => {
reject(new Error(`timed out after ${WORKFLOW_MOVE_POLICY_TIMEOUT_MS}ms`));
}, WORKFLOW_MOVE_POLICY_TIMEOUT_MS);
Promise.resolve(extension.evaluate?.(input))
.then((value) => {
clearTimeout(timer);
resolve(value as Awaited<ReturnType<NonNullable<typeof extension.evaluate>>>);
})
.catch((error) => {
clearTimeout(timer);
reject(error);
});
});
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
storeLog.warn("Workflow move-policy extension faulted", {
@@ -6364,7 +6443,10 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
fallback: extension.fallback,
error: message,
});
if (extension.fallback === "degradeToDefault") continue;
if (extension.fallback === "degradeToDefault") {
getWorkflowExtensionRegistry().degrade([definition.id], "runtime-fault", message);
continue;
}
throw new TransitionRejectionError(
makeTransitionRejection(
"guard-rejected",
@@ -6417,10 +6499,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
// call sites map onto it. Capacity (KTD-10) is NEVER bypassed by this — the
// capacity check is not a guard (U6 fills the enforcement; U4 leaves a
// pass-through slot). An explicit option value wins; otherwise derive it.
const bypassGuards =
options?.recoveryRehome === true ||
(options?.bypassGuards ??
(moveSource === "engine" || moveSource === "scheduler" || options?.skipMergeBlocker === true));
const bypassGuards = this.resolveWorkflowBypassGuards(moveSource, options);
const workflowIr: WorkflowIr | undefined = useWorkflow
? this.resolveTaskWorkflowIrSync(id)
: undefined;
@@ -6526,15 +6605,29 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
`Valid targets: ${allowed.join(", ") || "none"}`,
);
}
await this.evaluateWorkflowMovePolicies({
task,
workflow: workflowIr,
const skipWorkflowMovePolicies = this.shouldSkipWorkflowMovePolicies({
fromColumn,
toColumn,
actor: this.resolveWorkflowMoveActor(moveSource, internal, options),
source: options?.workflowMoveSource ?? moveSource,
metadata: options?.workflowMoveMetadata,
moveSource,
bypassGuards,
options,
});
if (!skipWorkflowMovePolicies) {
if (
internal.movePolicyPreflight?.fromColumn !== fromColumn ||
internal.movePolicyPreflight?.toColumn !== toColumn
) {
throw new TransitionRejectionError(
makeTransitionRejection(
"guard-rejected",
"transition.rejected.workflowMovePolicy",
true,
"Workflow move policy preflight is stale; retry the move",
),
`Cannot move ${id} to '${toColumn}': workflow move policy preflight is stale`,
);
}
}
// 3. Sync trait guards (in-lock). Skipped entirely when bypassGuards
// (engine/recovery moves, KTD-9). The default workflow's merge-blocker
// trait reads the same getTaskMergeBlocker.

12
packages/core/src/workflow-extension-registry.ts
View File

@@ -24,7 +24,7 @@ export interface WorkflowExtensionDefinition {
pluginId: string;
extension: WorkflowExtensionContribution;
degraded?: {
reason: "force-disabled" | "plugin-unloaded";
reason: "force-disabled" | "plugin-unloaded" | "runtime-fault";
message: string;
};
}
@@ -61,6 +61,16 @@ export class WorkflowExtensionRegistry {
return definition;
}
upsert(pluginId: string, extension: WorkflowExtensionContribution): WorkflowExtensionDefinition {
const id = workflowExtensionRegistryId(pluginId, extension.extensionId);
const existing = this.definitions.get(id);
if (existing) {
existing.extension = extension;
return existing;
}
return this.register(pluginId, extension);
}
unregister(id: string): boolean {
return this.definitions.delete(id);
}

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

@@ -980,7 +980,12 @@ function validateExtensionFieldValue(
);
}
const enumValue: string = value;
if (field.enumValues && !field.enumValues.includes(enumValue)) {
if (!field.enumValues || field.enumValues.length === 0) {
throw new WorkflowIrError(
`${owner} extension '${key}' field '${field.key}' is enum but has no enumValues defined`,
);
}
if (!field.enumValues.includes(enumValue)) {
throw new WorkflowIrError(
`${owner} extension '${key}' field '${field.key}' must be one of: ${field.enumValues.join(", ")}`,
);

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", () => {
});
});

35
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";
@@ -154,6 +161,7 @@ describe("PluginRunner", () => {
});
afterEach(() => {
__resetWorkflowExtensionRegistryForTests();
vi.clearAllMocks();
});
@@ -901,6 +909,31 @@ describe("PluginRunner", () => {
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: [] } },

77
packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts
View File

@@ -65,4 +65,81 @@ describe("workflow node-handler extensions", () => {
}));
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:custom-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:custom-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",
});
});
});

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

@@ -8159,14 +8159,18 @@ export async function aiMergeTask(
const providerManualRoute =
autoMergeFacts.route === "manual-required" ||
autoMergeFacts.route === "blocked";
const initialState = task.autoMerge === false || providerManualRoute ? "manual-required" : "queued";
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");

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

@@ -39,6 +39,7 @@ import {
getTraitRegistry,
getWorkflowExtensionRegistry,
resolveWorkflowIrForTask,
workflowExtensionRegistryId,
} from "@fusion/core";
import { createLogger, executorLog } from "./logger.js";
import type { WorkflowCustomNodeRunner } from "./workflow-node-handlers.js";
@@ -415,13 +416,8 @@ export class PluginRunner {
getPluginWorkflowExtensions(): Array<{ pluginId: string; extension: WorkflowExtensionContribution }> {
if (!this.cachedWorkflowExtensions || this.cachedWorkflowExtensions.version !== this.workflowExtensionsCacheVersion) {
const loader = this.options.pluginLoader as unknown as {
getPluginWorkflowExtensions?: () => Array<{ pluginId: string; extension: WorkflowExtensionContribution }>;
};
this.cachedWorkflowExtensions = {
extensions: typeof loader.getPluginWorkflowExtensions === "function"
? loader.getPluginWorkflowExtensions()
: [],
extensions: this.options.pluginLoader.getPluginWorkflowExtensions(),
version: this.workflowExtensionsCacheVersion,
};
}
@@ -531,8 +527,6 @@ export class PluginRunner {
for (const [pluginId, contributions] of byPlugin) {
try {
const previous = this.registeredPluginWorkflowExtensionIds.get(pluginId);
if (previous) unregisterPluginWorkflowExtensions(registry, previous);
const ids = registerPluginWorkflowExtensions({ registry, pluginId, contributions });
this.registeredPluginWorkflowExtensionIds.set(pluginId, ids);
} catch (err) {
@@ -580,7 +574,7 @@ export class PluginRunner {
if (tracked && tracked.length > 0) return tracked;
return this.getPluginWorkflowExtensions()
.filter((entry) => entry.pluginId === pluginId)
.map((entry) => `plugin:${pluginId}:${entry.extension.extensionId}`);
.map((entry) => workflowExtensionRegistryId(pluginId, entry.extension.extensionId));
}
/**

4
packages/engine/src/plugin-workflow-extension-adapter.ts
View File

@@ -12,9 +12,7 @@ export function registerPluginWorkflowExtensions(params: {
const registered: string[] = [];
for (const contribution of params.contributions) {
const id = workflowExtensionRegistryId(params.pluginId, contribution.extensionId);
if (!params.registry.get(id)) {
params.registry.register(params.pluginId, contribution);
}
params.registry.upsert(params.pluginId, contribution);
registered.push(id);
}
return registered;

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

@@ -494,7 +494,7 @@ export class WorkflowGraphExecutor {
}
return {
outcome: "success",
value: result.outcome.slice("outcome:".length),
value: result.value ?? result.outcome.slice("outcome:".length),
contextPatch: result.contextPatch,
};
}
@@ -525,7 +525,19 @@ export class WorkflowGraphExecutor {
});
return this.normalizePluginNodeResult(result);
} catch (error) {
if (extension.fallback === "degradeToDefault") continue;
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",