feat(core): Answer lastNode webhooks from execution responses (#38959)

2026-10-11 22:50:06 +00:00 · 2026-09-23 08:59:11 +00:00
parent bc779f2216
commit dc3283da19
13 changed files with 739 additions and 20 deletions
@@ -1,6 +1,7 @@
 import { z } from 'zod';

 import { Config, Env } from '../decorators';
+import { positiveIntSchema } from '../schemas';

 /**
 * Floor for the CP → DP shared secret. Kept in step with the engine's identity
@@ -62,4 +63,12 @@ export class EngineConfig {
 	/** Where the engine dials the control plane server. Defaults to loopback; set it when that is not reachable. */
 	@Env('N8N_ENGINE_CONTROL_PLANE_BASE_URL')
 	controlPlaneBaseUrl: string = '';
+
+	/**
+	 * How long (in ms) the control plane waits for a run to answer before it
+	 * answers the webhook request itself. The response channel is at-most-once,
+	 * so a lost answer must not hold the request open forever.
+	 */
+	@Env('N8N_ENGINE_WEBHOOK_RESPONSE_TIMEOUT', positiveIntSchema.max(2147483647))
+	webhookResponseTimeout: number = 2 * 60 * 1000;
 }
@@ -3,6 +3,7 @@ import { EngineConfig, ExecutionsConfig } from '@n8n/config';
 import { Container } from '@n8n/di';

 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
+import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';

 import { EngineControlPlaneServer } from '../engine-control-plane-server';
 import { EngineDataPlaneClient } from '../engine-data-plane-client';
@@ -25,6 +26,7 @@ describe('EngineV2Module', () => {
 		runtime = mockInstance(EngineV2Runtime);
 		client = mockInstance(EngineDataPlaneClient);
 		controlPlaneServer = mockInstance(EngineControlPlaneServer);
+		mockInstance(EngineV2WebhookResponder);
 		Container.set(EngineDataPlaneProxyService, new EngineDataPlaneProxyService());

 		module = new EngineV2Module();
@@ -39,7 +39,7 @@ export class EngineV2Module implements ModuleInterface {
 		const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js');
 		await Container.get(EngineControlPlaneServer).start();

-		// Create both endpoints before the engine starts. A short run can answer
+		// Hand both endpoints over before the engine starts. A short run can answer
 		// before `startExecution` returns, and responses are not replayed.
 		const { InMemoryExecutionResponseChannel } = await import(
 			'./response-channel/in-memory-execution-response-channel.js'
@@ -50,6 +50,9 @@ export class EngineV2Module implements ModuleInterface {
 		const { InMemoryExecutionResponseReceiver } = await import(
 			'./response-channel/in-memory-execution-response-receiver.js'
 		);
+		const { EngineV2WebhookResponder } = await import(
+			'@/services/engine-v2-webhook-responder.service.js'
+		);
 		// In-memory for now because both planes share this process. Redis endpoints
 		// can use the same response contracts when the planes run separately.
 		const responseChannel = new InMemoryExecutionResponseChannel();
@@ -58,6 +61,7 @@ export class EngineV2Module implements ModuleInterface {
 			responseChannel,
 			Container.get(Logger).scoped('engine-v2'),
 		);
+		Container.get(EngineV2WebhookResponder).useReceiver(responseReceiver);
 		this.responseSender = responseSender;
 		this.responseReceiver = responseReceiver;

@@ -0,0 +1,181 @@
+import type { Logger } from '@n8n/backend-common';
+import type { EngineConfig } from '@n8n/config';
+import type { ExecutionResponse } from '@n8n/engine';
+import { mock } from 'vitest-mock-extended';
+
+import { createExecutionIdV2 } from '@/executions/execution-id';
+import type { ExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/execution-response-receiver';
+import {
+	EngineV2WebhookResponder,
+	MAX_PENDING_WEBHOOKS,
+} from '@/services/engine-v2-webhook-responder.service';
+
+const TIMEOUT_MS = 50_000;
+
+/**
+ * Stands in for the receiver. Only `src/modules/engine-v2/**` may import
+ * `@n8n/engine` at runtime, and the receiver's own suite covers the frame round
+ * trip; what matters here is what the responder does with a response.
+ *
+ * Handlers are per execution, as the real receiver's are. A response for
+ * another run reaches nobody.
+ */
+function fakeReceiver() {
+	const handlers = new Map<string, Array<(response: ExecutionResponse) => void>>();
+
+	return {
+		receiver: {
+			receive: (executionId: string, handler: (r: ExecutionResponse) => void) => {
+				const forExecution = handlers.get(executionId) ?? [];
+				handlers.set(executionId, [...forExecution, handler]);
+
+				return () => handlers.delete(executionId);
+			},
+		} as unknown as ExecutionResponseReceiver,
+		deliver: (response: ExecutionResponse) =>
+			handlers.get(response.executionId)?.forEach((h) => h(response)),
+	};
+}
+
+const endedResponse = (executionId: string, overrides: Record<string, unknown> = {}) => ({
+	type: 'ended' as const,
+	executionId,
+	workflowId: 'wf-1',
+	status: 'completed' as const,
+	lastStep: {
+		nodeId: 'a',
+		nodeName: 'A',
+		status: 'completed' as const,
+		outputs: [[{ json: { a: 1 } }]],
+	},
+	...overrides,
+});
+
+const newResponder = (timeoutMs = TIMEOUT_MS) =>
+	new EngineV2WebhookResponder(
+		mock<EngineConfig>({ webhookResponseTimeout: timeoutMs }),
+		mock<Logger>({ scoped: () => mock<Logger>() }),
+	);
+
+describe('EngineV2WebhookResponder', () => {
+	let deliver: (response: ExecutionResponse) => void;
+	let responder: EngineV2WebhookResponder;
+
+	beforeEach(() => {
+		const fake = fakeReceiver();
+		deliver = fake.deliver;
+		responder = newResponder();
+		responder.useReceiver(fake.receiver);
+	});
+
+	it('listens under the id the run is started with', () => {
+		const executionId = createExecutionIdV2();
+
+		expect(responder.waitForResponse(executionId).executionId).toBe(executionId);
+	});
+
+	it('refuses to listen before the host hands over a receiver', () => {
+		expect(() => newResponder().waitForResponse(createExecutionIdV2())).toThrow(
+			'without a receiver',
+		);
+	});
+
+	it('refuses a run once it listens for as many as it can hold', () => {
+		for (let i = 0; i < MAX_PENDING_WEBHOOKS; i++) {
+			responder.waitForResponse(createExecutionIdV2());
+		}
+
+		// Refused before dispatch, so no run starts that nothing can answer.
+		expect(() => responder.waitForResponse(createExecutionIdV2())).toThrow('Try again later');
+	});
+
+	it('listens again once an answered run releases its slot', () => {
+		const pending = Array.from({ length: MAX_PENDING_WEBHOOKS }, () =>
+			responder.waitForResponse(createExecutionIdV2()),
+		);
+
+		pending[0].release();
+
+		expect(() => responder.waitForResponse(createExecutionIdV2())).not.toThrow();
+	});
+
+	it('leaves an unrelated pending run unaffected', async () => {
+		const other = responder.waitForResponse(createExecutionIdV2());
+		const pending = responder.waitForResponse(createExecutionIdV2());
+
+		deliver(endedResponse(pending.executionId));
+
+		await expect(pending.settled).resolves.toMatchObject({ status: 'completed' });
+
+		const stillPending = Symbol('still pending');
+		await expect(Promise.race([other.settled, Promise.resolve(stillPending)])).resolves.toBe(
+			stillPending,
+		);
+	});
+
+	it('reports the step the run ended with', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2());
+
+		deliver(endedResponse(pending.executionId));
+
+		await expect(pending.settled).resolves.toEqual({
+			status: 'completed',
+			lastNode: { nodeName: 'A', outputs: [[{ json: { a: 1 } }]] },
+		});
+	});
+
+	it('reports no last node when that step produced nothing', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2());
+
+		deliver(
+			endedResponse(pending.executionId, {
+				lastStep: { nodeId: 'c', nodeName: 'C', status: 'skipped', outputs: null },
+			}),
+		);
+
+		await expect(pending.settled).resolves.toEqual({ status: 'completed', lastNode: undefined });
+	});
+
+	it('reports a failure with the node that caused it', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2());
+
+		deliver(
+			endedResponse(pending.executionId, {
+				status: 'failed',
+				lastStep: {
+					nodeId: 'c',
+					nodeName: 'C',
+					status: 'failed',
+					outputs: null,
+					error: { name: 'NodeOperationError', message: 'it broke' },
+				},
+			}),
+		);
+
+		await expect(pending.settled).resolves.toEqual({
+			status: 'failed',
+			nodeName: 'C',
+			error: { name: 'NodeOperationError', message: 'it broke' },
+		});
+	});
+
+	it('times out rather than waiting forever for a lost answer', async () => {
+		const impatient = newResponder(1);
+		impatient.useReceiver(fakeReceiver().receiver);
+
+		await expect(impatient.waitForResponse(createExecutionIdV2()).settled).resolves.toEqual({
+			status: 'timeout',
+		});
+	});
+
+	it('drops later responses for a released run', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2());
+		pending.release();
+
+		deliver(endedResponse(pending.executionId));
+
+		await expect(Promise.race([pending.settled, Promise.resolve('still waiting')])).resolves.toBe(
+			'still waiting',
+		);
+	});
+});
@@ -8,9 +8,10 @@ import type {
 	WorkflowExecuteMode,
 } from 'n8n-workflow';
 import { classifyTriggerIdentity, isTriggerNodeType, UserError } from 'n8n-workflow';
+import assert from 'node:assert';

 import { toWorkflowDocument } from '@/executions/execution-data/types';
-import { createExecutionIdV2 } from '@/executions/execution-id';
+import { createExecutionIdV2, isExecutionIdV2 } from '@/executions/execution-id';
 import { CredentialsPermissionChecker } from '@/executions/pre-execution-checks';
 import type { ResumableExecution } from '@/interfaces';
 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
@@ -82,7 +83,12 @@ export class EngineV2Dispatcher {
 		return workflowData.settings?.engineType === 'v2' && ROUTED_MODES.has(executionMode);
 	}

-	/** Returns the execution id this dispatch minted. */
+	/**
+	 * Returns the execution id this run uses.
+	 *
+	 * A caller that has to wait for the run's answer mints the id itself, so it
+	 * can subscribe before the run can produce one.
+	 */
 	async start(data: IWorkflowExecutionDataProcess): Promise<string> {
 		const trigger = this.resolveFiredTrigger(data);

@@ -98,7 +104,9 @@ export class EngineV2Dispatcher {

 		const graph = new V1WorkflowConverter().convert(workflowData, trigger.name);

-		const executionId = createExecutionIdV2();
+		const executionId = data.engineExecutionId ?? createExecutionIdV2();
+		// A caller that minted the id is waiting on that exact run.
+		assert(isExecutionIdV2(executionId), 'Engine 2.0 was given an id it cannot run');
 		// At the session cap this can evict another run's session, uncaught below. Rare; not worth fixing.
 		this.registerPushSession(executionId, data, trigger);

@@ -0,0 +1,122 @@
+import { Logger } from '@n8n/backend-common';
+import { EngineConfig } from '@n8n/config';
+import { Service } from '@n8n/di';
+import type { EndedMessage, ExecutionResponse } from '@n8n/engine';
+import { OperationalError, UnexpectedError } from 'n8n-workflow';
+
+import type { ExecutionIdV2 } from '@/executions/execution-id';
+import type {
+	ExecutionResponseReceiver,
+	UnsubscribeExecutionResponse,
+} from '@/modules/engine-v2/response-channel/execution-response-receiver';
+import { PendingWebhookResponse } from '@/services/pending-webhook-response';
+
+/** A request that is still open, and the subscription that feeds its answer. */
+type PendingWebhook = {
+	response: PendingWebhookResponse;
+	unsubscribe: UnsubscribeExecutionResponse;
+};
+
+/**
+ * How many runs this replica listens for at once. Every entry has its own
+ * response timeout, so the map drains on its own; this only bounds it.
+ */
+export const MAX_PENDING_WEBHOOKS = 5000;
+
+/**
+ * A service for hooking up responses from the data plane with webhook callers
+ * expecting that response.
+ *
+ * `waitForResponse` should be called before starting the execution in case a
+ * fast execution sends a response before we're listening for it.
+ */
+@Service()
+export class EngineV2WebhookResponder {
+	private receiver?: ExecutionResponseReceiver;
+
+	private readonly pendingWebhooks = new Map<string, PendingWebhook>();
+
+	constructor(
+		private readonly engineConfig: EngineConfig,
+		private readonly logger: Logger,
+	) {
+		this.logger = this.logger.scoped('engine-v2');
+	}
+
+	/** The host calls this once with the receiver for execution responses. */
+	useReceiver(receiver: ExecutionResponseReceiver): void {
+		if (this.receiver) {
+			throw new UnexpectedError('Engine 2.0 webhook response receiver is already set');
+		}
+		this.receiver = receiver;
+	}
+
+	/**
+	 * Call this before starting the execution. Otherwise, a fast execution can
+	 * finish before we are waiting for its response.
+	 *
+	 * @param executionId The caller must mint this ID, use it here, and pass it to
+	 * `StartExecution`.
+	 * @throws {UnexpectedError} If the execution response receiver is not set.
+	 * @throws {OperationalError} If the service is at capacity.
+	 */
+	waitForResponse(executionId: ExecutionIdV2): PendingWebhookResponse {
+		const { receiver } = this;
+		if (!receiver) {
+			throw new UnexpectedError('Engine 2.0 cannot wait for a response without a receiver');
+		}
+
+		if (this.pendingWebhooks.size >= MAX_PENDING_WEBHOOKS) {
+			throw new OperationalError(
+				`Engine 2.0 already awaits ${MAX_PENDING_WEBHOOKS} webhook responses. Try again later.`,
+			);
+		}
+
+		const response = new PendingWebhookResponse({
+			executionId,
+			timeoutMs: this.engineConfig.webhookResponseTimeout,
+			onRelease: (id) => this.release(id),
+		});
+		const unsubscribe = receiver.receive(executionId, (received) =>
+			this.handle(received, response),
+		);
+		this.pendingWebhooks.set(executionId, { response, unsubscribe });
+
+		return response;
+	}
+
+	private handle(received: ExecutionResponse, response: PendingWebhookResponse): void {
+		try {
+			this.onEnded(received, response);
+		} catch (error) {
+			this.logger.error('Failed to relay an engine 2.0 response', {
+				executionId: received.executionId,
+				type: received.type,
+				error,
+			});
+		}
+	}
+
+	private onEnded(received: EndedMessage, response: PendingWebhookResponse): void {
+		const { nodeName, outputs, error } = received.lastStep;
+
+		if (received.status === 'failed') {
+			// The step that ended a failed run is the one that failed, so its name
+			// and error are what the caller reports.
+			response.resolve({ status: 'failed', nodeName, error });
+			return;
+		}
+
+		response.resolve({
+			status: 'completed',
+			// A skipped or failed step carries nothing to answer with.
+			lastNode: outputs ? { nodeName, outputs } : undefined,
+		});
+	}
+
+	/** Ends the run's subscription: nothing more can arrive for it. */
+	private release(executionId: string): void {
+		this.pendingWebhooks.get(executionId)?.unsubscribe();
+		this.pendingWebhooks.delete(executionId);
+	}
+}
@@ -0,0 +1,56 @@
+import type { StepSlots } from '@n8n/engine';
+import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
+
+import type { ExecutionIdV2 } from '@/executions/execution-id';
+
+/** The last node that ran, as the `lastNode` response mode answers with. */
+export type LastNode = { nodeName: string; outputs: StepSlots };
+
+/** What a data-plane run produced, from the open request's point of view. */
+export type WebhookRunOutcome =
+	| { status: 'completed'; lastNode?: LastNode }
+	| { status: 'failed'; nodeName: string; error?: { name: string; message: string } }
+	| { status: 'timeout' };
+
+export type PendingWebhookResponseOptions = {
+	/** The run whose answer this handle carries. */
+	executionId: ExecutionIdV2;
+	/** How long to hold the request before it is answered without the run. */
+	timeoutMs: number;
+	/** Called once the answer is no longer needed, so the listener can go. */
+	onRelease: (executionId: string) => void;
+};
+
+/**
+ * This represents a pending webhook response. The webhook responder connects it
+ * to the execution responses coming from the engine.
+ */
+export class PendingWebhookResponse {
+	/** Resolves when the run ends, or when the hold runs out — whichever is first. */
+	readonly settled: Promise<WebhookRunOutcome>;
+
+	readonly executionId: ExecutionIdV2;
+
+	private readonly onRelease: (executionId: string) => void;
+
+	private readonly answer = createDeferredPromise<WebhookRunOutcome>();
+
+	/** Held so an answered run does not leave a timer behind for the whole hold. */
+	private readonly timer: NodeJS.Timeout;
+
+	constructor({ executionId, timeoutMs, onRelease }: PendingWebhookResponseOptions) {
+		this.executionId = executionId;
+		this.onRelease = onRelease;
+		this.settled = this.answer.promise;
+		this.timer = setTimeout(() => this.answer.resolve({ status: 'timeout' }), timeoutMs).unref();
+	}
+
+	resolve(outcome: WebhookRunOutcome): void {
+		this.answer.resolve(outcome);
+	}
+
+	release(): void {
+		clearTimeout(this.timer);
+		this.onRelease(this.executionId);
+	}
+}
@@ -0,0 +1,75 @@
+import type { StepSlots } from '@n8n/engine';
+import { WorkflowOperationError } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
+import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
+import { EngineV2PayloadGuard } from '@/services/engine-v2-payload-guard.service';
+import { EngineV2Webhooks } from '@/webhooks/engine-v2-webhooks';
+
+const engineV2Webhooks = new EngineV2Webhooks(
+	mock<EngineV2Dispatcher>(),
+	mock<EngineV2PayloadGuard>(),
+	mock<EngineDataPlaneProxyService>(),
+);
+
+describe('EngineV2Webhooks.toRun', () => {
+	it('converts a completed outcome with outputs', async () => {
+		const outputs: StepSlots = [[{ json: { answer: 42 } }], null];
+
+		const run = await engineV2Webhooks.toRun(
+			{ status: 'completed', lastNode: { nodeName: 'Last node', outputs } },
+			'webhook',
+		);
+
+		expect(run).toMatchObject({
+			mode: 'webhook',
+			status: 'success',
+			storedAt: 'db',
+			data: {
+				resultData: {
+					lastNodeExecuted: 'Last node',
+					runData: {
+						'Last node': [
+							{
+								executionIndex: 0,
+								executionStatus: 'success',
+								source: [],
+								executionTime: 0,
+								data: { main: [[{ json: { answer: 42 } }], []] },
+							},
+						],
+					},
+				},
+			},
+		});
+		expect(run.startedAt).toBeInstanceOf(Date);
+		expect(run.data.resultData.runData['Last node'][0].startTime).toEqual(expect.any(Number));
+	});
+
+	it('converts a failed outcome', async () => {
+		const run = await engineV2Webhooks.toRun(
+			{
+				status: 'failed',
+				nodeName: 'Broken node',
+				error: { name: 'Error', message: 'The node failed' },
+			},
+			'webhook',
+		);
+
+		expect(run.status).toBe('error');
+		expect(run.data.resultData.lastNodeExecuted).toBe('Broken node');
+		expect(run.data.resultData.runData).toEqual({});
+		expect(run.data.resultData.error).toBeInstanceOf(WorkflowOperationError);
+		expect(run.data.resultData.error?.message).toBe('The node failed');
+	});
+
+	it('converts a completed outcome with no last node data', async () => {
+		const run = await engineV2Webhooks.toRun({ status: 'completed' }, 'webhook');
+
+		expect(run.status).toBe('success');
+		expect(run.data.resultData.lastNodeExecuted).toBeUndefined();
+		expect(run.data.resultData.runData).toEqual({});
+		expect(run.data.resultData.error).toBeUndefined();
+	});
+});
@@ -3,6 +3,7 @@ import { mockInstance } from '@n8n/backend-test-utils';
 import type { Project, User } from '@n8n/db';
 import { UserRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
+import type { EndedMessage, ExecutionResponse } from '@n8n/engine';
 import type express from 'express';
 import {
 	BinaryDataService,
@@ -12,6 +13,8 @@ import {
 	isWebhookHtmlSandboxingDisabled,
 } from 'n8n-core';

+import type { ExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/execution-response-receiver';
+
 vi.mock('n8n-core', async () => ({
 	...(await vi.importActual<typeof import('n8n-core')>('n8n-core')),
 	isWebhookHtmlSandboxingDisabled: vi.fn(),
@@ -58,7 +61,9 @@ import { AuthService } from '@/auth/auth.service';
 import type { ResponseError } from '@/errors/response-errors/abstract/response.error';
 import { EventService } from '@/events/event.service';
 import { WebhookResponseRelay } from '@/scaling/webhook-response-relay';
+import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
 import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
+import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
 import { OwnershipService } from '@/services/ownership.service';
 import type { ProtectedResource } from '@/services/protected-resource.registry';
 import { ProtectedResourceRegistry } from '@/services/protected-resource.registry';
@@ -1127,6 +1132,7 @@ const activeExecutions = mockInstance(ActiveExecutions);
 const resourceRegistry = mockInstance(ProtectedResourceRegistry);
 // Off by default: every block but the engine 2.0 one exercises the v1 path.
 const engineV2Dispatcher = mockInstance(EngineV2Dispatcher);
+const engineDataPlaneProxy = mockInstance(EngineDataPlaneProxyService);
 const executionContextService = mockInstance(ExecutionContextService);
 const userRepository = mockInstance(UserRepository);
 mockInstance(AuthService);
@@ -1944,6 +1950,8 @@ describe('executeWebhook in responseNode mode when the Respond node never runs',

 describe('executeWebhook on engine 2.0', () => {
 	const errorReporter = Container.get(ErrorReporter);
+	/** Response handlers registered by the responder, by execution ID. */
+	let dataPlane: Map<string, (response: ExecutionResponse) => void>;
 	/** The data plane mints uuidv7 ids, not the numeric ids v1 uses. */
 	const ENGINE_EXECUTION_ID = '019606a1-0000-7000-8000-000000000001';

@@ -2034,7 +2042,34 @@ describe('executeWebhook on engine 2.0', () => {
 		// The webhook path asks before it can build the run data, so it goes through
 		// `handlesWorkflow`, not `routesToEngineV2`.
 		engineV2Dispatcher.handlesWorkflow.mockReturnValue(true);
+		engineDataPlaneProxy.isAvailable.mockReturnValue(true);
 		workflowRunner.run.mockResolvedValue(ENGINE_EXECUTION_ID);
+		dataPlane = new Map();
+	});
+
+	/** Ends the run the request is waiting on, as the data plane would. */
+	const answerRun = (status: EndedMessage['status'], lastStep: EndedMessage['lastStep']): void => {
+		const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string;
+		dataPlane.get(executionId)?.({
+			type: 'ended',
+			executionId,
+			workflowId: WORKFLOW_ID,
+			status,
+			lastStep,
+		});
+	};
+
+	beforeAll(() => {
+		// The host hands the responder its receiver at boot. Keeping each run's
+		// handler is how a test plays the data plane answering.
+		Container.get(EngineV2WebhookResponder).useReceiver(
+			mock<ExecutionResponseReceiver>({
+				receive: vi.fn((executionId: string, handler: (r: ExecutionResponse) => void) => {
+					dataPlane.set(executionId, handler);
+					return vi.fn();
+				}),
+			}),
+		);
 	});

 	describe('onReceived', () => {
@@ -2065,6 +2100,93 @@ describe('executeWebhook on engine 2.0', () => {
 		});
 	});

+	describe('lastNode', () => {
+		it('dispatches the run under the id the answer is listened for on', async () => {
+			await startWebhook({ responseMode: 'lastNode' });
+
+			// The listener is created before the run starts, and the run has to use
+			// the id it listens under, so a fast answer is not lost.
+			expect(workflowRunner.run).toHaveBeenCalledTimes(1);
+			const dispatchedId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string;
+			expect(dataPlane.has(dispatchedId)).toBe(true);
+		});
+
+		it('answers with the data of the step the run ended on', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'lastNode' });
+
+			answerRun('completed', {
+				nodeId: 'edit-fields',
+				nodeName: 'Edit Fields',
+				status: 'completed',
+				outputs: [[{ json: { ok: true } }]],
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback).toHaveBeenCalledWith(
+				null,
+				expect.objectContaining({ body: [{ ok: true }], code: 200 }),
+			);
+		});
+
+		it('answers that nothing was returned when the step produced no data', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'lastNode' });
+
+			answerRun('completed', {
+				nodeId: 'edit-fields',
+				nodeName: 'Edit Fields',
+				status: 'skipped',
+				outputs: null,
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback.mock.calls[0]).toEqual([
+				null,
+				{
+					data: {
+						message: 'Workflow executed successfully but the last node did not return any data',
+					},
+					responseCode: 200,
+				},
+			]);
+		});
+
+		it('answers with a failure when the run ended on a failed step', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'lastNode' });
+
+			answerRun('failed', {
+				nodeId: 'edit-fields',
+				nodeName: 'Edit Fields',
+				status: 'failed',
+				outputs: null,
+				error: { name: 'NodeOperationError', message: 'it broke' },
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback.mock.calls[0]).toEqual([
+				null,
+				{ data: { message: 'Error in workflow' }, responseCode: 500 },
+			]);
+		});
+
+		it('answers with a timeout when the run does not send an ended message', async () => {
+			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			const { responseCallback } = await startWebhook({ responseMode: 'lastNode' });
+			const pending = waitForResponse.mock.results[0]?.value;
+
+			expect(pending).toBeDefined();
+			pending?.resolve({ status: 'timeout' });
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback.mock.calls[0]).toEqual([
+				null,
+				{
+					data: { message: 'The workflow did not answer in time' },
+					responseCode: 504,
+				},
+			]);
+		});
+	});
+
 	describe('rejections', () => {
 		const reasonFrom = (responseCallback: ReturnType<typeof vi.fn>) => {
 			const error = responseCallback.mock.calls[0][0] as ResponseError;
@@ -2072,12 +2194,6 @@ describe('executeWebhook on engine 2.0', () => {
 		};

 		it.each([
-			{
-				name: 'the lastNode response mode',
-				options: { responseMode: 'lastNode' },
-				message:
-					"Engine 2.0 does not support the 'lastNode' response mode yet. Respond immediately instead.",
-			},
 			{
 				name: 'the responseNode response mode',
 				options: { responseMode: 'responseNode' },
@@ -2142,6 +2258,19 @@ describe('executeWebhook on engine 2.0', () => {
 			expect(workflowRunner.run).not.toHaveBeenCalled();
 		});

+		it('answers 400, not 500, when the engine-v2 module is disabled', async () => {
+			engineDataPlaneProxy.isAvailable.mockReturnValue(false);
+
+			const { responseCallback } = await startWebhook();
+
+			expect(reasonFrom(responseCallback)).toEqual({
+				status: 400,
+				message:
+					'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.',
+			});
+			expect(workflowRunner.run).not.toHaveBeenCalled();
+		});
+
 		it('answers 400 for a file upload, and deletes what the node already stored', async () => {
 			const binaryDataService = Container.get(BinaryDataService);
 			const { responseCallback } = await startWebhook({
@@ -1,6 +1,8 @@
 import { Service } from '@n8n/di';
 import type {
 	INode,
+	IRun,
+	IRunData,
 	IWebhookResponseData,
 	IWorkflowBase,
 	WebhookResponseMode,
@@ -9,16 +11,20 @@ import type {
 import {
 	CHAT_TRIGGER_NODE_TYPE,
 	classifyTriggerIdentity,
+	createRunExecutionData,
 	FORM_NODE_TYPE,
 	FORM_TRIGGER_NODE_TYPE,
 	MICROSOFT_AGENT365_TRIGGER_NODE_TYPE,
 	UserError,
 	WAIT_NODE_TYPE,
+	WorkflowOperationError,
 } from 'n8n-workflow';

 import { MCP_TRIGGER_NODE_TYPE } from '@/constants';
+import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
 import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
 import { EngineV2PayloadGuard } from '@/services/engine-v2-payload-guard.service';
+import type { WebhookRunOutcome } from '@/services/pending-webhook-response';

 /**
 * Trigger types the v2 path cannot serve. Each carries machinery the engine
@@ -34,6 +40,13 @@ const UNSUPPORTED_TRIGGERS = new Set<string>([
 	WAIT_NODE_TYPE,
 ]);

+/**
+ * Response modes the v2 path serves. Each mode is added here as its support
+ * lands, so a mode that is not ready yet fails with a reason rather than
+ * answering wrongly.
+ */
+const SUPPORTED_RESPONSE_MODES = new Set<WebhookResponseMode>(['onReceived', 'lastNode']);
+
 /** What the request says about a run, before the webhook node has produced anything. */
 export type EngineV2WebhookRequest = {
 	workflowStartNode: INode;
@@ -54,6 +67,7 @@ export class EngineV2Webhooks {
 	constructor(
 		private readonly dispatcher: EngineV2Dispatcher,
 		private readonly payloadGuard: EngineV2PayloadGuard,
+		private readonly proxy: EngineDataPlaneProxyService,
 	) {}

 	/** Whether this webhook run starts on the engine 2.0 data plane. */
@@ -77,6 +91,15 @@ export class EngineV2Webhooks {
 	 * Ordered so the user hears the most fundamental reason first.
 	 */
 	assertSupported({ workflowStartNode, responseMode, executionId }: EngineV2WebhookRequest): void {
+		// Checked first: `EngineV2WebhookResponder.waitForResponse` assumes the module
+		// registered its channel, and throws an internal error otherwise. Only a check
+		// that precedes that call can turn "module off" into a 400 instead of a 500.
+		if (!this.proxy.isAvailable()) {
+			throw new UserError(
+				'Engine 2.0 is not available. Enable the `engine-v2` module with N8N_ENABLED_MODULES.',
+			);
+		}
+
 		// A v2 run keeps no control-plane execution row, so there is nothing to resume.
 		if (executionId !== undefined) {
 			throw new UserError('Engine 2.0 cannot resume a waiting execution yet.');
@@ -99,14 +122,56 @@ export class EngineV2Webhooks {
 			);
 		}

-		// TODO(CAT-4313): support `lastNode`. TODO(CAT-4079): support `responseNode`.
-		if (responseMode !== 'onReceived') {
+		// TODO(CAT-4079): Support `responseNode`.
+		if (!SUPPORTED_RESPONSE_MODES.has(responseMode)) {
 			throw new UserError(
 				`Engine 2.0 does not support the '${responseMode}' response mode yet. Respond immediately instead.`,
 			);
 		}
 	}

+	/** Converts the data plane's answer to the shape the v1 response path reads. */
+	async toRun(
+		outcome: Exclude<WebhookRunOutcome, { status: 'timeout' }>,
+		executionMode: WorkflowExecuteMode,
+	): Promise<IRun> {
+		const runData: IRunData = {};
+		let lastNodeExecuted = outcome.status === 'failed' ? outcome.nodeName : undefined;
+
+		if (outcome.status === 'completed' && outcome.lastNode) {
+			const { fromStepInputs } = await import('@n8n/node-engine-compatibility');
+			lastNodeExecuted = outcome.lastNode.nodeName;
+			runData[lastNodeExecuted] = [
+				{
+					startTime: Date.now(),
+					executionIndex: 0,
+					source: [],
+					executionTime: 0,
+					executionStatus: 'success',
+					data: { main: fromStepInputs(outcome.lastNode.outputs) },
+				},
+			];
+		}
+
+		return {
+			mode: executionMode,
+			startedAt: new Date(),
+			status: outcome.status === 'failed' ? 'error' : 'success',
+			// The data plane holds the run. This object never reaches a store.
+			storedAt: 'db',
+			data: createRunExecutionData({
+				resultData: {
+					runData,
+					lastNodeExecuted,
+					error:
+						outcome.status === 'failed'
+							? new WorkflowOperationError(outcome.error?.message ?? 'The workflow failed')
+							: undefined,
+				},
+			}),
+		};
+	}
+
 	/**
 	 * Rejects a payload the engine cannot carry.
 	 *
@@ -72,6 +72,7 @@ import { InternalServerError } from '@/errors/response-errors/internal-server.er
 import { NotFoundError } from '@/errors/response-errors/not-found.error';
 import { UnsupportedMediaTypeError } from '@/errors/response-errors/unsupported-media-type.error';
 import { EventService } from '@/events/event.service';
+import { createExecutionIdV2 } from '@/executions/execution-id';
 import { parseBody } from '@/middlewares';
 import { WebhookResponseRelay } from '@/scaling/webhook-response-relay';
 import {
@@ -81,6 +82,8 @@ import {
 import { OAuth2FlowProxy } from '@/services/oauth2-flow-proxy.service';
 import { OwnershipService } from '@/services/ownership.service';
 import { ProtectedResourceRegistry } from '@/services/protected-resource.registry';
+import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
+import type { PendingWebhookResponse } from '@/services/pending-webhook-response';
 import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
 import { WaitTracker } from '@/wait-tracker';
 import { EXECUTION_ENDED_WITHOUT_RESPONSE } from '@/webhooks/constants';
@@ -835,6 +838,7 @@ export async function executeWebhook(
 	let didSendResponse = false;
 	/** Whether this run goes to the engine 2.0 data plane instead of the v1 path. */
 	let routesToEngineV2 = false;
+	let pendingEngineV2Response: PendingWebhookResponse | undefined;
 	let runExecutionDataMerge = {};
 	const engineV2Webhooks = Container.get(EngineV2Webhooks);
 	let cleanupMultipartFiles: (() => Promise<void>) | undefined;
@@ -1105,6 +1109,16 @@ export async function executeWebhook(
 			didSendResponse = true;
 		}

+		// Before the run, because a short workflow answers before `startExecution`
+		// returns and nothing replays a missed response. The id is minted here, so
+		// the run and the listener agree on it.
+		if (routesToEngineV2 && responseMode !== 'onReceived') {
+			const engineExecutionId = createExecutionIdV2();
+			pendingEngineV2Response =
+				Container.get(EngineV2WebhookResponder).waitForResponse(engineExecutionId);
+			runData.engineExecutionId = engineExecutionId;
+		}
+
 		// Extract W3C trace context from webhook headers for OTEL propagation.
 		const traceparent = req.headers.traceparent;
 		if (
@@ -1195,14 +1209,46 @@ export async function executeWebhook(
 			{ executionId },
 		);

-		// Engine 2.0 serves `onReceived` only, so the response is already out. Nothing
-		// below applies: the run has no control-plane execution to wait on.
-		if (routesToEngineV2) return executionId;
+		if (routesToEngineV2 && responseMode === 'onReceived') return executionId;

-		const activeExecutions = Container.get(ActiveExecutions);
+		/**
+		 * A callback to handle the pending execution response from the data plane.
+		 * Returns the run data if we get a response in time. Otherwise sends a 504
+		 * timeout to the webhook caller.
+		 */
+		const waitForDataPlaneRun = async (waiting: PendingWebhookResponse) => {
+			try {
+				const outcome = await waiting.settled;
+				if (outcome.status !== 'timeout') {
+					return await engineV2Webhooks.toRun(outcome, executionMode);
+				}

-		// Get a promise which resolves when the workflow did execute and send then response
-		const executePromise = activeExecutions.getPostExecutePromise(executionId);
+				Container.get(Logger).warn('No answer arrived for an engine 2.0 webhook run', {
+					executionId,
+					workflowId: workflowData.id,
+				});
+				// The webhook node can answer before the execution starts. Do not send a
+				// second response if the execution response later times out.
+				if (!didSendResponse) {
+					responseCallback(null, {
+						data: { message: 'The workflow did not answer in time' },
+						responseCode: 504,
+					});
+					didSendResponse = true;
+				}
+
+				return undefined;
+			} finally {
+				waiting.release();
+			}
+		};
+
+		// Get a promise which resolves when the workflow did execute and send then response.
+		// Engine 2.0 keeps no control-plane execution to wait on, so its answer comes
+		// off the response channel, shaped like the run the handler below reads.
+		const executePromise = pendingEngineV2Response
+			? waitForDataPlaneRun(pendingEngineV2Response)
+			: Container.get(ActiveExecutions).getPostExecutePromise(executionId);

 		const { parentExecution } = runExecutionData;
 		if (WorkflowHelpers.shouldRestartParentExecution(parentExecution)) {
@@ -1332,6 +1378,9 @@ export async function executeWebhook(
 		}
 		return executionId;
 	} catch (e) {
+		// Nothing will ever answer this one, so stop waiting for it.
+		pendingEngineV2Response?.release();
+
 		let error: Error;
 		if (e instanceof ResponseError && e.httpStatusCode < 500) {
 			error = e;
@@ -6,6 +6,7 @@
 * assert what reaches the data plane.
 */

+import { Logger } from '@n8n/backend-common';
 import { createWorkflow, mockInstance, testDb } from '@n8n/backend-test-utils';
 import { GlobalConfig } from '@n8n/config';
 import { UUID_V7_PATTERN } from '@n8n/constants';
@@ -17,7 +18,10 @@ import { randomUUID } from 'node:crypto';
 import { agent as testAgent } from 'supertest';

 import { CacheService } from '@/services/cache/cache.service';
+import { InMemoryExecutionResponseChannel } from '@/modules/engine-v2/response-channel/in-memory-execution-response-channel';
+import { InMemoryExecutionResponseReceiver } from '@/modules/engine-v2/response-channel/in-memory-execution-response-receiver';
 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
+import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
 import { Telemetry } from '@/telemetry';
 import { WebhookServer } from '@/webhooks/webhook-server';

@@ -73,6 +77,13 @@ beforeAll(async () => {
 		getExecution,
 		searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }),
 	});
+	// The host hands the responder its receiver at boot (`EngineV2Module.init`).
+	// This test drives the webhook route directly, without the module, so it
+	// wires the same receiver by hand.
+	const responseChannel = new InMemoryExecutionResponseChannel();
+	Container.get(EngineV2WebhookResponder).useReceiver(
+		new InMemoryExecutionResponseReceiver(responseChannel, Container.get(Logger)),
+	);

 	// `/webhook-test/*` is mounted only when a server opts into test webhooks.
 	class EditorFacingWebhookServer extends WebhookServer {
@@ -134,7 +145,7 @@ describe('webhook runs on engine 2.0', () => {
 	test('answers 400 with the reason when the response mode is unsupported', async () => {
 		const webhookId = randomUUID();
 		const trigger = webhookNode(webhookId);
-		trigger.parameters.responseMode = 'lastNode';
+		trigger.parameters.responseMode = 'responseNode';
 		const workflow = await createV2Workflow(trigger);

 		await startListening(workflow.id);
@@ -144,7 +155,9 @@ describe('webhook runs on engine 2.0', () => {
 			.send({ order: 42 });

 		expect(response.statusCode).toBe(400);
-		expect(response.body.message).toContain("does not support the 'lastNode' response mode yet");
+		expect(response.body.message).toContain(
+			"does not support the 'responseNode' response mode yet",
+		);
 		expect(startExecution).not.toHaveBeenCalled();
 	});
 });
@@ -3767,6 +3767,12 @@ export interface IWorkflowExecutionDataProcess {
 	agentRequest?: AiAgentRequest;
 	httpResponse?: express.Response; // Used for streaming responses
 	streamingEnabled?: boolean;
+	/**
+	 * Only engine 2.0 reads this. The caller mints the data-plane execution id
+	 * when it has to wait for the run's answer, so it can subscribe before the
+	 * run starts.
+	 */
+	engineExecutionId?: string;
 	startedAt?: Date;

 	// MCP-specific fields for queue mode support