feat(engine): Route manual runs to the engine 2.0 data plane (#36830)

packages/@n8n/node-engine-compatibility/src/__tests__/acceptance-fixtures.ts

@@ -1,5 +1,6 @@
 import type {
 	createDataSource,
+	ExecutionMode,
 	StartExecutionResult,
 	TriggerOutputs,
 	WorkflowGraph,
@@ -93,7 +94,11 @@ export function setWorkflow(assignments: Assignment[]) {
 type EngineDataSource = ReturnType<typeof createDataSource>;
 
 export function makeRunWorkflow(getDataSource: () => EngineDataSource) {
-	return async function runWorkflow(graph: WorkflowGraph, triggerOutputs: TriggerOutputs | null) {
+	return async function runWorkflow(
+		graph: WorkflowGraph,
+		triggerOutputs: TriggerOutputs | null,
+		mode?: ExecutionMode,
+	) {
 		const dataSource = getDataSource();
 
 		let done!: () => void;
@@ -127,7 +132,7 @@ export function makeRunWorkflow(getDataSource: () => EngineDataSource) {
 		const response = await request(runtime.app)
 			.post('/api/workflow-executions')
 			.set('Authorization', `Bearer ${mintIdentityToken(authSecret, caller)}`)
-			.send({ workflowId: 'wf-m1', graph, triggerOutputs })
+			.send({ workflowId: 'wf-m1', graph, triggerOutputs, mode })
 			.expect(201);
 		const { executionId } = response.body as StartExecutionResult;
 

packages/@n8n/node-engine-compatibility/src/__tests__/m1-acceptance.integration.test.ts

@@ -76,6 +76,35 @@ describe('M1 acceptance (integration)', () => {
 		expect(execution.status).toBe('completed');
 	});
 
+	it('runs a manual execution started with the default trigger payload', async () => {
+		const graph = converter.convert(
+			v1Workflow(
+				[
+					TRIGGER,
+					setNode('node-a', 'A', [{ name: 'a', value: 'from-a', type: 'string' }]),
+					setNode('node-b', 'B', [{ name: 'b', value: '={{ $json.a }}-b', type: 'string' }]),
+				],
+				{
+					'When clicking Execute': mainTo('A'),
+					A: mainTo('B'),
+				},
+			),
+		);
+
+		// what the control plane sends for a manual run with no trigger data
+		const { execution, steps, byNode } = await runWorkflow(graph, [[{ json: {} }]], 'manual');
+
+		expect(steps).toHaveLength(3);
+		for (const nodeId of ['trigger', 'node-a', 'node-b']) {
+			expect(byNode(nodeId)?.status).toBe('completed');
+		}
+		expect(byNode('node-b')?.outputs).toEqual([
+			[expect.objectContaining({ json: { b: 'from-a-b' } })],
+		]);
+		expect(execution.mode).toBe('manual');
+		expect(execution.status).toBe('completed');
+	});
+
 	it('processes an n-item input into an n-item output, serially and in order', async () => {
 		const graph = converter.convert(
 			v1Workflow(

packages/@n8n/node-engine-compatibility/src/index.ts

@@ -1,5 +1,6 @@
 export { V1WorkflowConverter } from './v1-workflow-converter';
 export { V1StepExecutor } from './v1-step-executor';
 export { createEngineStepDataLoader } from './engine-step-data-loader';
+export { toStepOutputs } from './io';
 export { UnsupportedTriggerError, UnsupportedWorkflowError } from './errors';
 export type { StepData, StepDataLoader, V1StepExecutorDeps } from './types';

packages/cli/src/__tests__/workflow-runner.test.ts

@@ -38,6 +38,7 @@ import { ExecutionNotFoundError } from '@/errors/execution-not-found-error';
 import * as ExecutionLifecycleHooks from '@/execution-lifecycle/execution-lifecycle-hooks';
 import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks';
 import { ManualExecutionService } from '@/manual-execution.service';
+import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
 import { OwnershipService } from '@/services/ownership.service';
 import { Telemetry } from '@/telemetry';
 import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
@@ -270,6 +271,57 @@ describe('run', () => {
 		expect(addSpy).toHaveBeenCalledWith(data, existingExecution);
 	});
 
+	describe('engine 2.0 dispatch', () => {
+		it('hands the run to the dispatcher without registering a control-plane execution', async () => {
+			// ARRANGE
+			const dispatcher = Container.get(EngineV2Dispatcher);
+			vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(true);
+			const startSpy = vi.spyOn(dispatcher, 'start').mockResolvedValueOnce('dp-uuid');
+			const addSpy = vi.spyOn(Container.get(ActiveExecutions), 'add');
+
+			const data = mock<IWorkflowExecutionDataProcess>();
+
+			// ACT
+			const executionId = await runner.run(data);
+
+			// ASSERT
+			expect(executionId).toBe('dp-uuid');
+			expect(startSpy).toHaveBeenCalledWith(data);
+			expect(addSpy).not.toHaveBeenCalled();
+		});
+
+		it('leaves the v1 path alone when the run does not route to engine 2.0', async () => {
+			// ARRANGE
+			const dispatcher = Container.get(EngineV2Dispatcher);
+			vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(false);
+			const startSpy = vi.spyOn(dispatcher, 'start');
+			const activeExecutions = Container.get(ActiveExecutions);
+			const addSpy = vi.spyOn(activeExecutions, 'add').mockResolvedValue('1');
+			vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValueOnce();
+			vi.spyOn(Container.get(CredentialsPermissionChecker), 'check').mockResolvedValueOnce();
+			vi.spyOn(WorkflowExecute.prototype, 'run').mockReturnValueOnce(
+				new PCancelable(() => mock<IRun>()),
+			);
+
+			const data = mock<IWorkflowExecutionDataProcess>({
+				triggerToStartFrom: undefined,
+				workflowData: { nodes: [], staticData: {} },
+				executionData: undefined,
+				startNodes: undefined,
+				destinationNode: undefined,
+				runData: undefined,
+			});
+
+			// ACT
+			const executionId = await runner.run(data);
+
+			// ASSERT
+			expect(executionId).toBe('1');
+			expect(startSpy).not.toHaveBeenCalled();
+			expect(addSpy).toHaveBeenCalledWith(data, undefined);
+		});
+	});
+
 	it('run partial execution with additional data', async () => {
 		// ARRANGE
 		const activeExecutions = Container.get(ActiveExecutions);

packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts

@@ -17,6 +17,14 @@ describe('EngineDataPlaneProxyService', () => {
 		proxy = new EngineDataPlaneProxyService();
 	});
 
+	it('is unavailable until a provider registers', () => {
+		expect(proxy.isAvailable()).toBe(false);
+
+		proxy.registerProvider(mock<EngineDataPlaneProvider>());
+
+		expect(proxy.isAvailable()).toBe(true);
+	});
+
 	it('explains how to enable the engine when no provider is registered', async () => {
 		await expect(proxy.startExecution(request)).rejects.toThrow(UserError);
 		await expect(proxy.startExecution(request)).rejects.toThrow('N8N_ENABLED_MODULES');

packages/cli/src/services/__tests__/engine-v2-dispatcher.service.test.ts

@@ -0,0 +1,321 @@
+import { UnsupportedTriggerError } from '@n8n/node-engine-compatibility';
+import type {
+	INode,
+	IPinData,
+	IRunData,
+	ITaskData,
+	ITaskDataConnections,
+	IWorkflowBase,
+	IWorkflowExecutionDataProcess,
+	StartNodeData,
+	WorkflowExecuteMode,
+} from 'n8n-workflow';
+import { NodeConnectionTypes, UserError } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import type { CredentialsPermissionChecker } from '@/executions/pre-execution-checks';
+import type { ResumableExecution } from '@/interfaces';
+import type { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
+import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
+
+const node = (id: string, name: string, type: string): INode => ({
+	id,
+	name,
+	type,
+	typeVersion: 1,
+	position: [0, 0],
+	parameters: {},
+});
+
+const MANUAL_TRIGGER = node('trigger-id', 'When clicking Execute', 'n8n-nodes-base.manualTrigger');
+const SET_NODE = node('set-id', 'Edit Fields', 'n8n-nodes-base.set');
+
+function workflow(overrides: Partial<IWorkflowBase> = {}): IWorkflowBase {
+	return {
+		id: 'wf-1',
+		name: 'My workflow',
+		active: false,
+		isArchived: false,
+		activeVersionId: null,
+		createdAt: new Date(),
+		updatedAt: new Date(),
+		nodes: [MANUAL_TRIGGER, SET_NODE],
+		connections: {
+			[MANUAL_TRIGGER.name]: {
+				main: [[{ node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 }]],
+			},
+		},
+		settings: { engineType: 'v2' },
+		...overrides,
+	};
+}
+
+const taskData = (main: ITaskDataConnections['main']): ITaskData => ({
+	startTime: 0,
+	executionIndex: 0,
+	executionTime: 0,
+	source: [],
+	data: { main },
+});
+
+function runData(
+	overrides: Partial<IWorkflowExecutionDataProcess> = {},
+): IWorkflowExecutionDataProcess {
+	return {
+		executionMode: 'manual',
+		workflowData: workflow(),
+		triggerToStartFrom: { name: MANUAL_TRIGGER.name },
+		...overrides,
+	};
+}
+
+describe('EngineV2Dispatcher', () => {
+	const proxy = mock<EngineDataPlaneProxyService>();
+	const credentialsPermissionChecker = mock<CredentialsPermissionChecker>();
+
+	let dispatcher: EngineV2Dispatcher;
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		proxy.isAvailable.mockReturnValue(true);
+		proxy.startExecution.mockResolvedValue({ executionId: 'dp-uuid' });
+		dispatcher = new EngineV2Dispatcher(proxy, credentialsPermissionChecker);
+	});
+
+	describe('routesToEngineV2', () => {
+		it('routes a manual run of a workflow that opted into engine 2.0', () => {
+			expect(dispatcher.routesToEngineV2(runData())).toBe(true);
+		});
+
+		it.each([
+			{ name: 'no engineType', settings: {} },
+			{ name: 'engineType v1', settings: { engineType: 'v1' as const } },
+		])('does not route a workflow with $name', ({ settings }) => {
+			const data = runData({ workflowData: workflow({ settings }) });
+
+			expect(dispatcher.routesToEngineV2(data)).toBe(false);
+		});
+
+		it.each<WorkflowExecuteMode>(['webhook', 'trigger', 'retry', 'chat', 'evaluation'])(
+			'does not route a %s run',
+			(executionMode) => {
+				expect(dispatcher.routesToEngineV2(runData({ executionMode }))).toBe(false);
+			},
+		);
+
+		it('does not route a resumed execution', () => {
+			const existingExecution = mock<ResumableExecution>({ executionId: '42' });
+
+			expect(dispatcher.routesToEngineV2(runData(), existingExecution)).toBe(false);
+		});
+	});
+
+	describe('start', () => {
+		it('returns the data plane execution id', async () => {
+			await expect(dispatcher.start(runData())).resolves.toBe('dp-uuid');
+
+			expect(proxy.startExecution).toHaveBeenCalledWith(
+				expect.objectContaining({ workflowId: 'wf-1', mode: 'manual' }),
+			);
+		});
+
+		it('converts the workflow to a graph', async () => {
+			await dispatcher.start(runData());
+
+			const { graph } = proxy.startExecution.mock.calls[0][0];
+			expect(graph.nodes).toEqual([
+				{ id: MANUAL_TRIGGER.id, name: MANUAL_TRIGGER.name, type: 'trigger' },
+				expect.objectContaining({ id: SET_NODE.id, type: 'v1-node' }),
+			]);
+			expect(graph.edges).toEqual([
+				{ from: MANUAL_TRIGGER.id, to: SET_NODE.id, outputIndex: 0, inputIndex: 0 },
+			]);
+		});
+
+		it('converts only the branch of the selected manual trigger', async () => {
+			const otherTrigger = node(
+				'other-trigger-id',
+				'When clicking Other Execute',
+				'n8n-nodes-base.manualTrigger',
+			);
+			const otherSetNode = node('other-set-id', 'Other Edit Fields', 'n8n-nodes-base.set');
+			const data = runData({
+				workflowData: workflow({
+					nodes: [MANUAL_TRIGGER, SET_NODE, otherTrigger, otherSetNode],
+					connections: {
+						[MANUAL_TRIGGER.name]: {
+							main: [[{ node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 }]],
+						},
+						[otherTrigger.name]: {
+							main: [
+								[
+									{ node: SET_NODE.name, type: NodeConnectionTypes.Main, index: 0 },
+									{ node: otherSetNode.name, type: NodeConnectionTypes.Main, index: 0 },
+								],
+							],
+						},
+					},
+				}),
+			});
+
+			await dispatcher.start(data);
+
+			const { graph } = proxy.startExecution.mock.calls[0][0];
+			expect(graph.nodes).toEqual([
+				{ id: MANUAL_TRIGGER.id, name: MANUAL_TRIGGER.name, type: 'trigger' },
+				expect.objectContaining({ id: SET_NODE.id, type: 'v1-node' }),
+			]);
+			expect(graph.edges).toEqual([
+				{ from: MANUAL_TRIGGER.id, to: SET_NODE.id, outputIndex: 0, inputIndex: 0 },
+			]);
+		});
+
+		it('checks credential permissions before converting', async () => {
+			const failure = new UserError('Node "X" uses invalid credential');
+			credentialsPermissionChecker.check.mockRejectedValueOnce(failure);
+
+			await expect(dispatcher.start(runData())).rejects.toThrow(failure);
+
+			expect(credentialsPermissionChecker.check).toHaveBeenCalledWith('wf-1', [
+				MANUAL_TRIGGER,
+				SET_NODE,
+			]);
+			expect(proxy.startExecution).not.toHaveBeenCalled();
+		});
+
+		it('propagates a converter rejection for an unsupported trigger', async () => {
+			const scheduleTrigger = { ...MANUAL_TRIGGER, type: 'n8n-nodes-base.scheduleTrigger' };
+			const data = runData({
+				workflowData: workflow({ nodes: [scheduleTrigger, SET_NODE] }),
+			});
+
+			await expect(dispatcher.start(data)).rejects.toThrow(UnsupportedTriggerError);
+			expect(proxy.startExecution).not.toHaveBeenCalled();
+		});
+
+		describe('rejections', () => {
+			it('reports the module being off first', async () => {
+				proxy.isAvailable.mockReturnValue(false);
+				// also unsupported, to prove the module check wins
+				const data = runData({
+					destinationNode: { nodeName: SET_NODE.name, mode: 'inclusive' as const },
+				});
+
+				await expect(dispatcher.start(data)).rejects.toThrow(
+					'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.',
+				);
+			});
+
+			it.each([
+				{
+					name: 'a partial execution',
+					data: { runData: {} as IRunData },
+					message:
+						'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.',
+				},
+				{
+					name: 'a destination node',
+					data: { destinationNode: { nodeName: SET_NODE.name, mode: 'inclusive' as const } },
+					message:
+						'Engine 2.0 cannot run a workflow up to a single node yet. Run the whole workflow instead.',
+				},
+				{
+					name: 'selected start nodes',
+					data: { startNodes: [mock<StartNodeData>()] },
+					message:
+						'Engine 2.0 cannot start from selected nodes yet. Run the whole workflow instead.',
+				},
+				{
+					name: 'an AI tool run',
+					data: { agentRequest: { query: { [SET_NODE.name]: 'do it' }, tool: { name: 'tool' } } },
+					message: 'Engine 2.0 cannot run a workflow as an AI tool yet.',
+				},
+				{
+					name: 'pinned data on a non-trigger node',
+					data: { pinData: { [SET_NODE.name]: [{ json: { pinned: true } }] } as IPinData },
+					message:
+						'Engine 2.0 does not support pinned data on "Edit Fields" yet. Unpin it to run this workflow.',
+				},
+			])('rejects $name', async ({ data, message }) => {
+				const attempt = dispatcher.start(runData(data));
+
+				await expect(attempt).rejects.toThrow(UserError);
+				await expect(attempt).rejects.toThrow(message);
+				expect(proxy.startExecution).not.toHaveBeenCalled();
+			});
+		});
+
+		describe('triggerOutputs', () => {
+			const startedWith = () => proxy.startExecution.mock.calls[0][0].triggerOutputs;
+
+			it('defaults to one slot with one empty item', async () => {
+				await dispatcher.start(runData());
+
+				expect(startedWith()).toEqual([[{ json: {} }]]);
+			});
+
+			it('passes the trigger payload through', async () => {
+				const data = runData({
+					triggerToStartFrom: {
+						name: MANUAL_TRIGGER.name,
+						data: taskData([[{ json: { from: 'trigger' } }]]),
+					},
+				});
+
+				await dispatcher.start(data);
+
+				expect(startedWith()).toEqual([[{ json: { from: 'trigger' } }]]);
+			});
+
+			it('uses pinned data on the start trigger', async () => {
+				const data = runData({
+					pinData: { [MANUAL_TRIGGER.name]: [{ json: { from: 'pin' } }] } as IPinData,
+				});
+
+				await dispatcher.start(data);
+
+				expect(startedWith()).toEqual([[{ json: { from: 'pin' } }]]);
+			});
+
+			it('prefers the trigger payload over pinned data for the same trigger', async () => {
+				const data = runData({
+					triggerToStartFrom: {
+						name: MANUAL_TRIGGER.name,
+						data: taskData([[{ json: { from: 'trigger' } }]]),
+					},
+					pinData: { [MANUAL_TRIGGER.name]: [{ json: { from: 'pin' } }] } as IPinData,
+				});
+
+				await dispatcher.start(data);
+
+				expect(startedWith()).toEqual([[{ json: { from: 'trigger' } }]]);
+			});
+
+			it('collapses an empty slot to a dead edge', async () => {
+				const data = runData({
+					triggerToStartFrom: {
+						name: MANUAL_TRIGGER.name,
+						data: taskData([[]]),
+					},
+				});
+
+				await dispatcher.start(data);
+
+				expect(startedWith()).toEqual([null]);
+			});
+
+			it('sends null rather than an empty array when there are no slots', async () => {
+				const data = runData({
+					triggerToStartFrom: {
+						name: MANUAL_TRIGGER.name,
+						data: taskData([]),
+					},
+				});
+
+				await dispatcher.start(data);
+
+				expect(startedWith()).toBeNull();
+			});
+		});
+	});
+});

packages/cli/src/services/engine-data-plane-proxy.service.ts

@@ -18,9 +18,6 @@ export interface EngineDataPlaneProvider {
  * The module registers itself here on init. Without the module enabled there is
  * no provider, and calling into the engine throws rather than degrading silently:
  * a dropped execution would be worse than a loud failure.
- *
- * TODO(CAT-2877): nothing calls this yet. The dispatch that routes on the
- * per-workflow `engineType` setting lands with the parent ticket.
  */
 @Service()
 export class EngineDataPlaneProxyService implements EngineDataPlaneProvider {
@@ -30,6 +27,11 @@ export class EngineDataPlaneProxyService implements EngineDataPlaneProvider {
 		this.provider = provider;
 	}
 
+	/** Whether the `engine-v2` module is enabled and has registered itself. */
+	isAvailable(): boolean {
+		return this.provider !== null;
+	}
+
 	async startExecution(request: StartExecutionRequest): Promise<StartExecutionResult> {
 		if (!this.provider) {
 			throw new UserError(

packages/cli/src/services/engine-v2-dispatcher.service.ts

@@ -0,0 +1,174 @@
+import { Service } from '@n8n/di';
+import type { StepSlots, TriggerOutputs } from '@n8n/engine';
+import type {
+	INodeExecutionData,
+	IWorkflowBase,
+	IWorkflowExecutionDataProcess,
+} from 'n8n-workflow';
+import { getChildNodes, NodeConnectionTypes, UserError } from 'n8n-workflow';
+
+import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks';
+import type { ResumableExecution } from '@/interfaces';
+import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
+
+type ToStepOutputs = (outputs: INodeExecutionData[][]) => StepSlots;
+
+/** v1's payload for a manual run with no trigger data: one slot, one empty item. */
+const DEFAULT_MAIN_OUTPUT: INodeExecutionData[][] = [[{ json: {} }]];
+
+/**
+ * Routes a run to the engine 2.0 data plane and starts it there.
+ *
+ * The single dispatch point for the v2 path: {@link routesToEngineV2} decides,
+ * {@link start} runs. A workflow that opts into engine 2.0 never falls back to
+ * v1 — anything the v2 path cannot do fails with a user-facing reason instead,
+ * because a silent fallback would run the workflow on an engine the user did
+ * not pick.
+ *
+ * No control-plane execution row is created: the data plane is the source of
+ * truth, and the returned execution id is its UUID.
+ */
+@Service()
+export class EngineV2Dispatcher {
+	constructor(
+		private readonly proxy: EngineDataPlaneProxyService,
+		private readonly credentialsPermissionChecker: CredentialsPermissionChecker,
+	) {}
+
+	/**
+	 * Manual runs only for now: webhook (CAT-2920) and trigger (CAT-2921) entry
+	 * paths reuse this seam later. A resume must not start a fresh data-plane
+	 * execution, hence the `existingExecution` check.
+	 */
+	routesToEngineV2(
+		data: IWorkflowExecutionDataProcess,
+		existingExecution?: ResumableExecution,
+	): boolean {
+		return (
+			data.workflowData.settings?.engineType === 'v2' &&
+			data.executionMode === 'manual' &&
+			existingExecution === undefined
+		);
+	}
+
+	/** Returns the data plane's execution id. */
+	async start(data: IWorkflowExecutionDataProcess): Promise<string> {
+		this.assertSupported(data);
+
+		const { workflowData } = data;
+
+		await this.credentialsPermissionChecker.check(workflowData.id, workflowData.nodes);
+
+		// Lazily imported: a top-level import would pull the v1 step executor and
+		// its dependencies into every n8n process, including ones with the module off.
+		const { V1WorkflowConverter, toStepOutputs } = await import('@n8n/node-engine-compatibility');
+
+		const graph = new V1WorkflowConverter().convert(this.selectTriggerSubgraph(data));
+
+		const { executionId } = await this.proxy.startExecution({
+			workflowId: workflowData.id,
+			graph,
+			triggerOutputs: this.toTriggerOutputs(data, toStepOutputs),
+			mode: 'manual',
+		});
+
+		return executionId;
+	}
+
+	/** Keep only the branch that starts at the trigger selected for this manual run. */
+	private selectTriggerSubgraph(data: IWorkflowExecutionDataProcess): IWorkflowBase {
+		const { triggerToStartFrom, workflowData } = data;
+		if (triggerToStartFrom === undefined) return workflowData;
+
+		const includedNodeNames = new Set([
+			triggerToStartFrom.name,
+			...getChildNodes(
+				workflowData.connections,
+				triggerToStartFrom.name,
+				NodeConnectionTypes.Main,
+				-1,
+			),
+		]);
+
+		return {
+			...workflowData,
+			nodes: workflowData.nodes.filter((node) => includedNodeNames.has(node.name)),
+			connections: Object.fromEntries(
+				Object.entries(workflowData.connections).filter(([sourceNodeName]) =>
+					includedNodeNames.has(sourceNodeName),
+				),
+			),
+		};
+	}
+
+	/**
+	 * Rejects what the v2 path cannot do yet, in the order the user should hear
+	 * about it: the module being off comes first, so a workflow that would also
+	 * fail conversion does not report the conversion problem and hide the real
+	 * cause.
+	 */
+	private assertSupported(data: IWorkflowExecutionDataProcess): void {
+		if (!this.proxy.isAvailable()) {
+			throw new UserError(
+				'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.',
+			);
+		}
+
+		if (data.runData !== undefined) {
+			throw new UserError(
+				'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.',
+			);
+		}
+
+		// The engine cannot stop at a node, so ignoring this would run nodes the
+		// user did not ask for, with their side effects.
+		if (data.destinationNode !== undefined) {
+			throw new UserError(
+				'Engine 2.0 cannot run a workflow up to a single node yet. Run the whole workflow instead.',
+			);
+		}
+
+		if (data.startNodes?.length) {
+			throw new UserError(
+				'Engine 2.0 cannot start from selected nodes yet. Run the whole workflow instead.',
+			);
+		}
+
+		if (data.agentRequest !== undefined) {
+			throw new UserError('Engine 2.0 cannot run a workflow as an AI tool yet.');
+		}
+
+		const triggerName = data.triggerToStartFrom?.name;
+		const pinnedNode = Object.keys(data.pinData ?? {}).find((name) => name !== triggerName);
+		if (pinnedNode !== undefined) {
+			throw new UserError(
+				`Engine 2.0 does not support pinned data on "${pinnedNode}" yet. Unpin it to run this workflow.`,
+			);
+		}
+	}
+
+	/**
+	 * A manual run always carries a payload. Sending none would make the engine
+	 * record the trigger with no slots, so every successor edge reads as dead and
+	 * the execution completes having run nothing.
+	 */
+	private toTriggerOutputs(
+		data: IWorkflowExecutionDataProcess,
+		toStepOutputs: ToStepOutputs,
+	): TriggerOutputs | null {
+		const triggerName = data.triggerToStartFrom?.name;
+		// `IPinData` values are a flat item array; the Manual Trigger has one output.
+		const pinned = triggerName ? data.pinData?.[triggerName] : undefined;
+		const main =
+			data.triggerToStartFrom?.data?.data?.main ??
+			(pinned ? [pinned] : undefined) ??
+			DEFAULT_MAIN_OUTPUT;
+
+		// v1 uses `null` for a slot it has no data for; an empty slot says the same
+		// thing to the engine, which `toStepOutputs` collapses back to a dead edge.
+		const slots = toStepOutputs(main.map((slot) => slot ?? []));
+
+		// The wire schema rejects an empty array; `null` is how "no slots" is sent.
+		return slots.length === 0 ? null : slots;
+	}
+}

packages/cli/src/workflow-runner.ts

@@ -52,6 +52,7 @@ import { ManualExecutionService } from '@/manual-execution.service';
 import { NodeTypes } from '@/node-types';
 import type { ScalingService } from '@/scaling/scaling.service';
 import type { Job, JobData } from '@/scaling/scaling.types';
+import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
 import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
 import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
 
@@ -93,6 +94,7 @@ export class WorkflowRunner {
 		private readonly executionsConfig: ExecutionsConfig,
 		private readonly storageConfig: StorageConfig,
 		private readonly externalHooks: ExternalHooks,
+		private readonly engineV2Dispatcher: EngineV2Dispatcher,
 	) {}
 
 	/** The process did error */
@@ -246,6 +248,12 @@ export class WorkflowRunner {
 		existingExecution?: ResumableExecution,
 		responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
 	): Promise<string> {
+		// The engine 2.0 path owns the whole run: it keeps no control-plane
+		// execution row, so everything below here does not apply to it.
+		if (this.engineV2Dispatcher.routesToEngineV2(data, existingExecution)) {
+			return await this.engineV2Dispatcher.start(data);
+		}
+
 		const establishContextError = await this.establishContextForPersistence(data);
 
 		// Register a new execution

packages/cli/test/integration/workflows/workflows.controller.test.ts

@@ -42,11 +42,13 @@ import { v4 as uuid } from 'uuid';
 import { ActiveWorkflowManager } from '@/active-workflow-manager';
 import { CollaborationService } from '@/collaboration/collaboration.service';
 import { EventService } from '@/events/event.service';
+import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
 import { ProjectService } from '@/services/project.service.ee';
 import { WorkflowValidationService } from '@/workflows/workflow-validation.service';
 import { createFolder } from '@test-integration/db/folders';
 
 import { saveCredential } from '../shared/db/credentials';
+import { getAllExecutions } from '../shared/db/executions';
 import { createCustomRoleWithScopeSlugs, cleanupRolesAndScopes } from '../shared/db/roles';
 import { assignTagToWorkflow, createTag } from '../shared/db/tags';
 import {
@@ -4968,6 +4970,96 @@ describe('POST /workflows/:workflowId/run', () => {
 			'To run the workflow manually, specify either a trigger to start from or a destination node.',
 		);
 	});
+
+	describe('with engineType v2', () => {
+		const TRIGGER_NAME = 'When clicking Execute';
+		const SET_NAME = 'Edit Fields';
+
+		const startExecution = vi.fn();
+
+		beforeAll(() => {
+			Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution });
+		});
+
+		beforeEach(() => {
+			startExecution.mockResolvedValue({ executionId: 'a3c1e0f2-0000-4000-8000-000000000001' });
+		});
+
+		const createV2Workflow = async () =>
+			await createWorkflow(
+				{
+					nodes: [
+						{
+							id: uuid(),
+							name: TRIGGER_NAME,
+							type: 'n8n-nodes-base.manualTrigger',
+							parameters: {},
+							typeVersion: 1,
+							position: [0, 0],
+						},
+						{
+							id: uuid(),
+							name: SET_NAME,
+							type: 'n8n-nodes-base.set',
+							parameters: {},
+							typeVersion: 3.4,
+							position: [200, 0],
+						},
+					],
+					connections: {
+						[TRIGGER_NAME]: { main: [[{ node: SET_NAME, type: 'main', index: 0 }]] },
+					},
+					settings: { engineType: 'v2' },
+				},
+				owner,
+			);
+
+		test('should run on the data plane and persist no execution', async () => {
+			const dbWorkflow = await createV2Workflow();
+
+			const response = await authOwnerAgent
+				.post(`/workflows/${dbWorkflow.id}/run`)
+				.send({ triggerToStartFrom: { name: TRIGGER_NAME } });
+
+			expect(response.statusCode).toBe(200);
+			expect(response.body.data.executionId).toBe('a3c1e0f2-0000-4000-8000-000000000001');
+			expect(startExecution).toHaveBeenCalledWith(
+				objectContaining({
+					workflowId: dbWorkflow.id,
+					mode: 'manual',
+					triggerOutputs: [[{ json: {} }]],
+				}),
+			);
+
+			const executions = await getAllExecutions();
+			expect(executions.filter((e) => e.workflowId === dbWorkflow.id)).toHaveLength(0);
+		});
+
+		test('should return 400 for a partial execution', async () => {
+			const dbWorkflow = await createV2Workflow();
+
+			const response = await authOwnerAgent.post(`/workflows/${dbWorkflow.id}/run`).send({
+				destinationNode: { nodeName: SET_NAME, mode: 'inclusive' },
+				runData: {
+					[TRIGGER_NAME]: [
+						{
+							startTime: 0,
+							executionTime: 0,
+							executionIndex: 0,
+							source: [],
+							data: { main: [[{ json: {} }]] },
+						},
+					],
+				},
+			});
+
+			expect(response.statusCode).toBe(400);
+			expect(response.body.message).toBe(
+				'Engine 2.0 cannot run a workflow from existing data yet. Run the whole workflow instead.',
+			);
+			expect(startExecution).not.toHaveBeenCalled();
+		});
+	});
 });
 
 describe('POST /workflows/:workflowId/archive', () => {