feat(engine): Stream webhook responses from engine v2 on the control plane (no-changelog) (#39948)

packages/@n8n/config/src/configs/engine.config.ts

@@ -82,7 +82,9 @@ export class EngineConfig {
 	/**
 	 * 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.
+	 * so a lost answer must not hold the request open forever. For a streaming
+	 * request, each chunk starts the wait again, so this limits the silence
+	 * between chunks, not the length of the stream.
 	 */
 	@Env('N8N_ENGINE_WEBHOOK_RESPONSE_TIMEOUT', positiveIntSchema.max(2147483647))
 	webhookResponseTimeout: number = 2 * 60 * 1000;

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

@@ -1762,8 +1762,15 @@ describe('pre-persist context establishment', () => {
 });
 
 describe('streaming functionality', () => {
-	it('should setup heartbeat interval and sendChunk handler when streaming is enabled', async () => {
+	type StreamingResponse = Response & { flush: () => void };
+
+	afterEach(() => {
+		vi.useRealTimers();
+	});
+
+	it('should setup a v1 heartbeat interval and sendChunk handler when streaming is enabled', async () => {
 		// ARRANGE
+		vi.useFakeTimers();
 		const activeExecutions = Container.get(ActiveExecutions);
 		vi.spyOn(activeExecutions, 'add').mockResolvedValue('1');
 		vi.spyOn(activeExecutions, 'attachWorkflowExecution').mockReturnValueOnce();
@@ -1771,7 +1778,7 @@ describe('streaming functionality', () => {
 		const permissionChecker = Container.get(CredentialsPermissionChecker);
 		vi.spyOn(permissionChecker, 'check').mockResolvedValueOnce();
 
-		const mockResponse = mock<Response>({ writableEnded: false });
+		const mockResponse = mock<StreamingResponse>({ writableEnded: false });
 		const mockSetInterval = vi.spyOn(global, 'setInterval');
 
 		const data = mock<IWorkflowExecutionDataProcess>({
@@ -1804,6 +1811,24 @@ describe('streaming functionality', () => {
 		// sendChunk handler is still registered on lifecycle hooks
 		expect(mockHooks.addHandler).toHaveBeenCalledWith('sendChunk', expect.any(Function));
 
+		const closeHandler = mockResponse.once.mock.calls.find(([event]) => event === 'close')?.[1];
+		closeHandler?.();
 		mockSetInterval.mockRestore();
 	});
+
+	it('does not manage the response lifecycle for a v2 run', async () => {
+		const dispatcher = Container.get(EngineV2Dispatcher);
+		vi.spyOn(dispatcher, 'routesToEngineV2').mockReturnValueOnce(true);
+		vi.spyOn(dispatcher, 'start').mockResolvedValueOnce('dp-uuid');
+		const setIntervalSpy = vi.spyOn(global, 'setInterval');
+
+		await runner.run(
+			mock<IWorkflowExecutionDataProcess>({
+				streamingEnabled: true,
+				httpResponse: mock<StreamingResponse>({ writableEnded: false }),
+			}),
+		);
+
+		expect(setIntervalSpy).not.toHaveBeenCalled();
+	});
 });

packages/cli/src/modules/engine-v2/__tests__/webhook-outcome.test.ts

@@ -60,6 +60,7 @@ describe('toWebhookOutcome', () => {
 
 		expect(outcome).toEqual({
 			status: 'failed',
+			nodeId: 'c',
 			nodeName: 'C',
 			error: { name: 'NodeOperationError', message: 'it broke' },
 		});
@@ -110,7 +111,7 @@ describe('toWebhookOutcome', () => {
 		expect(Buffer.isBuffer((outcome as { response: { body: unknown } }).response.body)).toBe(true);
 	});
 
-	it.each([runEnd, { kind: 'none' } as const])(
+	it.each([runEnd, { kind: 'stream' } as const, { kind: 'none' } as const])(
 		'ignores a Respond node result when the expectation is %j',
 		(expectation) => {
 			const payload = { body: { ignored: true }, headers: {}, statusCode: 200 };

packages/cli/src/modules/engine-v2/__tests__/webhook-response-registry.service.test.ts

@@ -11,11 +11,13 @@ import {
 	MAX_PENDING_WEBHOOKS,
 	SUBSCRIBE_TIMEOUT_MS,
 } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
+import type { ResponseStream } from '@/webhooks/streaming-webhook-response-heartbeat';
 
 const TIMEOUT_MS = 50_000;
 
 const runEnd = { kind: 'runEnd' } as const;
 const stepResponse = { kind: 'stepResponse' } as const;
+const stream = { kind: 'stream' } as const;
 
 /**
  * Stands in for the receiver. Only `src/modules/engine-v2/**` may import
@@ -63,6 +65,10 @@ const newRegistry = (timeoutMs = TIMEOUT_MS) =>
 		mock<Logger>({ scoped: () => mock<Logger>() }),
 	);
 
+const stillPending = Symbol('still pending');
+const raceWithPending = async (outcome: Promise<unknown>) =>
+	await Promise.race([outcome, Promise.resolve(stillPending)]);
+
 describe('EngineV2WebhookResponseRegistry', () => {
 	let deliver: (response: ExecutionResponse) => void;
 	let registry: EngineV2WebhookResponseRegistry;
@@ -79,10 +85,6 @@ describe('EngineV2WebhookResponseRegistry', () => {
 		vi.useRealTimers();
 	});
 
-	const stillPending = Symbol('still pending');
-	const raceWithPending = async (outcome: Promise<unknown>) =>
-		await Promise.race([outcome, Promise.resolve(stillPending)]);
-
 	it('listens under the id the run is started with', async () => {
 		const executionId = createExecutionIdV2();
 
@@ -230,6 +232,319 @@ describe('EngineV2WebhookResponseRegistry', () => {
 	});
 });
 
+describe('EngineV2WebhookResponseRegistry with a streaming request', () => {
+	let deliver: (response: ExecutionResponse) => void;
+	let registry: EngineV2WebhookResponseRegistry;
+
+	beforeEach(() => {
+		vi.useFakeTimers();
+		const fake = fakeReceiver();
+		deliver = fake.deliver;
+		registry = newRegistry();
+		registry.useReceiver(fake.receiver);
+	});
+
+	afterEach(() => {
+		vi.useRealTimers();
+	});
+
+	const failedStep = {
+		status: 'failed',
+		lastStep: {
+			nodeId: 'c',
+			nodeName: 'C',
+			status: 'failed',
+			outputs: null,
+			error: { name: 'NodeOperationError', message: 'it broke' },
+		},
+	};
+
+	const openStream = () => mock<ResponseStream>({ writableEnded: false, destroyed: false });
+
+	const waitForStream = async (responseStream: ResponseStream) =>
+		await registry.waitForResponse(createExecutionIdV2(), stream, responseStream);
+
+	it('refuses to wait for a stream without a response stream', async () => {
+		await expect(registry.waitForResponse(createExecutionIdV2(), stream)).rejects.toThrow(
+			'without a response stream',
+		);
+	});
+
+	it.each([runEnd, stepResponse, { kind: 'none' } as const])(
+		'does not write chunks when the expectation is %j',
+		async (expectation) => {
+			const responseStream = openStream();
+			const pending = await registry.waitForResponse(
+				createExecutionIdV2(),
+				expectation,
+				responseStream,
+			);
+
+			deliver({
+				type: 'chunk',
+				executionId: pending.executionId,
+				payload: { type: 'item', content: 'ignored' },
+			});
+			deliver(endedResponse(pending.executionId));
+			await pending.outcome;
+
+			expect(responseStream.write).not.toHaveBeenCalled();
+			expect(responseStream.end).not.toHaveBeenCalled();
+		},
+	);
+
+	it('writes a chunk to the open response as one NDJSON line', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		const chunk = { type: 'item', content: 'hi' };
+		deliver({ type: 'chunk', executionId: pending.executionId, payload: chunk });
+
+		expect(responseStream.write).toHaveBeenCalledWith(JSON.stringify(chunk) + '\n');
+		expect(responseStream.flush).toHaveBeenCalledTimes(1);
+		pending.release();
+	});
+
+	it('ignores a Respond node result', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		deliver({
+			type: 'response',
+			executionId: pending.executionId,
+			payload: { body: { ignored: true }, headers: {}, statusCode: 200 },
+		});
+		deliver(endedResponse(pending.executionId));
+
+		await expect(pending.outcome).resolves.toMatchObject({ status: 'completed' });
+		expect(responseStream.write).not.toHaveBeenCalled();
+	});
+
+	it('ends a successful stream without an error chunk', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		deliver(endedResponse(pending.executionId));
+		await pending.outcome;
+
+		expect(responseStream.write).not.toHaveBeenCalled();
+		expect(responseStream.end).toHaveBeenCalledTimes(1);
+	});
+
+	it('writes one fallback error chunk before a failed stream ends', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		deliver(endedResponse(pending.executionId, failedStep));
+		await pending.outcome;
+
+		expect(responseStream.write).toHaveBeenCalledTimes(1);
+		expect(responseStream.write).toHaveBeenCalledWith(
+			expect.stringContaining('"nodeId":"c","nodeName":"C"'),
+		);
+		// The error message stays out of the stream.
+		expect(responseStream.write).toHaveBeenCalledWith(
+			expect.stringContaining('"content":"Workflow execution failed"'),
+		);
+		expect(responseStream.write).not.toHaveBeenCalledWith(expect.stringContaining('it broke'));
+		expect(responseStream.flush).toHaveBeenCalledTimes(1);
+		expect(responseStream.write.mock.invocationCallOrder[0]).toBeLessThan(
+			responseStream.end.mock.invocationCallOrder[0],
+		);
+	});
+
+	it('does not add a fallback error after the executor sent one', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+		const errorChunk = {
+			type: 'error' as const,
+			content: 'it broke',
+			metadata: { nodeId: 'c', nodeName: 'C', runIndex: 0, itemIndex: 0, timestamp: 1 },
+		};
+
+		deliver({ type: 'chunk', executionId: pending.executionId, payload: errorChunk });
+		deliver(endedResponse(pending.executionId, failedStep));
+		await pending.outcome;
+
+		expect(responseStream.write).toHaveBeenCalledTimes(1);
+		expect(responseStream.write).toHaveBeenCalledWith(`${JSON.stringify(errorChunk)}\n`);
+		expect(responseStream.end).toHaveBeenCalledTimes(1);
+	});
+
+	it('writes and flushes an undeliverable response before it ends the stream', async () => {
+		const flush = vi.fn();
+		const responseStream = mock<ResponseStream>({ writableEnded: false, flush });
+		const pending = await waitForStream(responseStream);
+
+		deliver({
+			type: 'undeliverable',
+			executionId: pending.executionId,
+			error: { code: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+		});
+
+		await expect(pending.outcome).resolves.toEqual({
+			status: 'undeliverable',
+			error: { name: 'RESPONSE_TOO_LARGE', message: 'The response is too large.' },
+		});
+		expect(responseStream.write).toHaveBeenCalledWith(
+			expect.stringContaining('"nodeId":"unknown","nodeName":"unknown","runIndex":0,"itemIndex":0'),
+		);
+		expect(flush).toHaveBeenCalledTimes(1);
+		expect(responseStream.end).toHaveBeenCalledTimes(1);
+		expect(flush.mock.invocationCallOrder[0]).toBeLessThan(
+			responseStream.end.mock.invocationCallOrder[0],
+		);
+	});
+
+	it('writes an error chunk and ends the stream when the wait times out', async () => {
+		const impatient = newRegistry(1);
+		impatient.useReceiver(fakeReceiver().receiver);
+		const responseStream = openStream();
+
+		const pending = await impatient.waitForResponse(createExecutionIdV2(), stream, responseStream);
+		await vi.advanceTimersByTimeAsync(1);
+
+		await expect(pending.outcome).resolves.toEqual({ status: 'timeout' });
+		expect(responseStream.write).toHaveBeenCalledTimes(1);
+		expect(responseStream.write).toHaveBeenCalledWith(
+			expect.stringContaining('"type":"error","content":"Workflow execution timed out"'),
+		);
+		expect(responseStream.end).toHaveBeenCalledTimes(1);
+		expect(responseStream.write.mock.invocationCallOrder[0]).toBeLessThan(
+			responseStream.end.mock.invocationCallOrder[0],
+		);
+	});
+
+	it('measures the timeout from the last chunk', async () => {
+		const fake = fakeReceiver();
+		const impatient = newRegistry(1_000);
+		impatient.useReceiver(fake.receiver);
+		const responseStream = openStream();
+		const pending = await impatient.waitForResponse(createExecutionIdV2(), stream, responseStream);
+		const chunk = {
+			type: 'chunk' as const,
+			executionId: pending.executionId,
+			payload: { type: 'item', content: 'hi' },
+		};
+
+		await vi.advanceTimersByTimeAsync(900);
+		fake.deliver(chunk);
+		await vi.advanceTimersByTimeAsync(900);
+		fake.deliver(chunk);
+		await vi.advanceTimersByTimeAsync(900);
+
+		// 2.7 seconds in total, but never 1 second of silence.
+		await expect(raceWithPending(pending.outcome)).resolves.toBe(stillPending);
+
+		await vi.advanceTimersByTimeAsync(100);
+		await expect(pending.outcome).resolves.toEqual({ status: 'timeout' });
+	});
+
+	it('does not measure the timeout from chunks when the request does not wait for a stream', async () => {
+		const fake = fakeReceiver();
+		const impatient = newRegistry(1_000);
+		impatient.useReceiver(fake.receiver);
+		const pending = await impatient.waitForResponse(createExecutionIdV2(), runEnd);
+
+		await vi.advanceTimersByTimeAsync(900);
+		fake.deliver({
+			type: 'chunk',
+			executionId: pending.executionId,
+			payload: { type: 'item', content: 'ignored' },
+		});
+		await vi.advanceTimersByTimeAsync(100);
+
+		await expect(pending.outcome).resolves.toEqual({ status: 'timeout' });
+	});
+
+	it('drops chunks after the run settles', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		deliver(endedResponse(pending.executionId));
+		await pending.outcome;
+		deliver({
+			type: 'chunk',
+			executionId: pending.executionId,
+			payload: { type: 'item', content: 'late' },
+		});
+
+		expect(responseStream.write).not.toHaveBeenCalled();
+	});
+
+	it('leaves the stream to the caller when the caller releases the wait', async () => {
+		const responseStream = openStream();
+		const pending = await waitForStream(responseStream);
+
+		pending.release();
+
+		expect(responseStream.end).not.toHaveBeenCalled();
+	});
+
+	describe('heartbeat', () => {
+		it('writes and flushes a keepalive after 30 seconds', async () => {
+			const responseStream = openStream();
+			const pending = await waitForStream(responseStream);
+
+			await vi.advanceTimersByTimeAsync(30_000);
+
+			expect(responseStream.write).toHaveBeenCalledWith('{"type":"keepalive"}\n');
+			expect(responseStream.flush).toHaveBeenCalledTimes(1);
+			pending.release();
+		});
+
+		it.each(['finish', 'close'] as const)('stops after the response emits %s', async (event) => {
+			const responseStream = openStream();
+			const pending = await waitForStream(responseStream);
+			const handler = responseStream.once.mock.calls.find(
+				([registered]) => registered === event,
+			)?.[1];
+
+			handler?.();
+			await vi.advanceTimersByTimeAsync(30_000);
+
+			expect(responseStream.write).not.toHaveBeenCalled();
+			pending.release();
+		});
+
+		it('stops when the wait is released', async () => {
+			const responseStream = openStream();
+			const pending = await waitForStream(responseStream);
+
+			pending.release();
+			await vi.advanceTimersByTimeAsync(30_000);
+
+			expect(responseStream.write).not.toHaveBeenCalled();
+		});
+
+		it('stops when the subscription fails', async () => {
+			const failing = newRegistry();
+			failing.useReceiver({
+				receive: async () => {
+					throw new Error('Redis is unavailable');
+				},
+				stop: async () => {},
+			});
+			const responseStream = openStream();
+
+			await expect(
+				failing.waitForResponse(createExecutionIdV2(), stream, responseStream),
+			).rejects.toThrow('Redis is unavailable');
+			await vi.advanceTimersByTimeAsync(30_000);
+
+			expect(responseStream.write).not.toHaveBeenCalled();
+		});
+
+		it('does not start for a request that does not wait for a stream', async () => {
+			const setIntervalSpy = vi.spyOn(global, 'setInterval');
+			const pending = await registry.waitForResponse(createExecutionIdV2(), runEnd);
+
+			expect(setIntervalSpy).not.toHaveBeenCalled();
+			pending.release();
+		});
+	});
+});
+
 describe('EngineV2WebhookResponseRegistry while the receiver subscribes', () => {
 	/** A receiver whose subscriptions complete only when the test says so. */
 	function slowReceiver() {

packages/cli/src/modules/engine-v2/webhook-response/streaming-response-writer.ts

@@ -0,0 +1,85 @@
+import type { JsonValue } from '@n8n/engine';
+import type { StructuredChunk } from 'n8n-workflow';
+
+import {
+	type ResponseStream,
+	StreamingWebhookResponseHeartbeat,
+} from '@/webhooks/streaming-webhook-response-heartbeat';
+
+import type { WebhookRunOutcome } from './webhook-outcome';
+
+/**
+ * Writes the NDJSON chunks of one run to its streaming webhook response. It
+ * keeps the stream open with a heartbeat, and it ends the stream when the run
+ * settles.
+ *
+ * It only writes. The response registry decides when the run is over, and
+ * when the writer stops.
+ */
+export class StreamingResponseWriter {
+	private deliveredError = false;
+
+	private stopped = false;
+
+	private readonly heartbeat: StreamingWebhookResponseHeartbeat;
+
+	constructor(private readonly stream: ResponseStream) {
+		this.heartbeat = new StreamingWebhookResponseHeartbeat(stream);
+	}
+
+	writeChunk(payload: JsonValue): void {
+		if (this.stopped) return;
+
+		// The payload is opaque on the channel. Only this plane knows the v1 chunk.
+		const chunk = payload as unknown as StructuredChunk;
+		this.write(chunk);
+		if (chunk.type === 'error') this.deliveredError = true;
+	}
+
+	/** Writes what the outcome needs the caller to know, and ends the stream. */
+	finish(outcome: WebhookRunOutcome): void {
+		if (this.stopped) return;
+
+		if (outcome.status === 'undeliverable') {
+			this.write(this.errorChunk(outcome.error.message));
+		}
+
+		if (outcome.status === 'timeout') {
+			// Timeout chunk to tell a cut-off stream apart from a complete one.
+			this.write(this.errorChunk('Workflow execution timed out'));
+		}
+
+		if (outcome.status === 'failed' && !this.deliveredError) {
+			// The error message can hold request details, so the caller gets a
+			// generic text. The non-streaming modes do the same.
+			this.write(this.errorChunk('Workflow execution failed', outcome.nodeId, outcome.nodeName));
+		}
+
+		this.stop();
+		if (!this.stream.writableEnded) this.stream.end();
+	}
+
+	/**
+	 * Stops the writes and the heartbeat, but leaves the stream open. The caller
+	 * that gives up on the run decides how the request ends.
+	 */
+	stop(): void {
+		if (this.stopped) return;
+		this.stopped = true;
+		this.heartbeat.stop();
+	}
+
+	private write(chunk: StructuredChunk): void {
+		this.stream.write(`${JSON.stringify(chunk)}\n`);
+		// The compression middleware buffers writes until it is flushed.
+		this.stream.flush?.();
+	}
+
+	private errorChunk(content: string, nodeId = 'unknown', nodeName = 'unknown'): StructuredChunk {
+		return {
+			type: 'error',
+			content,
+			metadata: { nodeId, nodeName, runIndex: 0, itemIndex: 0, timestamp: Date.now() },
+		};
+	}
+}

packages/cli/src/modules/engine-v2/webhook-response/webhook-outcome.ts

@@ -9,7 +9,12 @@ export type LastNode = { nodeName: string; outputs: StepSlots };
 export type WebhookRunOutcome =
 	| { status: 'response'; response: unknown }
 	| { status: 'completed'; lastNode?: LastNode }
-	| { status: 'failed'; nodeName: string; error?: { name: string; message: string } }
+	| {
+			status: 'failed';
+			nodeId: string;
+			nodeName: string;
+			error?: { name: string; message: string };
+	  }
 	/** The execution response could not be produced or delivered. */
 	| { status: 'undeliverable'; error: { name: string; message: string } }
 	| { status: 'timeout' };
@@ -43,13 +48,13 @@ export function toWebhookOutcome(
 			return undefined;
 
 		case 'ended': {
-			const { nodeName, outputs, error } = received.lastStep;
+			const { nodeId, nodeName, outputs, error } = received.lastStep;
 
 			switch (received.status) {
 				case 'failed':
 					// The step that ended a failed run is the one that failed, so its name
 					// and error are what the caller reports.
-					return { status: 'failed', nodeName, error };
+					return { status: 'failed', nodeId, nodeName, error };
 
 				case 'completed':
 					return {

packages/cli/src/modules/engine-v2/webhook-response/webhook-response-registry.service.ts

@@ -6,11 +6,13 @@ import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise
 import { OperationalError, UnexpectedError } from 'n8n-workflow';
 
 import type { ExecutionIdV2 } from '@/executions/execution-id';
+import type { ResponseStream } from '@/webhooks/streaming-webhook-response-heartbeat';
 
 import type {
 	ExecutionResponseReceiver,
 	UnsubscribeExecutionResponse,
 } from '../response-channel/execution-response-receiver';
+import { StreamingResponseWriter } from './streaming-response-writer';
 import { toWebhookOutcome, type WebhookRunOutcome } from './webhook-outcome';
 
 /** The caller's view of an open webhook request that waits for its run. */
@@ -30,10 +32,15 @@ type PendingWebhook = {
 	executionId: ExecutionIdV2;
 	expectation: ResponseExpectation;
 	answer: IDeferredPromise<WebhookRunOutcome>;
-	/** Held so an answered run does not leave a timer behind for the whole hold. */
+	/**
+	 * Held so an answered run does not leave a timer behind for the whole hold.
+	 * A streaming request refreshes it on each chunk.
+	 */
 	timeoutTimer: NodeJS.Timeout;
 	/** Set when the subscription is ready. */
 	unsubscribe?: UnsubscribeExecutionResponse;
+	/** Set only when the request waits for a stream. */
+	writer?: StreamingResponseWriter;
 	released: boolean;
 };
 
@@ -49,7 +56,8 @@ export const SUBSCRIBE_TIMEOUT_MS = 30_000;
 /**
  * Keeps track of the webhook requests that wait for a data-plane run. It owns
  * the lifecycle of each wait: the capacity limit, the subscription, the
- * response timeout, and the release.
+ * response timeout, and the release. For a streaming request, it passes the
+ * chunks to a `StreamingResponseWriter`.
  *
  * `waitForResponse` should be called before starting the execution in case a
  * fast execution sends a response before we're listening for it.
@@ -83,14 +91,18 @@ export class EngineV2WebhookResponseRegistry {
 	 * `StartExecution`.
 	 * @param expectation What the request waits for. Pass the same value to
 	 * `StartExecution`.
-	 * @throws {UnexpectedError} If the execution response receiver is not set, or
-	 * if the service already waits for this execution.
+	 * @param responseStream The HTTP response to write chunks to. The service
+	 * uses it only when the expectation is `stream`, and then it is required.
+	 * @throws {UnexpectedError} If the execution response receiver is not set,
+	 * if the service already waits for this execution, or if a stream is
+	 * expected without a response stream.
 	 * @throws {OperationalError} If the service is at capacity, or if it cannot
 	 * listen for the response within `SUBSCRIBE_TIMEOUT_MS`.
 	 */
 	async waitForResponse(
 		executionId: ExecutionIdV2,
 		expectation: ResponseExpectation,
+		responseStream?: ResponseStream,
 	): Promise<WebhookResponseWait> {
 		const { receiver } = this;
 		if (!receiver) {
@@ -110,6 +122,12 @@ export class EngineV2WebhookResponseRegistry {
 			});
 		}
 
+		if (expectation.kind === 'stream' && !responseStream) {
+			throw new UnexpectedError('Engine v2 cannot stream a response without a response stream', {
+				extra: { executionId },
+			});
+		}
+
 		const pending: PendingWebhook = {
 			executionId,
 			expectation,
@@ -119,6 +137,10 @@ export class EngineV2WebhookResponseRegistry {
 				this.engineConfig.webhookResponseTimeout,
 			).unref(),
 			released: false,
+			writer:
+				expectation.kind === 'stream' && responseStream
+					? new StreamingResponseWriter(responseStream)
+					: undefined,
 		};
 
 		// Hold the slot before the subscription is ready, so requests that arrive
@@ -192,6 +214,12 @@ export class EngineV2WebhookResponseRegistry {
 		if (pending.released) return;
 
 		try {
+			if (received.type === 'chunk' && pending.writer) {
+				pending.writer.writeChunk(received.payload);
+				// A chunk shows that the run is alive
+				pending.timeoutTimer.refresh();
+			}
+
 			const outcome = toWebhookOutcome(received, pending.expectation);
 			if (outcome) this.settle(pending, outcome);
 		} catch (error) {
@@ -207,6 +235,16 @@ export class EngineV2WebhookResponseRegistry {
 	private settle(pending: PendingWebhook, outcome: WebhookRunOutcome): void {
 		if (pending.released) return;
 
+		try {
+			pending.writer?.finish(outcome);
+		} catch (error) {
+			// A closed connection must not keep the request from settling.
+			this.logger.error('Failed to end an engine v2 webhook stream', {
+				executionId: pending.executionId,
+				error,
+			});
+		}
+
 		pending.answer.resolve(outcome);
 		this.release(pending);
 	}
@@ -217,6 +255,7 @@ export class EngineV2WebhookResponseRegistry {
 		pending.released = true;
 
 		clearTimeout(pending.timeoutTimer);
+		pending.writer?.stop();
 		pending.unsubscribe?.();
 		this.pendingWebhooks.delete(pending.executionId);
 	}

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

@@ -13,7 +13,10 @@ const engineV2Webhooks = new EngineV2Webhooks(
 );
 
 describe('EngineV2Webhooks.assertSupported', () => {
-	const assertSupported = (responseMode: WebhookResponseMode) => {
+	const assertSupported = (
+		responseMode: WebhookResponseMode,
+		parameters: INode['parameters'] = {},
+	) => {
 		const proxy = mock<EngineDataPlaneProxyService>();
 		proxy.isAvailable.mockReturnValue(true);
 		const webhooks = new EngineV2Webhooks(mock<EngineV2Dispatcher>(), proxy);
@@ -22,7 +25,7 @@ describe('EngineV2Webhooks.assertSupported', () => {
 			workflowStartNode: mock<INode>({
 				name: 'Webhook',
 				type: 'n8n-nodes-base.webhook',
-				parameters: {},
+				parameters,
 			}),
 			responseMode,
 			executionId: undefined,
@@ -31,12 +34,28 @@ describe('EngineV2Webhooks.assertSupported', () => {
 
 	// The engine mode no longer reaches this check, so these modes pass with a
 	// remote data plane and in-process alike.
-	it.each(['onReceived', 'lastNode', 'responseNode'] as const)(
+	it.each(['onReceived', 'lastNode', 'responseNode', 'streaming'] as const)(
 		'allows %s responses in every engine mode',
 		(responseMode) => {
 			expect(() => assertSupported(responseMode)).not.toThrow();
 		},
 	);
+
+	// The streaming node sends its headers before it outputs the raw body, so a
+	// later refusal could not answer with a 400.
+	it.each([true, '={{ true }}'])('refuses a streaming response with rawBody %s', (rawBody) => {
+		expect(() => assertSupported('streaming', { options: { rawBody } })).toThrow(
+			'Engine v2 cannot stream a response for the "Webhook" trigger with the Raw Body option yet.',
+		);
+	});
+
+	it('allows a streaming response with rawBody turned off', () => {
+		expect(() => assertSupported('streaming', { options: { rawBody: false } })).not.toThrow();
+	});
+
+	it('allows the raw body when the response does not stream', () => {
+		expect(() => assertSupported('lastNode', { options: { rawBody: true } })).not.toThrow();
+	});
 });
 
 describe('EngineV2Webhooks.toRun', () => {
@@ -77,6 +96,7 @@ describe('EngineV2Webhooks.toRun', () => {
 		const run = await engineV2Webhooks.toRun(
 			{
 				status: 'failed',
+				nodeId: 'broken-node',
 				nodeName: 'Broken node',
 				error: { name: 'Error', message: 'The node failed' },
 			},

packages/cli/src/webhooks/__tests__/streaming-webhook-response-heartbeat.test.ts

@@ -0,0 +1,120 @@
+import { EventEmitter } from 'node:events';
+
+import {
+	type ResponseStream,
+	StreamingWebhookResponseHeartbeat,
+} from '@/webhooks/streaming-webhook-response-heartbeat';
+
+const INTERVAL_MS = 30_000;
+const KEEPALIVE = '{"type":"keepalive"}\n';
+
+/** An HTTP response stand-in that emits `finish` and `close` like the real one. */
+class FakeResponseStream extends EventEmitter implements ResponseStream {
+	writableEnded = false;
+
+	destroyed = false;
+
+	readonly write = vi.fn();
+
+	readonly end = vi.fn();
+
+	readonly flush = vi.fn();
+}
+
+describe('StreamingWebhookResponseHeartbeat', () => {
+	let stream: FakeResponseStream;
+
+	beforeEach(() => {
+		vi.useFakeTimers();
+		stream = new FakeResponseStream();
+	});
+
+	afterEach(() => {
+		vi.useRealTimers();
+	});
+
+	it('writes and flushes a keepalive at each interval', () => {
+		const heartbeat = new StreamingWebhookResponseHeartbeat(stream);
+
+		vi.advanceTimersByTime(INTERVAL_MS - 1);
+		expect(stream.write).not.toHaveBeenCalled();
+
+		vi.advanceTimersByTime(1);
+		expect(stream.write).toHaveBeenCalledWith(KEEPALIVE);
+		expect(stream.flush).toHaveBeenCalledTimes(1);
+
+		vi.advanceTimersByTime(INTERVAL_MS);
+		expect(stream.write).toHaveBeenCalledTimes(2);
+
+		heartbeat.stop();
+	});
+
+	it('writes a keepalive to a stream without a flush method', () => {
+		const withoutFlush: ResponseStream = {
+			writableEnded: false,
+			write: vi.fn(),
+			end: vi.fn(),
+			once: vi.fn(),
+			off: vi.fn(),
+		};
+		const heartbeat = new StreamingWebhookResponseHeartbeat(withoutFlush);
+
+		vi.advanceTimersByTime(INTERVAL_MS);
+
+		expect(withoutFlush.write).toHaveBeenCalledWith(KEEPALIVE);
+		heartbeat.stop();
+	});
+
+	it('writes nothing after stop', () => {
+		const heartbeat = new StreamingWebhookResponseHeartbeat(stream);
+
+		heartbeat.stop();
+		vi.advanceTimersByTime(INTERVAL_MS * 2);
+
+		expect(stream.write).not.toHaveBeenCalled();
+		expect(vi.getTimerCount()).toBe(0);
+	});
+
+	it.each(['finish', 'close'] as const)('stops when the response emits %s', (event) => {
+		new StreamingWebhookResponseHeartbeat(stream);
+
+		stream.emit(event);
+		vi.advanceTimersByTime(INTERVAL_MS);
+
+		expect(stream.write).not.toHaveBeenCalled();
+		expect(vi.getTimerCount()).toBe(0);
+	});
+
+	it('removes its listeners when it stops', () => {
+		const heartbeat = new StreamingWebhookResponseHeartbeat(stream);
+		expect(stream.listenerCount('finish')).toBe(1);
+		expect(stream.listenerCount('close')).toBe(1);
+
+		heartbeat.stop();
+
+		expect(stream.listenerCount('finish')).toBe(0);
+		expect(stream.listenerCount('close')).toBe(0);
+	});
+
+	it.each([
+		['ended', { writableEnded: true }],
+		['destroyed', { destroyed: true }],
+	])('stops without a write when the response is %s', (_, state) => {
+		new StreamingWebhookResponseHeartbeat(stream);
+		Object.assign(stream, state);
+
+		vi.advanceTimersByTime(INTERVAL_MS);
+
+		expect(stream.write).not.toHaveBeenCalled();
+		expect(vi.getTimerCount()).toBe(0);
+		expect(stream.listenerCount('close')).toBe(0);
+	});
+
+	it('allows a second stop', () => {
+		const heartbeat = new StreamingWebhookResponseHeartbeat(stream);
+
+		heartbeat.stop();
+
+		expect(() => heartbeat.stop()).not.toThrow();
+	});
+});

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

@@ -1,6 +1,7 @@
 import { Logger } from '@n8n/backend-common';
 import { EventService } from '@n8n/backend-services';
 import { mockInstance } from '@n8n/backend-test-utils';
+import { EngineConfig } from '@n8n/config';
 import type { Project, User } from '@n8n/db';
 import { UserRepository } from '@n8n/db';
 import { Container } from '@n8n/di';
@@ -52,6 +53,7 @@ import {
 	SEND_AND_WAIT_OPERATION,
 	createRunExecutionData,
 	ExpressionError,
+	OperationalError,
 	UserError,
 } from 'n8n-workflow';
 import type { Readable } from 'stream';
@@ -2388,6 +2390,7 @@ describe('executeWebhook on engine v2', () => {
 	const errorReporter = Container.get(ErrorReporter);
 	/** Response handlers registered by the responder, by execution ID. */
 	let dataPlane: Map<string, (response: ExecutionResponse) => void>;
+	let unsubscribe = vi.fn<() => void>();
 	/** The data plane mints uuidv7 ids, not the numeric ids v1 uses. */
 	const ENGINE_EXECUTION_ID = '019606a1-0000-7000-8000-000000000001';
 	/** What the webhook node produces for a request that uploaded a file. */
@@ -2415,6 +2418,7 @@ describe('executeWebhook on engine v2', () => {
 		executionId,
 		destinationNode,
 		responseHeaders,
+		response = mock<express.Response>({ headersSent: false, writableEnded: false }),
 	}: {
 		responseMode?: string;
 		startNode?: INode;
@@ -2423,6 +2427,7 @@ describe('executeWebhook on engine v2', () => {
 		destinationNode?: IDestinationNode;
 		/** The response headers configured on the webhook node, as an expression. */
 		responseHeaders?: string;
+		response?: express.Response;
 	} = {}) => {
 		webhookService.runWebhook.mockResolvedValue(webhookResult);
 
@@ -2473,12 +2478,12 @@ describe('executeWebhook on engine v2', () => {
 					}),
 			executionId,
 			mock<WebhookRequest>({ method: 'POST', contentType: undefined, headers: {} }),
-			mock<express.Response>({ headersSent: false }),
+			response,
 			responseCallback,
 			destinationNode,
 		);
 
-		return { responseCallback, returned };
+		return { response, responseCallback, returned };
 	};
 
 	beforeEach(() => {
@@ -2497,6 +2502,7 @@ describe('executeWebhook on engine v2', () => {
 		engineDataPlaneProxy.isAvailable.mockReturnValue(true);
 		workflowRunner.run.mockResolvedValue(ENGINE_EXECUTION_ID);
 		dataPlane = new Map();
+		unsubscribe = vi.fn();
 	});
 
 	/**
@@ -2531,7 +2537,8 @@ describe('executeWebhook on engine v2', () => {
 			mock<ExecutionResponseReceiver>({
 				receive: vi.fn(async (executionId: string, handler: (r: ExecutionResponse) => void) => {
 					dataPlane.set(executionId, handler);
-					return vi.fn();
+					// Reassigned for each test, so read it when the listener goes.
+					return () => unsubscribe();
 				}),
 			}),
 		);
@@ -2617,11 +2624,15 @@ describe('executeWebhook on engine v2', () => {
 				'waitForResponse',
 			);
 
-			await startWebhook({ responseMode: 'lastNode' });
+			const { response } = await startWebhook({ responseMode: 'lastNode' });
 
 			const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0];
 			expect(engineV2Response).toEqual({ responseMode: 'lastNode' });
-			expect(waitForResponse).toHaveBeenCalledWith(engineV2ExecutionId, { kind: 'runEnd' });
+			expect(waitForResponse).toHaveBeenCalledWith(
+				engineV2ExecutionId,
+				{ kind: 'runEnd' },
+				response,
+			);
 		});
 
 		it('answers with a 500 when a Respond node fails because the caller waits for the run end', async () => {
@@ -2745,11 +2756,15 @@ describe('executeWebhook on engine v2', () => {
 				'waitForResponse',
 			);
 
-			await startWebhook({ responseMode: 'responseNode' });
+			const { response } = await startWebhook({ responseMode: 'responseNode' });
 
 			const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0];
 			expect(engineV2Response).toEqual({ responseMode: 'responseNode' });
-			expect(waitForResponse).toHaveBeenCalledWith(engineV2ExecutionId, { kind: 'stepResponse' });
+			expect(waitForResponse).toHaveBeenCalledWith(
+				engineV2ExecutionId,
+				{ kind: 'stepResponse' },
+				response,
+			);
 		});
 
 		it('answers with the response published by the data plane', async () => {
@@ -2836,6 +2851,112 @@ describe('executeWebhook on engine v2', () => {
 		});
 	});
 
+	describe('streaming', () => {
+		it('tells the dispatcher that the caller waits for a stream', async () => {
+			const waitForResponse = vi.spyOn(
+				Container.get(EngineV2WebhookResponseRegistry),
+				'waitForResponse',
+			);
+
+			const { response } = await startWebhook({ responseMode: 'streaming' });
+
+			const { engineV2Response, engineV2ExecutionId } = workflowRunner.run.mock.calls[0][0];
+			expect(engineV2Response).toEqual({ responseMode: 'streaming' });
+			expect(waitForResponse).toHaveBeenCalledWith(
+				engineV2ExecutionId,
+				{ kind: 'stream' },
+				response,
+			);
+
+			// End the run, so its heartbeat and listener do not outlive the test.
+			answerRun('completed', {
+				nodeId: 'webhook-node-id',
+				nodeName: 'Webhook',
+				status: 'completed',
+				outputs: null,
+			});
+			await vi.waitFor(() => expect(unsubscribe).toHaveBeenCalledTimes(1));
+		});
+
+		it('ends the stream when the run cannot start', async () => {
+			// The streaming node sends the headers before the run starts.
+			const response = mock<express.Response>({ headersSent: true, writableEnded: false });
+			workflowRunner.run.mockRejectedValueOnce(new Error('start failed'));
+
+			await expect(startWebhook({ responseMode: 'streaming', response })).rejects.toThrow(
+				'There was a problem executing the workflow',
+			);
+
+			expect(response.end).toHaveBeenCalledTimes(1);
+			expect(unsubscribe).toHaveBeenCalledTimes(1);
+		});
+
+		it('ends the stream when the responder cannot listen for the run', async () => {
+			const response = mock<express.Response>({ headersSent: true, writableEnded: false });
+			vi.spyOn(
+				Container.get(EngineV2WebhookResponseRegistry),
+				'waitForResponse',
+			).mockRejectedValueOnce(new OperationalError('Engine v2 is at capacity'));
+
+			await expect(startWebhook({ responseMode: 'streaming', response })).rejects.toThrow(
+				'There was a problem executing the workflow',
+			);
+
+			expect(response.end).toHaveBeenCalledTimes(1);
+			expect(workflowRunner.run).not.toHaveBeenCalled();
+		});
+
+		it('completes with one error record when the failed run already sent one', async () => {
+			const { response } = await startWebhook({ responseMode: 'streaming' });
+			const executionId = workflowRunner.run.mock.calls[0][0].engineV2ExecutionId as string;
+			expect(workflowRunner.run.mock.calls[0][0].httpResponse).toBeUndefined();
+			const errorChunk = {
+				type: 'error' as const,
+				content: 'it broke',
+				metadata: {
+					nodeId: 'edit-fields',
+					nodeName: 'Edit Fields',
+					runIndex: 0,
+					itemIndex: 0,
+					timestamp: 1,
+				},
+			};
+
+			dataPlane.get(executionId)?.({ type: 'chunk', executionId, payload: errorChunk });
+			answerRun('failed', {
+				nodeId: 'edit-fields',
+				nodeName: 'Edit Fields',
+				status: 'failed',
+				outputs: null,
+				error: { name: 'NodeOperationError', message: 'it broke' },
+			});
+
+			expect(response.write).toHaveBeenCalledTimes(1);
+			expect(response.write).toHaveBeenCalledWith(`${JSON.stringify(errorChunk)}\n`);
+			expect(response.flush).toHaveBeenCalledTimes(1);
+			expect(response.end).toHaveBeenCalledTimes(1);
+		});
+
+		it('ends an open response and releases its listener on timeout', async () => {
+			vi.useFakeTimers();
+			const engineConfig = Container.get(EngineConfig);
+			const previousTimeout = engineConfig.webhookResponseTimeout;
+			engineConfig.webhookResponseTimeout = 10;
+
+			try {
+				const { response, responseCallback } = await startWebhook({ responseMode: 'streaming' });
+				await vi.advanceTimersByTimeAsync(10);
+
+				expect(response.end).toHaveBeenCalledTimes(1);
+				expect(unsubscribe).toHaveBeenCalledTimes(1);
+				expect(responseCallback).not.toHaveBeenCalled();
+			} finally {
+				engineConfig.webhookResponseTimeout = previousTimeout;
+				vi.useRealTimers();
+			}
+		});
+	});
+
 	describe('rejections', () => {
 		const reasonFrom = (responseCallback: ReturnType<typeof vi.fn>) => {
 			const error = responseCallback.mock.calls[0][0] as ResponseError;
@@ -2843,12 +2964,6 @@ describe('executeWebhook on engine v2', () => {
 		};
 
 		it.each([
-			{
-				name: 'the streaming response mode',
-				options: { responseMode: 'streaming' },
-				message:
-					"Engine v2 does not support the 'streaming' response mode yet. Respond immediately instead.",
-			},
 			{
 				name: 'the hostedChat response mode',
 				options: { responseMode: 'hostedChat' },

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

@@ -14,6 +14,8 @@ export function toResponseExpectation(responseMode?: EngineV2ResponseMode): Resp
 			return { kind: 'runEnd' };
 		case 'responseNode':
 			return { kind: 'stepResponse' };
+		case 'streaming':
+			return { kind: 'stream' };
 		default:
 			// `onReceived`, triggers, pollers and manual runs: nobody listens.
 			return { kind: 'none' };

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

@@ -1,4 +1,5 @@
 import { Service } from '@n8n/di';
+import { isRecord } from '@n8n/utils/is-record';
 import type {
 	INode,
 	IRun,
@@ -47,6 +48,7 @@ const SUPPORTED_RESPONSE_MODES = new Set<WebhookResponseMode>([
 	'onReceived',
 	'lastNode',
 	'responseNode',
+	'streaming',
 ]);
 
 /** What the request says about a run, before the webhook node has produced anything. */
@@ -128,6 +130,22 @@ export class EngineV2Webhooks {
 				`Engine v2 does not support the '${responseMode}' response mode yet. Respond immediately instead.`,
 			);
 		}
+
+		// With the raw body, a streaming Webhook node outputs a file after it sent
+		// the stream headers. `assertPayloadSupported` then cannot answer with a
+		// 400, so refuse the configuration here.
+		if (responseMode === 'streaming' && this.keepsRawBody(workflowStartNode)) {
+			throw new UserError(
+				`Engine v2 cannot stream a response for the "${workflowStartNode.name}" trigger with the Raw Body option yet.`,
+			);
+		}
+	}
+
+	private keepsRawBody(node: INode): boolean {
+		const options: unknown = node.parameters.options;
+		if (!isRecord(options)) return false;
+		// An expression can turn the option on, so only a missing or `false` value is safe.
+		return options.rawBody !== undefined && options.rawBody !== false;
 	}
 
 	/** Converts the data plane's answer to the shape the v1 response path reads. */

packages/cli/src/webhooks/streaming-webhook-response-heartbeat.ts

@@ -0,0 +1,41 @@
+/** The HTTP response operations that a streaming webhook response uses. */
+export interface ResponseStream {
+	readonly writableEnded: boolean;
+	readonly destroyed?: boolean;
+	write(chunk: string): void;
+	end(): void;
+	flush?: () => void;
+	once(event: 'finish' | 'close', listener: () => void): void;
+	off(event: 'finish' | 'close', listener: () => void): void;
+}
+
+const STREAMING_HEARTBEAT_INTERVAL_MS = 30_000;
+const STREAMING_KEEPALIVE_CHUNK = '{"type":"keepalive"}\n';
+
+/** Keeps a streaming webhook response open until it ends or disconnects. */
+export class StreamingWebhookResponseHeartbeat {
+	private stopped = false;
+
+	private readonly timer: NodeJS.Timeout;
+
+	constructor(private readonly stream: ResponseStream) {
+		this.timer = setInterval(() => {
+			if (this.stream.writableEnded || this.stream.destroyed) {
+				this.stop();
+				return;
+			}
+			this.stream.write(STREAMING_KEEPALIVE_CHUNK);
+			this.stream.flush?.();
+		}, STREAMING_HEARTBEAT_INTERVAL_MS);
+		this.stream.once('finish', this.stop);
+		this.stream.once('close', this.stop);
+	}
+
+	readonly stop = () => {
+		if (this.stopped) return;
+		this.stopped = true;
+		clearInterval(this.timer);
+		this.stream.off('finish', this.stop);
+		this.stream.off('close', this.stop);
+	};
+}

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

@@ -1232,7 +1232,8 @@ export async function executeWebhook(
 				{ executionId },
 			);
 			// TODO: Add check for streaming nodes here
-			runData.httpResponse = res;
+			// On engine v2, the response registry writes the stream instead.
+			if (!routesToEngineV2) runData.httpResponse = res;
 			runData.streamingEnabled = true;
 			// No `responder.respondWith()` here, unlike the formPage and hostedChat
 			// branches: streaming requires the trigger to have taken over the
@@ -1246,14 +1247,19 @@ export async function executeWebhook(
 		if (engineV2ExecutionId !== undefined) {
 			// Before the run, because a short workflow answers before `startExecution`
 			// returns and nothing replays a missed response.
-			if (responseMode === 'lastNode' || responseMode === 'responseNode') {
+			if (
+				responseMode === 'lastNode' ||
+				responseMode === 'responseNode' ||
+				responseMode === 'streaming'
+			) {
 				// Loaded here, because only an engine v2 run needs the module code.
 				const { EngineV2WebhookResponseRegistry } = await import(
 					'@/modules/engine-v2/webhook-response/webhook-response-registry.service.js'
 				);
+				// The registry writes to `res` only when the expectation is `stream`.
 				pendingEngineV2Response = await Container.get(
 					EngineV2WebhookResponseRegistry,
-				).waitForResponse(engineV2ExecutionId, toResponseExpectation(responseMode));
+				).waitForResponse(engineV2ExecutionId, toResponseExpectation(responseMode), res);
 				runData.engineV2Response = { responseMode };
 			}
 			// The files the node stored are under this id, so the run must use it.
@@ -1408,7 +1414,8 @@ export async function executeWebhook(
 				...(isUndeliverable ? { error: outcome.error } : {}),
 			});
 			// The webhook node can answer before the execution starts. Do not send a
-			// second response when the execution response later settles.
+			// second response when the execution response later settles. A streaming
+			// request is marked as answered, and the registry ends its stream.
 			if (!responder.hasResponded) {
 				responder.respondWith({
 					data: { message: errorResponse.responseMessage },
@@ -1551,6 +1558,11 @@ export async function executeWebhook(
 	} catch (e: unknown) {
 		// Nothing will ever answer this one, so stop waiting for it.
 		pendingEngineV2Response?.release();
+		// The trigger already sent the stream headers, so no error response can
+		// follow. End the stream, or the caller waits for chunks that never come.
+		if (routesToEngineV2 && responseMode === 'streaming' && res.headersSent && !res.writableEnded) {
+			res.end();
+		}
 
 		const error = ensureError(e);
 		let responseError: Error;

packages/cli/src/workflow-runner.ts

@@ -67,15 +67,10 @@ import type { PoolConfigService } from '@/scaling/pool-config.service.ee';
 import type { ScalingService } from '@/scaling/scaling.service';
 import type { Job, JobData } from '@/scaling/scaling.types';
 import { EngineV2Dispatcher } from '@/services/engine-v2-dispatcher.service';
+import { StreamingWebhookResponseHeartbeat } from '@/webhooks/streaming-webhook-response-heartbeat';
 import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
 import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
 
-/** Interval between keepalive writes on streaming responses to prevent proxy timeouts */
-const STREAMING_HEARTBEAT_INTERVAL_MS = 30_000;
-
-/** JSON chunk written periodically to keep the streaming connection alive through reverse proxies */
-const STREAMING_KEEPALIVE_CHUNK = '{"type":"keepalive"}\n';
-
 /** How long to keep rechecking the execution status after a max-stalled-count error before failing the run */
 const MAX_STALLED_COUNT_GRACE_WINDOW_MS = 30 * Time.seconds.toMilliseconds;
 
@@ -455,20 +450,9 @@ export class WorkflowRunner {
 		if (responsePromise) {
 			this.activeExecutions.attachResponsePromise(executionId, responsePromise);
 		}
-
-		// Set up streaming heartbeat on the main process that holds the HTTP response.
-		// This must happen BEFORE the queue/local decision because in queue mode the
-		// execution runs on a worker process that has no access to the HTTP response.
-		let heartbeatInterval: NodeJS.Timeout | undefined;
-		if (data.streamingEnabled === true && data.httpResponse) {
-			const res = data.httpResponse;
-			heartbeatInterval = setInterval(() => {
-				if (!res.writableEnded) {
-					res.write(STREAMING_KEEPALIVE_CHUNK);
-					flushResponse(res);
-				}
-			}, STREAMING_HEARTBEAT_INTERVAL_MS);
-		}
+		// The v1 main process owns the HTTP response. Start this before the
+		// queue/local decision because a worker cannot write to it.
+		const heartbeat = this.setupV1StreamingHeartbeat(data);
 
 		// @TODO: Reduce to true branch once feature is stable
 		const shouldEnqueue =
@@ -500,7 +484,7 @@ export class WorkflowRunner {
 		} catch (error) {
 			// A failed start means the post-execute promise that normally clears the
 			// heartbeat never settles, so clear it here.
-			if (heartbeatInterval) clearInterval(heartbeatInterval);
+			heartbeat?.stop();
 			throw error;
 		}
 
@@ -530,16 +514,23 @@ export class WorkflowRunner {
 		}
 
 		// Clean up the streaming heartbeat when the execution finishes
-		if (heartbeatInterval) {
+		if (heartbeat) {
 			const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId);
 			void postExecutePromise.finally(() => {
-				clearInterval(heartbeatInterval);
+				heartbeat.stop();
 			});
 		}
 
 		return executionId;
 	}
 
+	private setupV1StreamingHeartbeat(
+		data: IWorkflowExecutionDataProcess,
+	): StreamingWebhookResponseHeartbeat | undefined {
+		if (data.streamingEnabled !== true || !data.httpResponse) return undefined;
+		return new StreamingWebhookResponseHeartbeat(data.httpResponse);
+	}
+
 	private resolvePinData(data: IWorkflowExecutionDataProcess): IPinData | undefined {
 		if (['manual', 'evaluation'].includes(data.executionMode)) {
 			return data.pinData ?? data.workflowData.pinData;

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

@@ -20,6 +20,7 @@ import { agent as testAgent } from 'supertest';
 import { CacheService } from '@n8n/backend-services';
 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 { InMemoryExecutionResponseSender } from '@/modules/engine-v2/response-channel/in-memory-execution-response-sender';
 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
 import { EngineV2WebhookResponseRegistry } from '@/modules/engine-v2/webhook-response/webhook-response-registry.service';
 import { Telemetry } from '@/telemetry';
@@ -42,6 +43,7 @@ const getExecution = vi.fn();
 let builder: User;
 let webhookAgent: SuperAgentTest;
 let webhookTestEndpoint: string;
+let responseSender: InMemoryExecutionResponseSender;
 
 const webhookNode = (webhookId: string): INode => ({
 	id: randomUUID(),
@@ -84,6 +86,7 @@ beforeAll(async () => {
 	Container.get(EngineV2WebhookResponseRegistry).useReceiver(
 		new InMemoryExecutionResponseReceiver(responseChannel, Container.get(Logger)),
 	);
+	responseSender = new InMemoryExecutionResponseSender(responseChannel, Container.get(Logger));
 
 	// `/webhook-test/*` is mounted only when a server opts into test webhooks.
 	class EditorFacingWebhookServer extends WebhookServer {
@@ -142,11 +145,40 @@ describe('webhook runs on engine v2', () => {
 		expect(executions.filter((e) => e.workflowId === workflow.id)).toHaveLength(0);
 	});
 
-	test('answers 400 with the reason when the response mode is unsupported', async () => {
+	test('streams the response from the data plane', async () => {
 		const webhookId = randomUUID();
 		const trigger = webhookNode(webhookId);
 		trigger.parameters.responseMode = 'streaming';
 		const workflow = await createV2Workflow(trigger);
+		const chunk = {
+			type: 'item',
+			content: 'hello',
+			metadata: {
+				nodeId: trigger.id,
+				nodeName: trigger.name,
+				runIndex: 0,
+				itemIndex: 0,
+				timestamp: Date.now(),
+			},
+		};
+
+		startExecution.mockImplementationOnce(async (request) => {
+			responseSender.send({ type: 'chunk', executionId: request.executionId, payload: chunk });
+			responseSender.send({
+				type: 'ended',
+				executionId: request.executionId,
+				workflowId: workflow.id,
+				status: 'completed',
+				lastStep: {
+					nodeId: trigger.id,
+					nodeName: trigger.name,
+					status: 'completed',
+					outputs: [],
+				},
+			});
+
+			return { executionId: request.executionId };
+		});
 
 		await startListening(workflow.id);
 
@@ -154,8 +186,60 @@ describe('webhook runs on engine v2', () => {
 			.post(`/${webhookTestEndpoint}/${webhookId}`)
 			.send({ order: 42 });
 
-		expect(response.statusCode).toBe(400);
-		expect(response.body.message).toContain("does not support the 'streaming' response mode yet");
-		expect(startExecution).not.toHaveBeenCalled();
+		expect(response.statusCode).toBe(200);
+		expect(response.text).toBe(`${JSON.stringify(chunk)}\n`);
+		expect(startExecution).toHaveBeenCalledWith(
+			expect.objectContaining({ responseExpectation: { kind: 'stream' } }),
+		);
+	});
+
+	test('closes a failed stream without duplicating its error chunk', async () => {
+		const webhookId = randomUUID();
+		const trigger = webhookNode(webhookId);
+		trigger.parameters.responseMode = 'streaming';
+		const workflow = await createV2Workflow(trigger);
+		const errorChunk = {
+			type: 'error' as const,
+			content: 'it broke',
+			metadata: {
+				nodeId: trigger.id,
+				nodeName: trigger.name,
+				runIndex: 0,
+				itemIndex: 0,
+				timestamp: Date.now(),
+			},
+		};
+
+		startExecution.mockImplementationOnce(async (request) => {
+			responseSender.send({
+				type: 'chunk',
+				executionId: request.executionId,
+				payload: errorChunk,
+			});
+			responseSender.send({
+				type: 'ended',
+				executionId: request.executionId,
+				workflowId: workflow.id,
+				status: 'failed',
+				lastStep: {
+					nodeId: trigger.id,
+					nodeName: trigger.name,
+					status: 'failed',
+					outputs: null,
+					error: { name: 'NodeOperationError', message: 'it broke' },
+				},
+			});
+
+			return { executionId: request.executionId };
+		});
+
+		await startListening(workflow.id);
+
+		const response = await webhookAgent
+			.post(`/${webhookTestEndpoint}/${webhookId}`)
+			.send({ order: 42 });
+
+		expect(response.statusCode).toBe(200);
+		expect(response.text).toBe(`${JSON.stringify(errorChunk)}\n`);
 	});
 });

packages/workflow/src/interfaces.ts

@@ -3802,7 +3802,7 @@ export interface IWorkflowExecutionDataProcess {
 	 * which answer the caller waits for. Without this field, nobody waits.
 	 */
 	engineV2Response?: {
-		responseMode: 'lastNode' | 'responseNode';
+		responseMode: 'lastNode' | 'responseNode' | 'streaming';
 	};
 	startedAt?: Date;