feat(core): Support the responseNode webhook response mode on engine 2.0 (no-changelog) (#39003)

packages/@n8n/engine/package.json

@@ -28,6 +28,7 @@
     "@n8n/constants": "workspace:*",
     "@n8n/di": "workspace:*",
     "@n8n/typeorm": "workspace:*",
+    "@n8n/utils": "workspace:*",
     "express": "catalog:",
     "jsonwebtoken": "catalog:",
     "pg": "catalog:",

packages/@n8n/engine/src/dependencies/external-dependencies.ts

@@ -1,6 +1,7 @@
 import type { CallerContext, ExecutionMode, StepSlots, WaitDeclaration } from '../execution';
 import type { GraphNode } from '../graph';
 import type { LifecycleEventCallback } from '../lifecycle-events';
+import type { ResponseEmitter } from '../response-channel';
 
 /**
  * Host integration seam — how the engine reaches capabilities it does not own.
@@ -42,6 +43,11 @@ export interface StepExecutionRequest {
 	/** Input slots gathered from predecessor steps; slot contents are opaque. */
 	inputs: StepSlots;
 	context: StepExecutionContext;
+	/**
+	 * Response emitter allows the step executor to send messages to the control
+	 * plane side via the execution response channel.
+	 */
+	respond: ResponseEmitter;
 }
 
 /**

packages/@n8n/engine/src/execution/__tests__/step-execution.integration.test.ts

@@ -546,6 +546,7 @@ describe('step execution (integration)', () => {
 			orchestrationQueue,
 			{ v1StepExecutor: { execute } },
 			noopLifecycleEventPublisher,
+			noopExecutionResponseSender,
 		);
 
 		const executionId = generateId();

packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest';
 import type { ExternalDependencies, IStepExecutor } from '../../dependencies';
 import { deriveLoops, type WorkflowGraph } from '../../graph';
 import type { LifecycleEventPublisher, LifecycleEvent } from '../../lifecycle-events';
+import { noopExecutionResponseSender, type ExecutionResponseSender } from '../../response-channel';
 import type { OrchestrationMessage, WorkQueue } from '../../queue';
 import type { ExecutionRecord, ExecutionStore } from '../execution-store';
 import {
@@ -56,6 +57,7 @@ function makeHandler(
 	queue: WorkQueue<OrchestrationMessage>,
 	dependencies: ExternalDependencies,
 	lifecycleEventPublisher: LifecycleEventPublisher = makeLifecycleEventPublisher(),
+	responseSender: ExecutionResponseSender = noopExecutionResponseSender,
 	onStepSuspended?: () => void,
 ): StepReadyHandler {
 	return new StepReadyHandler(
@@ -64,6 +66,7 @@ function makeHandler(
 		queue,
 		dependencies,
 		lifecycleEventPublisher,
+		responseSender,
 		onStepSuspended,
 	);
 }
@@ -170,6 +173,8 @@ describe('StepReadyHandler', () => {
 				iteration: 0,
 				callerContext: { hostMode: 'trigger' },
 			},
+			// The step can answer the caller while it runs.
+			respond: { send: expect.any(Function) },
 		});
 		expect(stepStore.completeStep).toHaveBeenCalledWith('step-a', [[{ json: { ok: true } }]]);
 		expect(stepStore.failStep).not.toHaveBeenCalled();
@@ -757,6 +762,7 @@ describe('StepReadyHandler waits', () => {
 			makeQueue(),
 			{ v1StepExecutor: makeExecutor({ wait: timeWait }) },
 			makeLifecycleEventPublisher(),
+			noopExecutionResponseSender,
 			onStepSuspended,
 		);
 
@@ -775,6 +781,7 @@ describe('StepReadyHandler waits', () => {
 			makeQueue(),
 			{ v1StepExecutor: makeExecutor({ wait: timeWait }) },
 			makeLifecycleEventPublisher(),
+			noopExecutionResponseSender,
 			onStepSuspended,
 		);
 

packages/@n8n/engine/src/execution/step-ready-handler.ts

@@ -8,6 +8,7 @@ import {
 	type WorkflowLoop,
 } from '../graph';
 import type { LifecycleEventPublisher } from '../lifecycle-events';
+import type { ExecutionResponseSender } from '../response-channel';
 import { runBatchStep } from './batch-step';
 import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue';
 import type { ExecutionRecord, ExecutionStore } from './execution-store';
@@ -49,6 +50,7 @@ export class StepReadyHandler {
 		private readonly orchestrationQueue: WorkQueue<OrchestrationMessage>,
 		private readonly dependencies: ExternalDependencies,
 		private readonly lifecycleEventPublisher: LifecycleEventPublisher,
+		private readonly responseSender: ExecutionResponseSender,
 		/** Called once a wait is on the row, so the sweeper can re-arm on its deadline. */
 		private readonly onStepSuspended: () => void = () => {},
 	) {}
@@ -179,6 +181,7 @@ export class StepReadyHandler {
 				iteration: step.iteration,
 				callerContext: execution.callerContext,
 			},
+			respond: this.responseSender.emitterFor(execution.id),
 		});
 
 		if (result.wait) validateWaitDeclaration(step.id, result.wait);

packages/@n8n/engine/src/index.ts

@@ -37,11 +37,14 @@ export type {
 export {
 	executionResponseSchema,
 	noopExecutionResponseSender,
+	noopResponseEmitter,
 } from './response-channel';
 export type {
 	EndedMessage,
 	ExecutionResponse,
 	ExecutionResponseSender,
+	ResponseEmitter,
+	ResponseMessage,
 	UndeliverableMessage,
 } from './response-channel';
 

packages/@n8n/engine/src/response-channel/execution-response-sender.ts

@@ -1,13 +1,21 @@
-import type { ExecutionResponse } from './execution-response.types';
+import { createResultOk, type Result } from '@n8n/utils/result';
+
+import {
+	noopResponseEmitter,
+	type ExecutionResponse,
+	type ResponseEmitter,
+} from './execution-response.types';
 
 /** Sends responses from an execution back to its caller. */
 export interface ExecutionResponseSender {
-	send(response: ExecutionResponse): void;
+	send(response: ExecutionResponse): Result<void, Error>;
+	emitterFor(executionId: string): ResponseEmitter;
 	stop(): Promise<void>;
 }
 
 /** Response sender for a host that discards execution responses. */
 export const noopExecutionResponseSender: ExecutionResponseSender = Object.freeze({
-	send: () => {},
+	send: () => createResultOk(undefined),
+	emitterFor: (_executionId: string) => noopResponseEmitter,
 	stop: async () => {},
 });

packages/@n8n/engine/src/response-channel/execution-response.schema.ts

@@ -18,6 +18,12 @@ export const executionResponseSchema = z.discriminatedUnion('type', [
 			message: z.string().min(1),
 		}),
 	}),
+	z.object({
+		type: z.literal('response'),
+		executionId: z.string().min(1),
+		// A missing or undefined payload means an empty response.
+		payload: jsonValueSchema.optional(),
+	}),
 	z.object({
 		type: z.literal('ended'),
 		executionId: z.string().min(1),

packages/@n8n/engine/src/response-channel/execution-response.types.ts

@@ -1,3 +1,4 @@
+import { createResultOk, type Result } from '@n8n/utils/result';
 import type { z } from 'zod';
 
 import type { executionResponseSchema } from './execution-response.schema';
@@ -28,3 +29,16 @@ export type UndeliverableMessage = Extract<ExecutionResponse, { type: 'undeliver
  * to look further.
  */
 export type EndedMessage = Extract<ExecutionResponse, { type: 'ended' }>;
+
+/** A message that a step produced a response */
+export type ResponseMessage = Extract<ExecutionResponse, { type: 'response' }>;
+
+/** An emitter that allows the step execution to produce a response */
+export interface ResponseEmitter {
+	send(payload: unknown): Result<void, Error>;
+}
+
+/** For a step whose responses nobody wants. */
+export const noopResponseEmitter: ResponseEmitter = Object.freeze({
+	send: () => createResultOk(undefined),
+});

packages/@n8n/engine/src/response-channel/index.ts

@@ -1,8 +1,11 @@
 export { noopExecutionResponseSender } from './execution-response-sender';
 export type { ExecutionResponseSender } from './execution-response-sender';
 export { executionResponseSchema } from './execution-response.schema';
+export { noopResponseEmitter } from './execution-response.types';
 export type {
 	EndedMessage,
 	ExecutionResponse,
+	ResponseEmitter,
+	ResponseMessage,
 	UndeliverableMessage,
 } from './execution-response.types';

packages/@n8n/engine/src/runtime/create-engine-runtime.ts

@@ -114,6 +114,7 @@ export function createEngineRuntime({
 			orchestrationQueue,
 			dependencies,
 			lifecycleEventPublisher,
+			responseSender,
 			// A deadline set after the sweeper armed would otherwise wait for its next pass.
 			() => waitSweeper.noteSuspended(),
 		),

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

@@ -1,4 +1,5 @@
 import type { JsonObject, StepExecutionRequest, StepSlots, WorkflowGraph } from '@n8n/engine';
+import { noopResponseEmitter } from '@n8n/engine';
 import type { ExecuteContext } from 'n8n-core';
 import { UnrecognizedNodeTypeError } from 'n8n-core';
 import { NoOp } from 'n8n-nodes-base/nodes/NoOp/NoOp.node';
@@ -261,6 +262,7 @@ export const stepRequest = (
 		iteration: 0,
 		callerContext: { hostMode: 'manual' },
 	},
+	respond: noopResponseEmitter,
 });
 
 export const testStepExecutor = (

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

@@ -0,0 +1,82 @@
+import type { ResponseEmitter, StepExecutionRequest } from '@n8n/engine';
+import { ExecutionLifecycleHooks } from 'n8n-core';
+import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
+import { describe, expect, it, vi } from 'vitest';
+
+import { attachResponseHooks } from '../v1-response-hooks';
+
+const newRequest = () => {
+	const respond: ResponseEmitter = {
+		send: vi.fn(() => ({ ok: true as const, result: undefined })),
+	};
+	const request = {
+		context: {
+			executionId: 'exec-1',
+			stepId: 'step-1',
+			workflowId: 'wf-1',
+			mode: 'production',
+			iteration: 0,
+			callerContext: { hostMode: 'webhook' },
+		},
+		respond,
+	} as unknown as StepExecutionRequest;
+
+	return { request, respond };
+};
+
+const newAdditionalData = () => ({}) as IWorkflowExecuteAdditionalData;
+
+describe('attachResponseHooks', () => {
+	it('preserves hooks supplied by the host', async () => {
+		const { request, respond } = newRequest();
+		const existingHandler = vi.fn();
+		const hooks = new ExecutionLifecycleHooks('webhook', 'exec-1', {} as IWorkflowBase);
+		hooks.addHandler('sendChunk', existingHandler);
+		const additionalData = { ...newAdditionalData(), hooks };
+		const response = { body: { ok: true }, statusCode: 200 };
+
+		attachResponseHooks(additionalData, request);
+		await additionalData.hooks.runHook('sendResponse', [response]);
+
+		expect(additionalData.hooks).toBe(hooks);
+		expect(additionalData.hooks.handlers.sendChunk).toEqual([existingHandler]);
+		expect(respond.send).toHaveBeenCalledWith(response);
+	});
+
+	it('sends what the Respond node produced to the channel', async () => {
+		const { request, respond } = newRequest();
+		const additionalData = newAdditionalData();
+
+		attachResponseHooks(additionalData, request);
+		await additionalData.hooks?.runHook('sendResponse', [{ body: { ok: true }, statusCode: 200 }]);
+
+		expect(respond.send).toHaveBeenCalledWith({ body: { ok: true }, statusCode: 200 });
+	});
+
+	it('surfaces errors from the response channel', async () => {
+		const { request, respond } = newRequest();
+		const error = new Error('Response failed');
+		vi.mocked(respond.send).mockReturnValue({ ok: false, error });
+		const additionalData = newAdditionalData();
+
+		attachResponseHooks(additionalData, request);
+		await expect(
+			additionalData.hooks?.runHook('sendResponse', [{ body: { ok: true }, statusCode: 200 }]),
+		).rejects.toBe(error);
+		expect(respond.send).toHaveBeenCalledOnce();
+	});
+
+	it.each([
+		['a buffer body', { body: Buffer.from('hi') }],
+		['a stream body', { body: { pipe: () => {} } }],
+		['a binary reference', { body: { binaryData: { id: 'file-1' } } }],
+	])('refuses %s, which has no JSON form', async (_name, response) => {
+		const { request } = newRequest();
+		const additionalData = newAdditionalData();
+		attachResponseHooks(additionalData, request);
+
+		await expect(additionalData.hooks?.runHook('sendResponse', [response])).rejects.toThrow(
+			/Engine 2.0 cannot/,
+		);
+	});
+});

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

@@ -0,0 +1,81 @@
+import type { JsonValue, StepExecutionRequest } from '@n8n/engine';
+import { ExecutionLifecycleHooks } from 'n8n-core';
+import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
+import { UserError } from 'n8n-workflow';
+
+import { toV1ExecuteMode } from './v1-adapters';
+
+/**
+ * `ExecutionLifecycleHooks` takes a workflow, but `sendResponse` and `sendChunk`
+ * never read one, and this layer holds a `Workflow` instance rather than the
+ * `IWorkflowBase` the constructor wants.
+ */
+function stubWorkflow(id: string): IWorkflowBase {
+	const now = new Date();
+	return {
+		id,
+		name: '',
+		active: false,
+		isArchived: false,
+		activeVersionId: null,
+		createdAt: now,
+		updatedAt: now,
+		nodes: [],
+		connections: {},
+	};
+}
+
+/**
+ * Glues the v1 `sendResponse` hook to the v2 execution response channel.
+ * Allows the node execution to send a response to the control plane.
+ */
+export function attachResponseHooks(
+	additionalData: IWorkflowExecuteAdditionalData,
+	request: StepExecutionRequest,
+): void {
+	const { context, respond } = request;
+	additionalData.hooks ??= new ExecutionLifecycleHooks(
+		toV1ExecuteMode(context),
+		context.executionId,
+		stubWorkflow(context.workflowId),
+	);
+
+	additionalData.hooks.addHandler('sendResponse', (response) => {
+		const result = respond.send(toJsonPayload(response));
+		if (!result.ok) {
+			// Fail the node so the caller can see the response error.
+			throw result.error;
+		}
+	});
+}
+
+/**
+ * A response crosses a process boundary as JSON, so anything without a JSON
+ * form fails here rather than arriving silently mangled.
+ */
+function toJsonPayload(value: unknown): JsonValue {
+	assertCarriable(value);
+
+	// Match Engine 1 queue mode. Nested Buffers and streams are not supported.
+	if (typeof value === 'object' && value !== null && 'body' in value) {
+		assertCarriable(value.body);
+	}
+
+	return value as JsonValue;
+}
+
+function assertCarriable(value: unknown): void {
+	if (Buffer.isBuffer(value)) {
+		throw new UserError('Engine 2.0 cannot send a binary webhook response yet.');
+	}
+
+	if (typeof value === 'object' && value !== null) {
+		if (typeof (value as { pipe?: unknown }).pipe === 'function') {
+			throw new UserError('Engine 2.0 cannot stream a webhook response body yet.');
+		}
+
+		if ('binaryData' in value) {
+			throw new UserError('Engine 2.0 cannot send a binary webhook response yet.');
+		}
+	}
+}

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

@@ -17,6 +17,7 @@ import {
 } from './errors';
 import { isV1NodeStepConfig } from './guards';
 import { fromStepInputs, toStepOutputs } from './io';
+import { attachResponseHooks } from './v1-response-hooks';
 import type {
 	ExecutableNodeType,
 	NodeRunResult,
@@ -61,6 +62,7 @@ export class V1StepExecutor implements IStepExecutor {
 		const additionalData = await this.deps.additionalDataFactory(
 			toAdditionalDataContext(request.context),
 		);
+		attachResponseHooks(additionalData, request);
 
 		const context = toV1ExecuteContext({
 			node,

packages/cli/src/modules/engine-v2/__tests__/in-memory-execution-response.test.ts

@@ -69,6 +69,19 @@ describe('in-memory execution responses', () => {
 		);
 	});
 
+	it('stamps the execution ID on a response from a step', () => {
+		const channel = new InMemoryExecutionResponseChannel();
+		const publish = vi.spyOn(channel, 'publish');
+		const sender = new InMemoryExecutionResponseSender(channel, mockLogger());
+
+		sender.emitterFor('exec-1').send({ ok: true });
+
+		expect(publish).toHaveBeenCalledExactlyOnceWith(
+			'exec-1',
+			JSON.stringify({ type: 'response', executionId: 'exec-1', payload: { ok: true } }),
+		);
+	});
+
 	it('validates a response before delivering it', () => {
 		const channel = new InMemoryExecutionResponseChannel();
 		const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());

packages/cli/src/modules/engine-v2/response-channel/in-memory-execution-response-sender.ts

@@ -1,12 +1,21 @@
 import type { Logger } from '@n8n/backend-common';
-import type { ExecutionResponse, ExecutionResponseSender, UndeliverableMessage } from '@n8n/engine';
-import { toResult } from '@n8n/utils/result';
+import type {
+	ExecutionResponse,
+	ExecutionResponseSender,
+	JsonValue,
+	ResponseEmitter,
+	UndeliverableMessage,
+} from '@n8n/engine';
+import { createResultError, createResultOk, toResult, type Result } from '@n8n/utils/result';
+import { UnexpectedError } from 'n8n-workflow';
 
 import type { InMemoryExecutionResponseChannel } from './in-memory-execution-response-channel';
 
 /** A response channel must be able to carry every frame this class produces. */
 const MAX_FRAME_BYTES = 5 * 1024 * 1024;
 
+type FrameResult = { ok: true; frame: string } | { ok: false; frame: string; error: Error };
+
 export class InMemoryExecutionResponseSender implements ExecutionResponseSender {
 	private stopped = false;
 
@@ -16,17 +25,29 @@ export class InMemoryExecutionResponseSender implements ExecutionResponseSender
 		private readonly maxFrameBytes: number = MAX_FRAME_BYTES,
 	) {}
 
-	send(response: ExecutionResponse): void {
-		if (this.stopped) return;
+	send(response: ExecutionResponse): Result<void, Error> {
+		if (this.stopped) {
+			return createResultError(new UnexpectedError('The execution response sender has stopped.'));
+		}
 
-		this.channel.publish(response.executionId, this.toFrame(response));
+		const frameResult = this.toFrame(response);
+		this.channel.publish(response.executionId, frameResult.frame);
+
+		return frameResult.ok ? createResultOk(undefined) : createResultError(frameResult.error);
+	}
+
+	/** Gives one step a response sender without exposing execution routing. */
+	emitterFor(executionId: string): ResponseEmitter {
+		return {
+			send: (payload: JsonValue) => this.send({ type: 'response', executionId, payload }),
+		};
 	}
 
 	async stop(): Promise<void> {
 		this.stopped = true;
 	}
 
-	private toFrame(response: ExecutionResponse): string {
+	private toFrame(response: ExecutionResponse): FrameResult {
 		const serialized = toResult(() => JSON.stringify(response));
 		if (!serialized.ok) {
 			this.logger.warn('Could not serialize an execution response', {
@@ -51,14 +72,27 @@ export class InMemoryExecutionResponseSender implements ExecutionResponseSender
 			});
 		}
 
-		return serialized.result;
+		return { ok: true, frame: serialized.result };
 	}
 
-	private undeliverableFrame(executionId: string, error: UndeliverableMessage['error']): string {
-		return JSON.stringify({
+	private undeliverableFrame(
+		executionId: string,
+		error: UndeliverableMessage['error'],
+	): FrameResult {
+		const frame = JSON.stringify({
 			type: 'undeliverable',
 			executionId,
 			error,
 		} satisfies UndeliverableMessage);
+
+		return {
+			ok: false,
+			frame,
+			error: new UnexpectedError(error.message, {
+				extra: {
+					code: error.code,
+				},
+			}),
+		};
 	}
 }

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

@@ -136,6 +136,47 @@ describe('EngineV2WebhookResponder', () => {
 		await expect(pending.settled).resolves.toEqual({ status: 'completed', lastNode: undefined });
 	});
 
+	it('reports the response produced by the Respond node', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2(), true);
+
+		deliver({
+			type: 'response',
+			executionId: pending.executionId,
+			payload: { body: { ok: true }, headers: {}, statusCode: 200 },
+		});
+
+		await expect(pending.settled).resolves.toEqual({
+			status: 'response',
+			response: { body: { ok: true }, headers: {}, statusCode: 200 },
+		});
+	});
+
+	it('keeps the first terminal outcome', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2(), true);
+
+		deliver({
+			type: 'response',
+			executionId: pending.executionId,
+			payload: { body: { ok: true }, headers: {}, statusCode: 200 },
+		});
+		deliver(endedResponse(pending.executionId));
+
+		await expect(pending.settled).resolves.toMatchObject({ status: 'response' });
+	});
+
+	it('ignores a Respond node result when the response mode waits for the last node', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2());
+
+		deliver({
+			type: 'response',
+			executionId: pending.executionId,
+			payload: { body: { ignored: true }, headers: {}, statusCode: 200 },
+		});
+		deliver(endedResponse(pending.executionId));
+
+		await expect(pending.settled).resolves.toMatchObject({ status: 'completed' });
+	});
+
 	it('reports a failure with the node that caused it', async () => {
 		const pending = responder.waitForResponse(createExecutionIdV2());
 

packages/cli/src/services/engine-v2-webhook-responder.service.ts

@@ -60,7 +60,7 @@ export class EngineV2WebhookResponder {
 	 * @throws {UnexpectedError} If the execution response receiver is not set.
 	 * @throws {OperationalError} If the service is at capacity.
 	 */
-	waitForResponse(executionId: ExecutionIdV2): PendingWebhookResponse {
+	waitForResponse(executionId: ExecutionIdV2, acceptsResponse = false): PendingWebhookResponse {
 		const { receiver } = this;
 		if (!receiver) {
 			throw new UnexpectedError('Engine 2.0 cannot wait for a response without a receiver');
@@ -74,6 +74,7 @@ export class EngineV2WebhookResponder {
 
 		const response = new PendingWebhookResponse({
 			executionId,
+			acceptsResponse,
 			timeoutMs: this.engineConfig.webhookResponseTimeout,
 			onRelease: (id) => this.release(id),
 		});
@@ -87,15 +88,7 @@ export class EngineV2WebhookResponder {
 
 	private handle(received: ExecutionResponse, response: PendingWebhookResponse): void {
 		try {
-			if (received.type === 'undeliverable') {
-				response.resolve({
-					status: 'undeliverable',
-					error: { name: received.error.code, message: received.error.message },
-				});
-				return;
-			}
-
-			this.onEnded(received, response);
+			this.route(received, response);
 		} catch (error) {
 			this.logger.error('Failed to relay an engine 2.0 response', {
 				executionId: received.executionId,
@@ -105,6 +98,25 @@ export class EngineV2WebhookResponder {
 		}
 	}
 
+	private route(received: ExecutionResponse, response: PendingWebhookResponse): void {
+		switch (received.type) {
+			case 'undeliverable':
+				response.resolve({
+					status: 'undeliverable',
+					error: { name: received.error.code, message: received.error.message },
+				});
+				return;
+
+			case 'response':
+				response.resolveResponse(received.payload);
+				return;
+
+			case 'ended':
+				this.onEnded(received, response);
+				return;
+		}
+	}
+
 	private onEnded(received: EndedMessage, response: PendingWebhookResponse): void {
 		const { nodeName, outputs, error } = received.lastStep;
 

packages/cli/src/services/pending-webhook-response.ts

@@ -8,6 +8,7 @@ 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: 'response'; response: unknown }
 	| { status: 'completed'; lastNode?: LastNode }
 	| { status: 'failed'; nodeName: string; error?: { name: string; message: string } }
 	/** The execution response could not be produced or delivered. */
@@ -17,6 +18,8 @@ export type WebhookRunOutcome =
 export type PendingWebhookResponseOptions = {
 	/** The run whose answer this handle carries. */
 	executionId: ExecutionIdV2;
+	/** Whether a Respond to Webhook result answers this request. */
+	acceptsResponse: boolean;
 	/** 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. */
@@ -28,11 +31,13 @@ export type PendingWebhookResponseOptions = {
  * 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. */
+	/** Resolves when the request gets an answer, or when the hold runs out. */
 	readonly settled: Promise<WebhookRunOutcome>;
 
 	readonly executionId: ExecutionIdV2;
 
+	private readonly acceptsResponse: boolean;
+
 	private readonly onRelease: (executionId: string) => void;
 
 	private readonly answer = createDeferredPromise<WebhookRunOutcome>();
@@ -40,8 +45,14 @@ export class PendingWebhookResponse {
 	/** 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) {
+	constructor({
+		executionId,
+		acceptsResponse,
+		timeoutMs,
+		onRelease,
+	}: PendingWebhookResponseOptions) {
 		this.executionId = executionId;
+		this.acceptsResponse = acceptsResponse;
 		this.onRelease = onRelease;
 		this.settled = this.answer.promise;
 		this.timer = setTimeout(() => this.answer.resolve({ status: 'timeout' }), timeoutMs).unref();
@@ -51,6 +62,10 @@ export class PendingWebhookResponse {
 		this.answer.resolve(outcome);
 	}
 
+	resolveResponse(response: unknown): void {
+		if (this.acceptsResponse) this.answer.resolve({ status: 'response', response });
+	}
+
 	release(): void {
 		clearTimeout(this.timer);
 		this.onRelease(this.executionId);

packages/cli/src/webhooks/__tests__/webhook-helpers.test.ts

@@ -2356,6 +2356,94 @@ describe('executeWebhook on engine 2.0', () => {
 		});
 	});
 
+	describe('responseNode', () => {
+		it('answers with the response published by the data plane', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
+			const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string;
+
+			dataPlane.get(executionId)?.({
+				type: 'response',
+				executionId,
+				payload: { body: { ok: true }, headers: {}, statusCode: 200 },
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(workflowRunner.run).toHaveBeenCalledTimes(1);
+			expect(responseCallback).toHaveBeenCalledWith(null, {
+				data: { ok: true },
+				headers: {},
+				responseCode: 200,
+			});
+		});
+
+		it('answers with an empty body when the response has no body key', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
+			const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string;
+
+			dataPlane.get(executionId)?.({
+				type: 'response',
+				executionId,
+				payload: { headers: { location: 'https://example.com' }, statusCode: 307 },
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback).toHaveBeenCalledWith(null, {
+				data: undefined,
+				headers: { location: 'https://example.com' },
+				responseCode: 307,
+			});
+		});
+
+		it('answers with an empty body when the Respond node never runs', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
+
+			answerRun('completed', {
+				nodeId: 'edit-fields',
+				nodeName: 'Edit Fields',
+				status: 'completed',
+				outputs: [[{ json: { ignored: true } }]],
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback).toHaveBeenCalledWith(null, {
+				data: undefined,
+				responseCode: 200,
+			});
+		});
+
+		it('answers with the channel error when the response is undeliverable', async () => {
+			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
+			const executionId = workflowRunner.run.mock.calls[0][0].engineExecutionId as string;
+
+			dataPlane.get(executionId)?.({
+				type: 'undeliverable',
+				executionId,
+				error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+			});
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback).toHaveBeenCalledWith(null, {
+				data: { message: 'The response is too large.' },
+				responseCode: 500,
+			});
+		});
+
+		it('answers with a timeout when no terminal outcome arrives', async () => {
+			const waitForResponse = vi.spyOn(Container.get(EngineV2WebhookResponder), 'waitForResponse');
+			const { responseCallback } = await startWebhook({ responseMode: 'responseNode' });
+			const pending = waitForResponse.mock.results[0]?.value;
+
+			expect(pending).toBeDefined();
+			pending?.resolve({ status: 'timeout' });
+
+			await vi.waitFor(() => expect(responseCallback).toHaveBeenCalledTimes(1));
+			expect(responseCallback).toHaveBeenCalledWith(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;
@@ -2363,12 +2451,6 @@ describe('executeWebhook on engine 2.0', () => {
 		};
 
 		it.each([
-			{
-				name: 'the responseNode response mode',
-				options: { responseMode: 'responseNode' },
-				message:
-					"Engine 2.0 does not support the 'responseNode' response mode yet. Respond immediately instead.",
-			},
 			{
 				name: 'the streaming response mode',
 				options: { responseMode: 'streaming' },

packages/cli/src/webhooks/engine-v2-webhooks.ts

@@ -45,7 +45,11 @@ const UNSUPPORTED_TRIGGERS = new Set<string>([
  * 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']);
+const SUPPORTED_RESPONSE_MODES = new Set<WebhookResponseMode>([
+	'onReceived',
+	'lastNode',
+	'responseNode',
+]);
 
 /** What the request says about a run, before the webhook node has produced anything. */
 export type EngineV2WebhookRequest = {
@@ -122,7 +126,6 @@ export class EngineV2Webhooks {
 			);
 		}
 
-		// 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.`,
@@ -132,7 +135,7 @@ export class EngineV2Webhooks {
 
 	/** Converts the data plane's answer to the shape the v1 response path reads. */
 	async toRun(
-		outcome: Exclude<WebhookRunOutcome, { status: 'timeout' | 'undeliverable' }>,
+		outcome: Exclude<WebhookRunOutcome, { status: 'response' | 'timeout' | 'undeliverable' }>,
 		executionMode: WorkflowExecuteMode,
 	): Promise<IRun> {
 		const runData: IRunData = {};

packages/cli/src/webhooks/webhook-helpers.ts

@@ -4,6 +4,7 @@ import type { Project } from '@n8n/db';
 import { UserRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
+import { isRecord } from '@n8n/utils/is-record';
 import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
 import type express from 'express';
 import merge from 'lodash/merge';
@@ -417,6 +418,76 @@ export const handleFormRedirectionCase = (
 const { formDataFileSizeMax } = Container.get(GlobalConfig).endpoints;
 const parseFormData = createMultiFormDataParser(formDataFileSizeMax);
 
+function isHttpFullResponse(response: unknown): response is IN8nHttpFullResponse {
+	// A missing body means that the response has an empty body.
+	return (
+		isRecord(response) && isRecord(response.headers) && typeof response.statusCode === 'number'
+	);
+}
+
+async function sendResponseNodeResponse(
+	response: IN8nHttpFullResponse,
+	res: express.Response,
+	responseCallback: (error: Error | null, data: IWebhookResponseCallbackData) => void,
+	workflowStartNode: INode,
+	executionId: string | undefined,
+	workflow: Workflow,
+): Promise<void> {
+	try {
+		const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData;
+		if (binaryData?.id) {
+			if (response.statusCode) {
+				res.status(response.statusCode);
+			}
+			WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res);
+			applySandboxCSP(res);
+			try {
+				const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id);
+				res.once('close', () => stream.destroy());
+				stream.pipe(res, { end: false });
+				await finished(stream);
+			} finally {
+				void Container.get(WebhookResponseRelay).deleteOffloadedBody(response, {
+					workflowId: workflow.id,
+					executionId,
+				});
+			}
+			responseCallback(null, { noWebhookResponse: true });
+		} else if (Buffer.isBuffer(response.body)) {
+			if (response.statusCode) {
+				res.status(response.statusCode);
+			}
+			WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res);
+			applySandboxCSP(res);
+			res.end(response.body);
+			responseCallback(null, { noWebhookResponse: true });
+		} else {
+			// TODO: This probably needs some more changes depending on the options on the
+			//       Webhook Response node
+
+			let data: IWebhookResponseCallbackData = {
+				data: response.body as IDataObject,
+				headers: response.headers,
+				responseCode: response.statusCode,
+			};
+
+			data = handleFormRedirectionCase(data, workflowStartNode);
+
+			responseCallback(null, data);
+		}
+
+		process.nextTick(() => res.end());
+	} catch (e: unknown) {
+		const error = ensureError(e);
+		Container.get(ErrorReporter).error(error);
+		Container.get(Logger).error(
+			`Error with Webhook-Response for execution "${executionId}": "${error.message}"`,
+			{ executionId, workflowId: workflow.id },
+		);
+		responseCallback(error, {});
+	}
+}
+
 export function setupResponseNodePromise(
 	responsePromise: IDeferredPromise<IN8nHttpFullResponse>,
 	res: express.Response,
@@ -426,7 +497,7 @@ export function setupResponseNodePromise(
 	workflow: Workflow,
 ): void {
 	void responsePromise.promise
-		.then(async (response: IN8nHttpFullResponse) => {
+		.then(async (response) => {
 			if (response === EXECUTION_ENDED_WITHOUT_RESPONSE) {
 				// The execution ended without the Respond to Webhook node running. The
 				// post-execute handler answers instead, because only it knows whether the
@@ -434,49 +505,14 @@ export function setupResponseNodePromise(
 				return;
 			}
 
-			const binaryData = (response.body as IDataObject)?.binaryData as IBinaryData;
-			if (binaryData?.id) {
-				if (response.statusCode) {
-					res.status(response.statusCode);
-				}
-				WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res);
-				applySandboxCSP(res);
-				try {
-					const stream = await Container.get(BinaryDataService).getAsStream(binaryData.id);
-					res.once('close', () => stream.destroy());
-					stream.pipe(res, { end: false });
-					await finished(stream);
-				} finally {
-					void Container.get(WebhookResponseRelay).deleteOffloadedBody(response, {
-						workflowId: workflow.id,
-						executionId,
-					});
-				}
-				responseCallback(null, { noWebhookResponse: true });
-			} else if (Buffer.isBuffer(response.body)) {
-				if (response.statusCode) {
-					res.status(response.statusCode);
-				}
-				WebhookResponseHeaders.fromObject(response.headers).applyToResponse(res);
-				applySandboxCSP(res);
-				res.end(response.body);
-				responseCallback(null, { noWebhookResponse: true });
-			} else {
-				// TODO: This probably needs some more changes depending on the options on the
-				//       Webhook Response node
-
-				let data: IWebhookResponseCallbackData = {
-					data: response.body as IDataObject,
-					headers: response.headers,
-					responseCode: response.statusCode,
-				};
-
-				data = handleFormRedirectionCase(data, workflowStartNode);
-
-				responseCallback(null, data);
-			}
-
-			process.nextTick(() => res.end());
+			await sendResponseNodeResponse(
+				response,
+				res,
+				responseCallback,
+				workflowStartNode,
+				executionId,
+				workflow,
+			);
 		})
 		.catch(async (e: unknown) => {
 			const error = ensureError(e);
@@ -1076,7 +1112,7 @@ export async function executeWebhook(
 		if (didPublishMcpRelay) return undefined;
 
 		let responsePromise: IDeferredPromise<IN8nHttpFullResponse> | undefined;
-		if (responseMode === 'responseNode') {
+		if (responseMode === 'responseNode' && !routesToEngineV2) {
 			responsePromise = createDeferredPromise<IN8nHttpFullResponse>();
 			// Mark the request as answered as soon as the node produces a response, before
 			// `setupResponseNodePromise` starts writing it. Streaming offloaded binary data
@@ -1121,8 +1157,10 @@ export async function executeWebhook(
 		// the run and the listener agree on it.
 		if (routesToEngineV2 && responseMode !== 'onReceived') {
 			const engineExecutionId = createExecutionIdV2();
-			pendingEngineV2Response =
-				Container.get(EngineV2WebhookResponder).waitForResponse(engineExecutionId);
+			pendingEngineV2Response = Container.get(EngineV2WebhookResponder).waitForResponse(
+				engineExecutionId,
+				responseMode === 'responseNode',
+			);
 			runData.engineExecutionId = engineExecutionId;
 		}
 
@@ -1229,7 +1267,30 @@ export async function executeWebhook(
 		const waitForDataPlaneRun = async (waiting: PendingWebhookResponse) => {
 			try {
 				const outcome = await waiting.settled;
+				if (outcome.status === 'response') {
+					if (!isHttpFullResponse(outcome.response)) {
+						throw new UnexpectedError('Engine 2.0 produced an invalid webhook response');
+					}
+
+					didSendResponse = true;
+					await sendResponseNodeResponse(
+						outcome.response,
+						res,
+						responseCallback,
+						workflowStartNode,
+						executionId,
+						workflow,
+					);
+					return undefined;
+				}
+
 				if (outcome.status === 'completed' || outcome.status === 'failed') {
+					if (outcome.status === 'completed' && responseMode === 'responseNode') {
+						responseCallback(null, { data: undefined, responseCode });
+						didSendResponse = true;
+						return undefined;
+					}
+
 					return await engineV2Webhooks.toRun(outcome, executionMode);
 				}
 
@@ -1325,7 +1386,7 @@ export async function executeWebhook(
 						return runData;
 					}
 
-					// in `responseNode` mode `responseCallback` is called by `responsePromise`
+					// In v1 `responseNode` mode, `responsePromise` calls `responseCallback`.
 					if (responseMode === 'responseNode' && responsePromise) {
 						await Promise.allSettled([responsePromise.promise]);
 						if (!didSendResponse) {

packages/cli/test/integration/engine-v2-webhooks.test.ts

@@ -145,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 = 'responseNode';
+		trigger.parameters.responseMode = 'streaming';
 		const workflow = await createV2Workflow(trigger);
 
 		await startListening(workflow.id);
@@ -155,9 +155,7 @@ describe('webhook runs on engine 2.0', () => {
 			.send({ order: 42 });
 
 		expect(response.statusCode).toBe(400);
-		expect(response.body.message).toContain(
-			"does not support the 'responseNode' response mode yet",
-		);
+		expect(response.body.message).toContain("does not support the 'streaming' response mode yet");
 		expect(startExecution).not.toHaveBeenCalled();
 	});
 });

pnpm-lock.yaml

@@ -2539,6 +2539,9 @@ importers:
       '@n8n/typeorm':
         specifier: workspace:*
         version: link:../typeorm
+      '@n8n/utils':
+        specifier: workspace:*
+        version: link:../utils
       express:
         specifier: 'catalog:'
         version: 5.1.0(supports-color@8.1.1)