feat(core): Resolve expressions in v1 step executor (no-changelog) (#35338)

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

@@ -1,5 +1,6 @@
 import type { JsonObject, JsonValue, StepExecutionRequest, WorkflowGraph } from '@n8n/engine';
 import type { ExecuteContext } from 'n8n-core';
+import { UnrecognizedNodeTypeError } from 'n8n-core';
 import { NoOp } from 'n8n-nodes-base/nodes/NoOp/NoOp.node';
 import type {
 	CloseFunction,
@@ -16,6 +17,8 @@ import type {
 } from 'n8n-workflow';
 import { Node, NodeConnectionTypes } from 'n8n-workflow';
 
+import { V1StepExecutor } from '../v1-step-executor';
+
 class EchoParam implements INodeType {
 	description = {
 		displayName: 'Echo Param',
@@ -193,8 +196,16 @@ const registry = new Map<string, INodeType>([
 ]);
 
 export const testNodeTypes: INodeTypes = {
-	getByName: (type: string): INodeType | IVersionedNodeType => registry.get(type)!,
-	getByNameAndVersion: (type: string): INodeType => registry.get(type)!,
+	getByName: (type: string): INodeType | IVersionedNodeType => {
+		const nodeType = registry.get(type);
+		if (!nodeType) throw new UnrecognizedNodeTypeError('test', type);
+		return nodeType;
+	},
+	getByNameAndVersion: (type: string): INodeType => {
+		const nodeType = registry.get(type);
+		if (!nodeType) throw new UnrecognizedNodeTypeError('test', type);
+		return nodeType;
+	},
 	getKnownTypes: (): IDataObject => ({}),
 };
 
@@ -239,7 +250,17 @@ export const stepRequest = (
 ): StepExecutionRequest => ({
 	node: graph.nodes.find((n) => n.id === nodeId)!,
 	inputs,
-	context: { executionId: 'exec-1', stepId: 'step-1', workflowId: 'wf-1', mode: 'manual' },
+	context: { executionId: 'exec-1', stepId: nodeId, workflowId: 'wf-1', mode: 'manual' },
 });
 
+export const testStepExecutor = (
+	graph: WorkflowGraph,
+	outputsByStepId: Record<string, JsonValue> = {},
+): V1StepExecutor =>
+	new V1StepExecutor({
+		nodeTypes: testNodeTypes,
+		additionalDataFactory: testAdditionalDataFactory,
+		loadStepData: async () => await Promise.resolve({ graph, outputsByStepId }),
+	});
+
 export const items = (...objects: JsonObject[]): JsonValue => [objects.map((json) => ({ json }))];

packages/@n8n/node-engine-compatibility/src/__tests__/setup-vm-evaluator.ts

@@ -0,0 +1,16 @@
+import { Expression } from 'n8n-workflow';
+import { afterAll, beforeAll } from 'vitest';
+
+beforeAll(async () => {
+	await Expression.initExpressionEngine({
+		engine: 'vm',
+		poolSize: 1,
+		maxCodeCacheSize: 1024,
+		bridgeTimeout: 5000,
+		bridgeMemoryLimit: 128,
+	});
+});
+
+afterAll(async () => {
+	await Expression.disposeExpressionEngine();
+});

packages/@n8n/node-engine-compatibility/src/__tests__/v1-adapters.test.ts

@@ -0,0 +1,105 @@
+import { describe, expect, it } from 'vitest';
+
+import { toV1Execution, toV1Sources } from '../v1-adapters';
+import { V1WorkflowConverter } from '../v1-workflow-converter';
+import { items, v1Workflow } from './fixtures';
+
+const converter = new V1WorkflowConverter();
+
+// The shared fixture graph, in the notation of the converter tests
+// (oN / iN mark output / input slots, (back) the loop-return edge):
+//
+// ┌───────┐    ┌─┐ o1    i1 ┌─┐    ┌────┐ o1    ┌────┐
+// │Trigger├───►│A├─────────►│B├───►│Loop├──────►│Body│
+// └───────┘    └─┘          └─┘    └─▲──┘       └──┬─┘
+//                                    └───(back)────┘
+const graph = converter.convert(
+	v1Workflow(
+		[
+			{ id: 't', name: 'Trigger', type: 'n8n-nodes-base.manualTrigger' },
+			{ id: 'a', name: 'A', type: 'test.echoParam' },
+			{ id: 'b', name: 'B', type: 'test.echoParam' },
+			{ id: 'loop', name: 'Loop', type: 'n8n-nodes-base.splitInBatches', typeVersion: 3 },
+			{ id: 'body', name: 'Body', type: 'n8n-nodes-base.noOp' },
+		],
+		{
+			Trigger: { main: [[{ node: 'A', type: 'main', index: 0 }]] },
+			A: { main: [[], [{ node: 'B', type: 'main', index: 1 }]] },
+			B: { main: [[{ node: 'Loop', type: 'main', index: 0 }]] },
+			Loop: { main: [[], [{ node: 'Body', type: 'main', index: 0 }]] },
+			Body: { main: [[{ node: 'Loop', type: 'main', index: 0 }]] },
+		},
+	),
+);
+
+describe('toV1Execution', () => {
+	it('rebuilds v1 nodes, including a stub for the trigger', () => {
+		const execution = toV1Execution(graph, {});
+
+		expect(execution.nodes.map((node) => node.name).sort()).toEqual([
+			'A',
+			'B',
+			'Body',
+			'Loop',
+			'Trigger',
+		]);
+		expect(execution.nodes.find((node) => node.name === 'Trigger')).toMatchObject({
+			type: 'n8n-nodes-base.manualTrigger',
+		});
+		expect(execution.nodes.find((node) => node.name === 'A')).toMatchObject({
+			type: 'test.echoParam',
+			typeVersion: 1,
+			continueOnFail: false,
+		});
+	});
+
+	it('rebuilds name-keyed connections preserving slots', () => {
+		const execution = toV1Execution(graph, {});
+
+		expect(execution.connections.A).toEqual({
+			main: [[], [{ node: 'B', type: 'main', index: 1 }]],
+		});
+	});
+
+	it('rebuilds run data with input provenance from the edges', () => {
+		const execution = toV1Execution(graph, { a: items({ x: 1 }) });
+
+		expect(execution.runData).toEqual({
+			A: [
+				{
+					startTime: 0,
+					executionTime: 0,
+					executionIndex: 0,
+					source: [{ previousNode: 'Trigger', previousNodeOutput: 0 }],
+					data: { main: [[{ json: { x: 1 } }]] },
+				},
+			],
+		});
+		expect(toV1Sources(graph).get('b')).toEqual([
+			null,
+			{ previousNode: 'A', previousNodeOutput: 1 },
+		]);
+	});
+
+	it('keeps back edges in connections but not in provenance', () => {
+		const execution = toV1Execution(graph, {});
+
+		expect(execution.connections.Body).toEqual({
+			main: [[{ node: 'Loop', type: 'main', index: 0 }]],
+		});
+		expect(toV1Sources(graph).get('loop')).toEqual([{ previousNode: 'B', previousNodeOutput: 0 }]);
+	});
+
+	it('omits nodes whose config is unreadable', () => {
+		const brokenGraph = {
+			nodes: [
+				{ id: 'ok', name: 'OK', type: 'v1-node' as const, config: graph.nodes[1].config },
+				{ id: 'bad', name: 'Bad', type: 'v1-node' as const, config: 'nonsense' },
+			],
+			edges: [],
+		};
+
+		const execution = toV1Execution(brokenGraph, {});
+		expect(execution.nodes.map((node) => node.name)).toEqual(['OK']);
+	});
+});

packages/@n8n/node-engine-compatibility/src/__tests__/v1-step-executor.test.ts

@@ -1,28 +1,20 @@
 import type { WorkflowGraph } from '@n8n/engine';
-import { describe, expect, it } from 'vitest';
+import { UnrecognizedNodeTypeError } from 'n8n-core';
+import type { IDataObject } from 'n8n-workflow';
+import { Expression, ExpressionError } from 'n8n-workflow';
+import { describe, expect, it, vi } from 'vitest';
 
 import {
 	EngineRequestNotSupportedError,
 	MalformedStepConfigError,
-	UnknownNodeTypeError,
 	UnsupportedNodeTypeError,
 	UnsupportedStepTypeError,
+	VmExpressionEngineRequiredError,
 } from '../errors';
-import { V1StepExecutor } from '../v1-step-executor';
 import { V1WorkflowConverter } from '../v1-workflow-converter';
-import {
-	items,
-	stepRequest,
-	testAdditionalDataFactory,
-	testNodeTypes,
-	v1Workflow,
-} from './fixtures';
+import { items, stepRequest, testStepExecutor, v1Workflow } from './fixtures';
 
 const converter = new V1WorkflowConverter();
-const executor = new V1StepExecutor({
-	nodeTypes: testNodeTypes,
-	additionalDataFactory: testAdditionalDataFactory,
-});
 
 const graphWith = (type: string, parameters = {}): WorkflowGraph =>
 	converter.convert(
@@ -33,9 +25,22 @@ const graphWith = (type: string, parameters = {}): WorkflowGraph =>
 	);
 
 describe('V1StepExecutor', () => {
+	it('rejects legacy expression engine', async () => {
+		vi.spyOn(Expression, 'getActiveImplementation').mockReturnValue('legacy');
+		try {
+			const graph = graphWith('test.echoParam', { message: 'hi' });
+			const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({})));
+			await expect(execution).rejects.toThrow(VmExpressionEngineRequiredError);
+		} finally {
+			vi.restoreAllMocks();
+		}
+	});
+
 	it('resolves `getNodeParameter` per item', async () => {
 		const graph = graphWith('test.echoParam', { message: 'hi' });
-		const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 }, { a: 2 })));
+		const result = await testStepExecutor(graph).execute(
+			stepRequest(graph, 'n', items({ a: 1 }, { a: 2 })),
+		);
 		expect(result.outputs).toEqual([[{ json: { message: 'hi' } }, { json: { message: 'hi' } }]]);
 		// eslint-disable-next-line n8n-local-rules/no-json-parse-json-stringify
 		expect(JSON.parse(JSON.stringify(result.outputs))).toEqual(result.outputs);
@@ -43,14 +48,14 @@ describe('V1StepExecutor', () => {
 
 	it('guarantees one item for empty input', async () => {
 		const graph = graphWith('test.echoParam', { message: 'solo' });
-		const result = await executor.execute(stepRequest(graph, 'n', []));
+		const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', []));
 		expect(result.outputs).toEqual([[{ json: { message: 'solo' } }]]);
 	});
 
 	it('rejects non-v1-node steps', async () => {
 		const graph = graphWith('n8n-nodes-base.noOp');
 		const triggerStep = stepRequest(graph, 't', []);
-		const execution = executor.execute(triggerStep);
+		const execution = testStepExecutor(graph).execute(triggerStep);
 		await expect(execution).rejects.toThrow(UnsupportedStepTypeError);
 	});
 
@@ -67,31 +72,31 @@ describe('V1StepExecutor', () => {
 		const graph = graphWith('n8n-nodes-base.noOp');
 		const request = stepRequest(graph, 'n', []);
 		request.node = { ...request.node, config };
-		const execution = executor.execute(request);
+		const execution = testStepExecutor(graph).execute(request);
 		await expect(execution).rejects.toThrow(MalformedStepConfigError);
 	});
 
 	it('rejects unknown node types', async () => {
 		const graph = graphWith('test.doesNotExist');
-		const execution = executor.execute(stepRequest(graph, 'n', []));
-		await expect(execution).rejects.toThrow(UnknownNodeTypeError);
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', []));
+		await expect(execution).rejects.toThrow(UnrecognizedNodeTypeError);
 	});
 
 	it('rejects node types without an execute method', async () => {
 		const graph = graphWith('test.noExecute');
-		const execution = executor.execute(stepRequest(graph, 'n', []));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', []));
 		await expect(execution).rejects.toThrow(UnsupportedNodeTypeError);
 	});
 
 	it('propagates node errors per the IStepExecutor failure contract', async () => {
 		const graph = graphWith('test.alwaysFails');
-		const execution = executor.execute(stepRequest(graph, 'n', []));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', []));
 		await expect(execution).rejects.toThrow('boom from node');
 	});
 
 	it('invokes new-style Node subclasses with the context as argument', async () => {
 		const graph = graphWith('test.newStyleEcho');
-		const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		expect(result.outputs).toEqual([[{ json: { a: 1, newStyle: true } }]]);
 	});
 
@@ -103,25 +108,27 @@ describe('V1StepExecutor', () => {
 		(workflow.nodes[1] as { continueOnFail?: boolean }).continueOnFail = true;
 		const graph = converter.convert(workflow);
 
-		const result = await executor.execute(stepRequest(graph, 'n', items({ keep: 'me' })));
+		const result = await testStepExecutor(graph).execute(
+			stepRequest(graph, 'n', items({ keep: 'me' })),
+		);
 		expect(result.outputs).toEqual([[{ json: { keep: 'me' } }]]);
 	});
 
 	it('propagates cleanup errors when the node succeeded', async () => {
 		const graph = graphWith('test.succeedsWithFailingCleanup');
-		const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		await expect(execution).rejects.toThrow('cleanup boom');
 	});
 
 	it('preserves the node error over cleanup errors when both fail', async () => {
 		const graph = graphWith('test.failsWithFailingCleanup');
-		const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		await expect(execution).rejects.toThrow('boom from node');
 	});
 
 	it('throws on EngineRequest results instead of dropping them', async () => {
 		const graph = graphWith('test.returnsEngineRequest');
-		const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		await expect(execution).rejects.toThrow(EngineRequestNotSupportedError);
 	});
 
@@ -133,7 +140,7 @@ describe('V1StepExecutor', () => {
 		(workflow.nodes[1] as { continueOnFail?: boolean }).continueOnFail = true;
 		const graph = converter.convert(workflow);
 
-		const execution = executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		await expect(execution).rejects.toThrow(EngineRequestNotSupportedError);
 	});
 
@@ -158,7 +165,7 @@ describe('V1StepExecutor', () => {
 			),
 		);
 
-		const result = await executor.execute(stepRequest(graph, 'n', items({ a: 1 })));
+		const result = await testStepExecutor(graph).execute(stepRequest(graph, 'n', items({ a: 1 })));
 		expect(result.outputs).toEqual([[{ json: { a: 1 } }], [{ json: { a: 1, second: true } }]]);
 	});
 
@@ -178,10 +185,90 @@ describe('V1StepExecutor', () => {
 			),
 		);
 
-		const execution = executor.execute(stepRequest(graph, 'loop', items({ a: 1 })));
+		const execution = testStepExecutor(graph).execute(stepRequest(graph, 'loop', items({ a: 1 })));
 		await expect(execution).rejects.toThrow(UnsupportedStepTypeError);
 	});
 
+	describe('expressions', () => {
+		const expressionWorkflow = (bParameters: IDataObject) =>
+			converter.convert(
+				v1Workflow(
+					[
+						{ id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' },
+						{ id: 'a', name: 'A', type: 'test.echoParam', parameters: { message: 'from-A' } },
+						{ id: 'b', name: 'B', type: 'test.echoParam', parameters: bParameters },
+					],
+					{
+						Manual: { main: [[{ node: 'A', type: 'main', index: 0 }]] },
+						A: { main: [[{ node: 'B', type: 'main', index: 0 }]] },
+					},
+				),
+			);
+
+		it('resolves $json per item against the step inputs', async () => {
+			const graph = graphWith('test.echoParam', { message: '={{ $json.a }}' });
+			const result = await testStepExecutor(graph).execute(
+				stepRequest(graph, 'n', items({ a: 1 }, { a: 2 })),
+			);
+			expect(result.outputs).toEqual([[{ json: { message: 1 } }, { json: { message: 2 } }]]);
+		});
+
+		it('resolves $(...) against a completed step output, convert to execute', async () => {
+			const graph = expressionWorkflow({ message: "={{ $('A').first().json.message }}" });
+
+			const aResult = await testStepExecutor(graph).execute(stepRequest(graph, 'a', items({})));
+			const bResult = await testStepExecutor(graph, { a: aResult.outputs }).execute(
+				stepRequest(graph, 'b', aResult.outputs),
+			);
+
+			expect(bResult.outputs).toEqual([[{ json: { message: 'from-A' } }]]);
+		});
+
+		it('resolves $prevNode from the input provenance in the graph', async () => {
+			const graph = expressionWorkflow({ message: '={{ $prevNode.name }}' });
+			const result = await testStepExecutor(graph).execute(stepRequest(graph, 'b', items({})));
+			expect(result.outputs).toEqual([[{ json: { message: 'A' } }]]);
+		});
+
+		it('rejects a reference to an unexecuted node', async () => {
+			const graph = expressionWorkflow({ message: "={{ $('A').first().json.message }}" });
+			const execution = testStepExecutor(graph).execute(stepRequest(graph, 'b', items({})));
+			await expect(execution).rejects.toThrow("Node 'A' hasn't been executed");
+		});
+
+		it('rejects a reference to a node absent from the graph', async () => {
+			const graph = graphWith('test.echoParam', { message: "={{ $('Ghost').first().json.x }}" });
+			const execution = testStepExecutor(graph).execute(stepRequest(graph, 'n', items({})));
+			await expect(execution).rejects.toThrow(ExpressionError);
+		});
+
+		it('resolves a sibling reference even when its node type is not installed', async () => {
+			const graph = converter.convert(
+				v1Workflow(
+					[
+						{ id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' },
+						{ id: 'a', name: 'A', type: 'community.notInstalled' },
+						{
+							id: 'b',
+							name: 'B',
+							type: 'test.echoParam',
+							parameters: { message: "={{ $('A').first().json.tag }}" },
+						},
+					],
+					{
+						Manual: { main: [[{ node: 'A', type: 'main', index: 0 }]] },
+						A: { main: [[{ node: 'B', type: 'main', index: 0 }]] },
+					},
+				),
+			);
+
+			const result = await testStepExecutor(graph, { a: items({ tag: 'ran-before' }) }).execute(
+				stepRequest(graph, 'b', items({})),
+			);
+			expect(result.outputs).toEqual([[{ json: { message: 'ran-before' } }]]);
+		});
+	});
+
 	it('honors onError=continueRegularOutput as passthrough', async () => {
 		const workflow = v1Workflow([
 			{ id: 't', name: 'Manual', type: 'n8n-nodes-base.manualTrigger' },
@@ -190,7 +277,9 @@ describe('V1StepExecutor', () => {
 		(workflow.nodes[1] as { onError?: string }).onError = 'continueRegularOutput';
 		const graph = converter.convert(workflow);
 
-		const result = await executor.execute(stepRequest(graph, 'n', items({ keep: 'me' })));
+		const result = await testStepExecutor(graph).execute(
+			stepRequest(graph, 'n', items({ keep: 'me' })),
+		);
 		expect(result.outputs).toEqual([[{ json: { keep: 'me' } }]]);
 	});
 });

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

@@ -0,0 +1,3 @@
+export const MANUAL_TRIGGER_TYPE = 'n8n-nodes-base.manualTrigger';
+export const SPLIT_IN_BATCHES_TYPE = 'n8n-nodes-base.splitInBatches';
+export const MAIN_CONNECTION_TYPE = 'main';

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

@@ -1,4 +1,4 @@
-import { UserError } from 'n8n-workflow';
+import { UnexpectedError, UserError } from 'n8n-workflow';
 
 const quote = (names: string[]) => names.map((name) => `"${name}"`).join(', ');
 
@@ -52,18 +52,20 @@ export class MalformedStepConfigError extends UserError {
 	}
 }
 
-export class UnknownNodeTypeError extends UserError {
-	constructor(nodeType: string) {
-		super(`Unknown node type "${nodeType}"`);
-	}
-}
-
 export class UnsupportedNodeTypeError extends UserError {
 	constructor(nodeType: string) {
 		super(`Node type "${nodeType}" has no execute method and cannot run as a step`);
 	}
 }
 
+export class VmExpressionEngineRequiredError extends UnexpectedError {
+	constructor() {
+		super(
+			'V1StepExecutor requires the VM expression engine. Initialize it via `Expression.initExpressionEngine()` before executing steps.',
+		);
+	}
+}
+
 export class EngineRequestNotSupportedError extends UserError {
 	constructor(nodeType: string) {
 		super(

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

@@ -1,4 +1,4 @@
 export { V1WorkflowConverter } from './v1-workflow-converter';
 export { V1StepExecutor } from './v1-step-executor';
 export { UnsupportedTriggerError, UnsupportedWorkflowError } from './errors';
-export type { V1StepExecutorDeps } from './types';
+export type { StepData, StepDataLoader, V1StepExecutorDeps } from './types';

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

@@ -1,11 +1,16 @@
-import type { StepExecutionContext } from '@n8n/engine';
+import type { JsonValue, StepExecutionContext, WorkflowGraph } from '@n8n/engine';
 import type { ExecuteContext } from 'n8n-core';
 import type {
+	INode,
 	INodeExecutionData,
 	INodeParameters,
 	INodeType,
 	INodeTypes,
+	IRunData,
+	ISourceData,
+	IWorkflowBase,
 	IWorkflowExecuteAdditionalData,
+	Workflow,
 } from 'n8n-workflow';
 
 export interface V1NodeStepConfig {
@@ -15,24 +20,38 @@ export interface V1NodeStepConfig {
 	continueOnFail: boolean;
 }
 
+export interface V1Execution extends Pick<IWorkflowBase, 'nodes' | 'connections'> {
+	runData: IRunData;
+}
+
+export interface StepData {
+	graph: WorkflowGraph;
+	outputsByStepId: Record<string, JsonValue>;
+}
+
+export type StepDataLoader = (context: StepExecutionContext) => Promise<StepData>;
+
 export interface V1StepExecutorDeps {
 	nodeTypes: INodeTypes;
 	additionalDataFactory: (executionId: string) => Promise<IWorkflowExecuteAdditionalData>;
+	loadStepData: StepDataLoader;
 }
 
 export type ExecutableNodeType = INodeType & { execute: NonNullable<INodeType['execute']> };
 
 export type NodeRunResult = { ok: true; value: unknown } | { ok: false; error: unknown };
 
-export interface CreateNodeExecuteContextParams {
-	nodeId: string;
-	nodeName: string;
-	stepConfig: V1NodeStepConfig;
+export interface CreateExecuteContextParams {
+	node: INode;
+	workflow: Workflow;
+	execution: V1Execution;
+	source: Array<ISourceData | null>;
+	additionalData: IWorkflowExecuteAdditionalData;
 	itemsByConnection: INodeExecutionData[][];
 	stepContext: StepExecutionContext;
 }
 
 export interface RunNodeParams {
 	nodeType: ExecutableNodeType;
-	nodeExecuteContext: ExecuteContext;
+	context: ExecuteContext;
 }

packages/@n8n/node-engine-compatibility/src/v1-adapters.ts

@@ -0,0 +1,221 @@
+/**
+ * Pure functions to turn engine v2 artifacts (graph, step outputs, step context)
+ * back into engine v1 artifacts (`INode`, `Workflow`, run data, node execute context).
+ */
+
+import type { GraphNode, JsonValue, WorkflowGraph } from '@n8n/engine';
+import { ExecuteContext, UnrecognizedNodeTypeError } from 'n8n-core';
+import type {
+	IConnections,
+	IExecuteData,
+	INode,
+	INodeType,
+	INodeTypes,
+	IRunData,
+	ISourceData,
+	ITaskDataConnections,
+	WorkflowExecuteMode,
+} from 'n8n-workflow';
+import { createRunExecutionData, Workflow } from 'n8n-workflow';
+
+import { MAIN_CONNECTION_TYPE, MANUAL_TRIGGER_TYPE } from './constants';
+import { isV1NodeStepConfig } from './guards';
+import { fromStepInputs } from './io';
+import type { CreateExecuteContextParams, V1Execution, V1NodeStepConfig } from './types';
+
+export function toV1Execution(
+	graph: WorkflowGraph,
+	outputsByStepId: Record<string, JsonValue>,
+): V1Execution {
+	return {
+		nodes: toV1Nodes(graph),
+		connections: toV1Connections(graph),
+		runData: toV1RunData(graph, outputsByStepId),
+	};
+}
+
+function toV1Nodes(graph: WorkflowGraph): INode[] {
+	return graph.nodes.flatMap((graphNode): INode[] => {
+		if (graphNode.type === 'trigger') {
+			return [
+				{
+					id: graphNode.id,
+					name: graphNode.name,
+					type: MANUAL_TRIGGER_TYPE,
+					typeVersion: 1,
+					position: [0, 0],
+					parameters: {},
+				},
+			];
+		}
+
+		if (!isV1NodeStepConfig(graphNode.config)) return [];
+
+		return [toV1Node(graphNode, graphNode.config)];
+	});
+}
+
+function toV1Connections(graph: WorkflowGraph): IConnections {
+	const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
+
+	const connections: IConnections = {};
+	for (const edge of graph.edges) {
+		const fromName = namesById.get(edge.from);
+		const toName = namesById.get(edge.to);
+		if (fromName === undefined || toName === undefined) continue;
+
+		const outputs = (connections[fromName] ??= { [MAIN_CONNECTION_TYPE]: [] })[
+			MAIN_CONNECTION_TYPE
+		];
+		const outputIndex = edge.outputIndex ?? 0;
+		while (outputs.length <= outputIndex) outputs.push([]);
+		outputs[outputIndex]!.push({
+			node: toName,
+			type: MAIN_CONNECTION_TYPE,
+			index: edge.inputIndex ?? 0,
+		});
+	}
+
+	return connections;
+}
+
+function toV1RunData(graph: WorkflowGraph, outputsByStepId: Record<string, JsonValue>): IRunData {
+	const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
+	const sourcesByStepId = toV1Sources(graph);
+
+	const runData: IRunData = {};
+	for (const [completedStepId, outputs] of Object.entries(outputsByStepId)) {
+		const nodeName = namesById.get(completedStepId);
+		if (nodeName === undefined) continue;
+
+		runData[nodeName] = [
+			{
+				startTime: 0,
+				executionTime: 0,
+				executionIndex: 0,
+				source: sourcesByStepId.get(completedStepId) ?? [],
+				data: { [MAIN_CONNECTION_TYPE]: fromStepInputs(outputs) },
+			},
+		];
+	}
+
+	return runData;
+}
+
+export function toV1Sources(graph: WorkflowGraph): Map<string, Array<ISourceData | null>> {
+	const namesById = new Map(graph.nodes.map((node) => [node.id, node.name]));
+
+	const sourcesByStepId = new Map<string, Array<ISourceData | null>>();
+	for (const edge of graph.edges) {
+		if (edge.isBackEdge === true) continue;
+		const fromName = namesById.get(edge.from);
+		if (fromName === undefined) continue;
+
+		const source = sourcesByStepId.get(edge.to) ?? [];
+		const inputIndex = edge.inputIndex ?? 0;
+		while (source.length <= inputIndex) source.push(null);
+		source[inputIndex] ??= { previousNode: fromName, previousNodeOutput: edge.outputIndex };
+		sourcesByStepId.set(edge.to, source);
+	}
+	return sourcesByStepId;
+}
+
+export function toV1Workflow(
+	workflowId: string,
+	node: INode,
+	execution: V1Execution,
+	nodeTypes: INodeTypes,
+): Workflow {
+	const siblingNodes = execution.nodes.filter((executionNode) => executionNode.id !== node.id);
+
+	return new Workflow({
+		id: workflowId,
+		nodes: [node, ...siblingNodes],
+		connections: execution.connections,
+		active: false,
+		nodeTypes: tolerantNodeTypes(nodeTypes),
+	});
+}
+
+export function toV1Node(graphNode: GraphNode, config: V1NodeStepConfig): INode {
+	return {
+		id: graphNode.id,
+		name: graphNode.name,
+		type: config.nodeType,
+		typeVersion: config.typeVersion,
+		position: [0, 0],
+		parameters: config.parameters,
+		continueOnFail: config.continueOnFail,
+	};
+}
+
+/**
+ * In v1, an uninstalled node type _fails the whole execution upfront_, at
+ * `Workflow` construction. In v2, where each step builds its own `Workflow`,
+ * preserving that failure would mean failing at every step of the execution.
+ * Instead, for fault isolation, v2 maps this to `undefined`, which `Workflow`
+ * tolerates by design (node kept, name and data visible to expressions, only
+ * parameter defaults skipped), so only the missing node type step fails.
+ */
+function tolerantNodeTypes(nodeTypes: INodeTypes): INodeTypes {
+	return {
+		getByName: (type) => nodeTypes.getByName(type),
+		getKnownTypes: () => nodeTypes.getKnownTypes(),
+		getByNameAndVersion: (type, version) => {
+			try {
+				return nodeTypes.getByNameAndVersion(type, version);
+			} catch (error) {
+				if (error instanceof UnrecognizedNodeTypeError) return undefined as unknown as INodeType;
+				throw error;
+			}
+		},
+	};
+}
+
+export function toV1ExecuteContext({
+	node,
+	workflow,
+	execution,
+	source,
+	additionalData,
+	stepContext,
+	itemsByConnection,
+}: CreateExecuteContextParams): ExecuteContext {
+	const firstInput = itemsByConnection[0] ?? [];
+	const connectionInputData = firstInput.length > 0 ? firstInput : [{ json: {} }];
+	const inputData: ITaskDataConnections = {
+		main: [connectionInputData, ...itemsByConnection.slice(1)],
+	};
+
+	const runExecutionData = createRunExecutionData({
+		startData: {},
+		resultData: { runData: execution.runData },
+		executionData: {
+			contextData: {},
+			nodeExecutionStack: [],
+			metadata: {},
+			waitingExecution: {},
+			waitingExecutionSource: null,
+		},
+	});
+
+	const executeData: IExecuteData = {
+		node,
+		data: inputData,
+		source: source.length > 0 ? { main: source } : null,
+	};
+	const mode: WorkflowExecuteMode = stepContext.mode === 'production' ? 'trigger' : 'manual';
+
+	return new ExecuteContext(
+		workflow,
+		node,
+		additionalData,
+		mode,
+		runExecutionData,
+		0,
+		connectionInputData,
+		inputData,
+		executeData,
+		[],
+	);
+}

packages/@n8n/node-engine-compatibility/src/v1-step-executor.ts

@@ -4,38 +4,33 @@ import type {
 	StepExecutionRequest,
 	StepExecutionResult,
 } from '@n8n/engine';
-import { ExecuteContext } from 'n8n-core';
-import type {
-	IExecuteData,
-	INode,
-	INodeExecutionData,
-	ITaskDataConnections,
-	WorkflowExecuteMode,
-} from 'n8n-workflow';
-import {
-	createRunExecutionData,
-	isNodeClassInstance,
-	UnexpectedError,
-	Workflow,
-} from 'n8n-workflow';
+import { UnrecognizedNodeTypeError } from 'n8n-core';
+import type { INodeExecutionData } from 'n8n-workflow';
+import { Expression, isNodeClassInstance, UnexpectedError } from 'n8n-workflow';
 
 import {
 	EngineRequestNotSupportedError,
 	MalformedStepConfigError,
 	UnsupportedNodeTypeError,
-	UnknownNodeTypeError,
 	UnsupportedStepTypeError,
+	VmExpressionEngineRequiredError,
 } from './errors';
 import { isV1NodeStepConfig } from './guards';
 import { fromStepInputs, toStepOutputs } from './io';
 import type {
-	CreateNodeExecuteContextParams,
 	ExecutableNodeType,
 	NodeRunResult,
 	RunNodeParams,
 	V1NodeStepConfig,
 	V1StepExecutorDeps,
 } from './types';
+import {
+	toV1ExecuteContext,
+	toV1Execution,
+	toV1Node,
+	toV1Sources,
+	toV1Workflow,
+} from './v1-adapters';
 
 /**
  * Runs `v1-node` steps by adapting them to the v1 node runtime.
@@ -48,20 +43,37 @@ export class V1StepExecutor implements IStepExecutor {
 	constructor(private readonly deps: V1StepExecutorDeps) {}
 
 	async execute(request: StepExecutionRequest): Promise<StepExecutionResult> {
+		this.validateExpressionEngine();
 		const stepConfig = this.validateStepConfig(request.node);
 		const nodeType = this.resolveNodeType(stepConfig);
+		const stepData = await this.deps.loadStepData(request.context);
 
-		const nodeExecuteContext = await this.createNodeExecuteContext({
-			nodeId: request.node.id,
-			nodeName: request.node.name,
+		const node = toV1Node(request.node, stepConfig);
+		const execution = toV1Execution(stepData.graph, stepData.outputsByStepId);
+		const workflow = toV1Workflow(request.context.workflowId, node, execution, this.deps.nodeTypes);
+
+		const additionalData = await this.deps.additionalDataFactory(request.context.executionId);
+
+		const context = toV1ExecuteContext({
+			node,
+			workflow,
+			execution,
+			source: toV1Sources(stepData.graph).get(node.id) ?? [],
+			additionalData,
 			stepContext: request.context,
-			stepConfig,
 			itemsByConnection: fromStepInputs(request.inputs),
 		});
 
-		const nodeResult = await this.runNode({ nodeType, nodeExecuteContext });
+		return await workflow.expression.withIsolate(async () => {
+			const nodeResult = await this.runNode({ nodeType, context });
+			return { outputs: toStepOutputs(nodeResult) };
+		});
+	}
 
-		return { outputs: toStepOutputs(nodeResult) };
+	private validateExpressionEngine(): void {
+		if (Expression.getActiveImplementation() !== 'vm') {
+			throw new VmExpressionEngineRequiredError();
+		}
 	}
 
 	private validateStepConfig({ type, name, config }: GraphNode): V1NodeStepConfig {
@@ -76,73 +88,17 @@ export class V1StepExecutor implements IStepExecutor {
 		typeVersion,
 	}: V1NodeStepConfig): ExecutableNodeType {
 		const nodeType = this.deps.nodeTypes.getByNameAndVersion(typeName, typeVersion);
-		if (!nodeType) throw new UnknownNodeTypeError(typeName);
+
+		if (!nodeType) {
+			const [packageName, name = typeName] = typeName.split('.');
+			throw new UnrecognizedNodeTypeError(packageName, name);
+		}
+
 		if (typeof nodeType.execute !== 'function') throw new UnsupportedNodeTypeError(typeName);
 
 		return nodeType as ExecutableNodeType;
 	}
 
-	private async createNodeExecuteContext({
-		nodeId,
-		nodeName,
-		stepConfig,
-		stepContext: stepCtx,
-		itemsByConnection,
-	}: CreateNodeExecuteContextParams) {
-		const node: INode = {
-			id: nodeId,
-			name: nodeName,
-			type: stepConfig.nodeType,
-			typeVersion: stepConfig.typeVersion,
-			position: [0, 0],
-			parameters: stepConfig.parameters,
-			continueOnFail: stepConfig.continueOnFail,
-		};
-
-		const workflow = new Workflow({
-			id: stepCtx.workflowId,
-			nodes: [node],
-			connections: {},
-			active: false,
-			nodeTypes: this.deps.nodeTypes,
-		});
-
-		const firstInput = itemsByConnection[0] ?? [];
-		const connectionInputData = firstInput.length > 0 ? firstInput : [{ json: {} }];
-		const inputData: ITaskDataConnections = {
-			main: [connectionInputData, ...itemsByConnection.slice(1)],
-		};
-
-		const runExecutionData = createRunExecutionData({
-			startData: {},
-			resultData: { runData: {} },
-			executionData: {
-				contextData: {},
-				nodeExecutionStack: [],
-				metadata: {},
-				waitingExecution: {},
-				waitingExecutionSource: null,
-			},
-		});
-
-		const executeData: IExecuteData = { node, data: inputData, source: null };
-		const mode: WorkflowExecuteMode = stepCtx.mode === 'production' ? 'trigger' : 'manual';
-		const additionalData = await this.deps.additionalDataFactory(stepCtx.executionId);
-
-		return new ExecuteContext(
-			workflow,
-			node,
-			additionalData,
-			mode,
-			runExecutionData,
-			0,
-			connectionInputData,
-			inputData,
-			executeData,
-			[],
-		);
-	}
-
 	/**
 	 * Runs the node and settles its cleanup functions.
 	 *
@@ -150,21 +106,18 @@ export class V1StepExecutor implements IStepExecutor {
 	 * failure with `continueOnFail` passes the input items through as output.
 	 * A null result yields no output.
 	 */
-	private async runNode({
-		nodeType,
-		nodeExecuteContext,
-	}: RunNodeParams): Promise<INodeExecutionData[][]> {
+	private async runNode({ nodeType, context }: RunNodeParams): Promise<INodeExecutionData[][]> {
 		let result: NodeRunResult;
 		try {
 			const value = isNodeClassInstance(nodeType)
-				? await nodeType.execute(nodeExecuteContext)
-				: await nodeType.execute.call(nodeExecuteContext);
+				? await nodeType.execute(context)
+				: await nodeType.execute.call(context);
 			result = { ok: true, value };
 		} catch (error) {
 			result = { ok: false, error };
 		}
 
-		const { closeFunctions } = nodeExecuteContext;
+		const { closeFunctions } = context;
 		if (closeFunctions.length > 0) {
 			const closeResults = await Promise.allSettled(closeFunctions.map(async (fn) => await fn()));
 			if (result.ok) {
@@ -175,21 +128,21 @@ export class V1StepExecutor implements IStepExecutor {
 					throw rejected.reason instanceof Error
 						? rejected.reason
 						: new UnexpectedError("Error on node's close function", {
-								extra: { nodeName: nodeExecuteContext.getNode().name },
+								extra: { nodeName: context.getNode().name },
 							});
 				}
 			}
 		}
 
 		if (!result.ok) {
-			if (!nodeExecuteContext.continueOnFail()) throw result.error;
-			return [nodeExecuteContext.getInputData()];
+			if (!context.continueOnFail()) throw result.error;
+			return [context.getInputData()];
 		}
 
 		if (result.value === null || result.value === undefined) return [];
 
 		if (Array.isArray(result.value)) return result.value as INodeExecutionData[][];
 
-		throw new EngineRequestNotSupportedError(nodeExecuteContext.getNode().type);
+		throw new EngineRequestNotSupportedError(context.getNode().type);
 	}
 }

packages/@n8n/node-engine-compatibility/src/v1-workflow-converter.ts

@@ -1,6 +1,7 @@
 import type { GraphEdge, GraphNode, WorkflowGraph } from '@n8n/engine';
 import type { INode, INodeConnections, IWorkflowBase } from 'n8n-workflow';
 
+import { MAIN_CONNECTION_TYPE, MANUAL_TRIGGER_TYPE, SPLIT_IN_BATCHES_TYPE } from './constants';
 import {
 	UnsupportedConnectionTypeError,
 	UnsupportedCycleError,
@@ -10,10 +11,6 @@ import {
 } from './errors';
 import type { V1NodeStepConfig } from './types';
 
-const MANUAL_TRIGGER_TYPE = 'n8n-nodes-base.manualTrigger';
-const SPLIT_IN_BATCHES_TYPE = 'n8n-nodes-base.splitInBatches';
-const MAIN_CONNECTION_TYPE = 'main';
-
 /**
  * Common v1 trigger types that don't end in "Trigger". Non-exhaustive: combined
  * with the name heuristic below, it exists to reject unsupported triggers with a

packages/@n8n/node-engine-compatibility/vitest.config.ts

@@ -2,4 +2,5 @@ import { createVitestConfig } from '@n8n/vitest-config/node';
 
 export default createVitestConfig({
 	exclude: ['**/node_modules/**', '**/dist/**'],
+	setupFiles: ['./src/__tests__/setup-vm-evaluator.ts'],
 });

packages/workflow/src/expression.ts

@@ -289,6 +289,21 @@ export class Expression {
 		if (Expression.vmEvaluator) await Expression.vmEvaluator.release(this);
 	}
 
+	async withIsolate<T>(fn: () => Promise<T>): Promise<T> {
+		const acquired = await this.acquireIsolate();
+		try {
+			return await fn();
+		} finally {
+			if (acquired) {
+				try {
+					await this.releaseIsolate();
+				} catch (error) {
+					LoggerProxy.error('Failed to release expression isolate', { error });
+				}
+			}
+		}
+	}
+
 	/**
 	 * Dispose the VM evaluator and release resources.
 	 * Should be called during application shutdown or test teardown.

packages/workflow/src/workflow-expression.ts

@@ -278,4 +278,8 @@ export class WorkflowExpression {
 	async releaseIsolate(): Promise<void> {
 		await this.expression.releaseIsolate();
 	}
+
+	async withIsolate<T>(fn: () => Promise<T>): Promise<T> {
+		return await this.expression.withIsolate(fn);
+	}
 }

packages/workflow/test/expression-with-isolate.test.ts

@@ -0,0 +1,55 @@
+import { Expression } from '../src/expression';
+
+describe('Expression.withIsolate', () => {
+	const expression = new Expression('Europe/Berlin');
+
+	afterEach(() => {
+		vi.restoreAllMocks();
+	});
+
+	it('returns the result of fn', async () => {
+		vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true);
+		const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue();
+
+		await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe(
+			'result',
+		);
+		expect(release).toHaveBeenCalledTimes(1);
+	});
+
+	it('releases when fn throws, propagating the error', async () => {
+		vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true);
+		const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue();
+
+		await expect(
+			expression.withIsolate(async () => {
+				return await Promise.reject(new Error('boom from fn'));
+			}),
+		).rejects.toThrow('boom from fn');
+		expect(release).toHaveBeenCalledTimes(1);
+	});
+
+	it('does not release an isolate it did not acquire', async () => {
+		vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(false);
+		const release = vi.spyOn(expression, 'releaseIsolate').mockResolvedValue();
+
+		await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe(
+			'result',
+		);
+		expect(release).not.toHaveBeenCalled();
+	});
+
+	it('never masks the outcome of fn with a release failure', async () => {
+		vi.spyOn(expression, 'acquireIsolate').mockResolvedValue(true);
+		vi.spyOn(expression, 'releaseIsolate').mockRejectedValue(new Error('release boom'));
+
+		await expect(expression.withIsolate(async () => await Promise.resolve('result'))).resolves.toBe(
+			'result',
+		);
+		await expect(
+			expression.withIsolate(async () => {
+				return await Promise.reject(new Error('boom from fn'));
+			}),
+		).rejects.toThrow('boom from fn');
+	});
+});