From 01523a65812efe43dbf3bce1a07a000a3ffe22d4 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 7 Jun 2026 21:19:10 -0700 Subject: [PATCH] fix(engine): address workflow extension PR feedback --- .changeset/workflow-extension-plugins.md | 2 +- .../src/__tests__/transition-parity.test.ts | 62 +++++++++ .../workflow-extension-registry.test.ts | 9 +- .../workflow-ir-extension-metadata.test.ts | 59 ++++++++- packages/core/src/store.ts | 119 ++++++++++++++++-- .../core/src/workflow-extension-registry.ts | 12 +- packages/core/src/workflow-ir.ts | 7 +- .../__tests__/merger-merge-lifecycle.test.ts | 33 ++++- .../src/__tests__/plugin-runner.test.ts | 35 +++++- .../workflow-node-handler-extensions.test.ts | 77 ++++++++++++ packages/engine/src/merger.ts | 8 +- packages/engine/src/plugin-runner.ts | 12 +- .../src/plugin-workflow-extension-adapter.ts | 4 +- .../engine/src/workflow-graph-executor.ts | 16 ++- 14 files changed, 417 insertions(+), 38 deletions(-) diff --git a/.changeset/workflow-extension-plugins.md b/.changeset/workflow-extension-plugins.md index 6abdeaf9df..1a66dc431c 100644 --- a/.changeset/workflow-extension-plugins.md +++ b/.changeset/workflow-extension-plugins.md @@ -1,5 +1,5 @@ --- -"@runfusion/fusion": patch +"@runfusion/fusion": minor --- Add workflow extension plugin contracts for move policies, work engines, node handlers, task verdict providers, auto-merge facts, and shared board action services. diff --git a/packages/core/src/__tests__/transition-parity.test.ts b/packages/core/src/__tests__/transition-parity.test.ts index 599ec2944e..828e0992df 100644 --- a/packages/core/src/__tests__/transition-parity.test.ts +++ b/packages/core/src/__tests__/transition-parity.test.ts @@ -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, { diff --git a/packages/core/src/__tests__/workflow-extension-registry.test.ts b/packages/core/src/__tests__/workflow-extension-registry.test.ts index 07079a5990..08d38d2955 100644 --- a/packages/core/src/__tests__/workflow-extension-registry.test.ts +++ b/packages/core/src/__tests__/workflow-extension-registry.test.ts @@ -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([ diff --git a/packages/core/src/__tests__/workflow-ir-extension-metadata.test.ts b/packages/core/src/__tests__/workflow-ir-extension-metadata.test.ts index 7dff28685f..651de9caa4 100644 --- a/packages/core/src/__tests__/workflow-ir-extension-metadata.test.ts +++ b/packages/core/src/__tests__/workflow-ir-extension-metadata.test.ts @@ -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 { @@ -20,6 +25,10 @@ function ir(overrides: Partial = {}): 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: [ diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 855ffb988a..6215ae902b 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -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 { 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 { // 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 { @@ -6346,6 +6353,65 @@ export class TaskStore extends EventEmitter { return { kind: "engine" }; } + private resolveWorkflowBypassGuards( + moveSource: NonNullable, + 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; + 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 { + 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 { const policies = getWorkflowExtensionRegistry().list("move-policy"); for (const definition of policies) { @@ -6354,7 +6420,20 @@ export class TaskStore extends EventEmitter { let decision: Awaited>>; try { - decision = await extension.evaluate(input); + decision = await new Promise>>>((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>>); + }) + .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 { 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 { // 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 { `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. diff --git a/packages/core/src/workflow-extension-registry.ts b/packages/core/src/workflow-extension-registry.ts index d0fba1909d..5937014bb0 100644 --- a/packages/core/src/workflow-extension-registry.ts +++ b/packages/core/src/workflow-extension-registry.ts @@ -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); } diff --git a/packages/core/src/workflow-ir.ts b/packages/core/src/workflow-ir.ts index 7812b5d481..a0a7f5b7cf 100644 --- a/packages/core/src/workflow-ir.ts +++ b/packages/core/src/workflow-ir.ts @@ -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(", ")}`, ); diff --git a/packages/engine/src/__tests__/merger-merge-lifecycle.test.ts b/packages/engine/src/__tests__/merger-merge-lifecycle.test.ts index 68bc763aa1..ef47b9b0b1 100644 --- a/packages/engine/src/__tests__/merger-merge-lifecycle.test.ts +++ b/packages/engine/src/__tests__/merger-merge-lifecycle.test.ts @@ -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).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).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", () => { }); }); - diff --git a/packages/engine/src/__tests__/plugin-runner.test.ts b/packages/engine/src/__tests__/plugin-runner.test.ts index 65244dfd5f..2adceb964c 100644 --- a/packages/engine/src/__tests__/plugin-runner.test.ts +++ b/packages/engine/src/__tests__/plugin-runner.test.ts @@ -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: [] } }, diff --git a/packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts b/packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts index d0c97c7110..b2c0784460 100644 --- a/packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts +++ b/packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts @@ -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", + }); + }); }); diff --git a/packages/engine/src/merger.ts b/packages/engine/src/merger.ts index 23ae662b67..7963347be2 100644 --- a/packages/engine/src/merger.ts +++ b/packages/engine/src/merger.ts @@ -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"); diff --git a/packages/engine/src/plugin-runner.ts b/packages/engine/src/plugin-runner.ts index 95ef7ddd29..a229a6c6bd 100644 --- a/packages/engine/src/plugin-runner.ts +++ b/packages/engine/src/plugin-runner.ts @@ -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)); } /** diff --git a/packages/engine/src/plugin-workflow-extension-adapter.ts b/packages/engine/src/plugin-workflow-extension-adapter.ts index 7663adfd40..2c85482ac3 100644 --- a/packages/engine/src/plugin-workflow-extension-adapter.ts +++ b/packages/engine/src/plugin-workflow-extension-adapter.ts @@ -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; diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index aeb695a90d..546dc2c043 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -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",