feat(engine): Support Buffer bodies in Engine 2.0 sendResponse (no-changelog) (#39483)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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

@@ -1,5 +1,5 @@
 import type { ResponseEmitter, StepExecutionRequest } from '@n8n/engine';
-import { ExecutionLifecycleHooks } from 'n8n-core';
+import { ENCODED_BUFFER_KEY, ExecutionLifecycleHooks } from 'n8n-core';
 import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
 import { describe, expect, it, vi } from 'vitest';
 
@@ -66,10 +66,42 @@ describe('attachResponseHooks', () => {
 		expect(respond.send).toHaveBeenCalledOnce();
 	});
 
+	describe('a Buffer body', () => {
+		const headers = { 'content-type': 'application/octet-stream', 'content-length': 5 };
+
+		it('is sent as a base64 envelope, with its headers and status code', async () => {
+			const { request, respond } = newRequest();
+			const additionalData = newAdditionalData();
+
+			attachResponseHooks(additionalData, request);
+			await additionalData.hooks?.runHook('sendResponse', [
+				{ body: Buffer.from('hello'), headers, statusCode: 201 },
+			]);
+
+			expect(respond.send).toHaveBeenCalledWith({
+				body: { [ENCODED_BUFFER_KEY]: 'aGVsbG8=' },
+				headers,
+				statusCode: 201,
+			});
+		});
+
+		it('leaves the response the node produced untouched', async () => {
+			const { request } = newRequest();
+			const additionalData = newAdditionalData();
+			const body = Buffer.from('hello');
+			const response = { body, headers, statusCode: 200 };
+
+			attachResponseHooks(additionalData, request);
+			await additionalData.hooks?.runHook('sendResponse', [response]);
+
+			expect(response.body).toBe(body);
+		});
+	});
+
 	it.each([
-		['a buffer body', { body: Buffer.from('hi') }],
 		['a stream body', { body: { pipe: () => {} } }],
 		['a binary reference', { body: { binaryData: { id: 'file-1' } } }],
+		['a bare Buffer in place of a response', Buffer.from('hi')],
 	])('refuses %s, which has no JSON form', async (_name, response) => {
 		const { request } = newRequest();
 		const additionalData = newAdditionalData();

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

@@ -15,3 +15,4 @@ export type {
 	StepDataLoader,
 	V1StepExecutorDeps,
 } from './types';
+export { attachResponseHooks } from './v1-response-hooks';

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

@@ -1,5 +1,5 @@
 import type { JsonValue, StepExecutionRequest } from '@n8n/engine';
-import { ExecutionLifecycleHooks } from 'n8n-core';
+import { encodeBufferBody, ExecutionLifecycleHooks } from 'n8n-core';
 import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
 import { UserError } from 'n8n-workflow';
 
@@ -52,16 +52,27 @@ export function attachResponseHooks(
 /**
  * A response crosses a process boundary as JSON, so anything without a JSON
  * form fails here rather than arriving silently mangled.
+ *
+ * A top-level Buffer body is base64-encoded into the envelope
+ * which the control plane decodes, the same one Engine v1 queue mode uses.
+ * Nested Buffers and streams are not supported, as in Engine v1 queue mode.
  */
 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) {
+	if (!hasBody(value)) return value as JsonValue;
+
+	if (!Buffer.isBuffer(value.body)) {
 		assertCarriable(value.body);
+		return value as JsonValue;
 	}
 
-	return value as JsonValue;
+	// A shallow copy, so a host hook that runs after this one still sees the Buffer.
+	return encodeBufferBody({ ...value }) as JsonValue;
+}
+
+function hasBody(value: unknown): value is { body: unknown } {
+	return typeof value === 'object' && value !== null && 'body' in value;
 }
 
 function assertCarriable(value: unknown): void {

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

@@ -0,0 +1,83 @@
+import type { Logger } from '@n8n/backend-common';
+import { mockLogger } from '@n8n/backend-test-utils';
+import type { EngineConfig } from '@n8n/config';
+import type { StepExecutionRequest } from '@n8n/engine';
+import { attachResponseHooks } from '@n8n/node-engine-compatibility';
+import type { IWorkflowExecuteAdditionalData } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import { createExecutionIdV2 } from '@/executions/execution-id';
+import { EngineV2WebhookResponder } from '@/services/engine-v2-webhook-responder.service';
+
+import { InMemoryExecutionResponseChannel } from '../response-channel/in-memory-execution-response-channel';
+import { InMemoryExecutionResponseReceiver } from '../response-channel/in-memory-execution-response-receiver';
+import { InMemoryExecutionResponseSender } from '../response-channel/in-memory-execution-response-sender';
+
+/**
+ * Runs `sendResponse` through the real hook, sender, receiver and responder.
+ * The unit tests for each of them check one step. This checks that the Buffer
+ * body that goes in is the one that comes out.
+ */
+describe('a Buffer webhook response through the response channel', () => {
+	const bytes = Buffer.from([0x00, 0xff, 0x10, 0x80]);
+	const headers = { 'content-type': 'image/png', 'content-length': bytes.length };
+
+	const buildPath = () => {
+		const channel = new InMemoryExecutionResponseChannel();
+		const sender = new InMemoryExecutionResponseSender(channel, mockLogger());
+		const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
+
+		const responder = new EngineV2WebhookResponder(
+			mock<EngineConfig>({ webhookResponseTimeout: 50_000 }),
+			mock<Logger>({ scoped: () => mock<Logger>() }),
+		);
+		responder.useReceiver(receiver);
+
+		const executionId = createExecutionIdV2();
+		const pending = responder.waitForResponse(executionId, true);
+
+		const additionalData = {} as IWorkflowExecuteAdditionalData;
+		attachResponseHooks(additionalData, {
+			context: {
+				executionId,
+				stepId: 'step-1',
+				workflowId: 'wf-1',
+				mode: 'production',
+				iteration: 0,
+				callerContext: { hostMode: 'webhook' },
+			},
+			respond: sender.emitterFor(executionId),
+		} as unknown as StepExecutionRequest);
+
+		return { additionalData, pending };
+	};
+
+	it('delivers the original bytes, headers and status code to the control plane', async () => {
+		const { additionalData, pending } = buildPath();
+
+		await additionalData.hooks?.runHook('sendResponse', [
+			{ body: Buffer.from(bytes), headers, statusCode: 201 },
+		]);
+
+		const outcome = await pending.settled;
+		expect(outcome).toEqual({
+			status: 'response',
+			response: { body: bytes, headers, statusCode: 201 },
+		});
+		const { response } = outcome as { response: { body: unknown } };
+		expect(Buffer.isBuffer(response.body)).toBe(true);
+	});
+
+	it('delivers a JSON body as before', async () => {
+		const { additionalData, pending } = buildPath();
+
+		await additionalData.hooks?.runHook('sendResponse', [
+			{ body: { ok: true }, headers: {}, statusCode: 200 },
+		]);
+
+		await expect(pending.settled).resolves.toEqual({
+			status: 'response',
+			response: { body: { ok: true }, headers: {}, statusCode: 200 },
+		});
+	});
+});

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

@@ -1,5 +1,6 @@
 import { mockLogger } from '@n8n/backend-test-utils';
 import type { ExecutionResponse } from '@n8n/engine';
+import { ENCODED_BUFFER_KEY } from 'n8n-core';
 
 import { InMemoryExecutionResponseChannel } from '../response-channel/in-memory-execution-response-channel';
 import { InMemoryExecutionResponseReceiver } from '../response-channel/in-memory-execution-response-receiver';
@@ -82,6 +83,23 @@ describe('in-memory execution responses', () => {
 		);
 	});
 
+	it('delivers a base64 Buffer envelope unchanged', () => {
+		const channel = new InMemoryExecutionResponseChannel();
+		const sender = new InMemoryExecutionResponseSender(channel, mockLogger());
+		const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());
+		const seen: ExecutionResponse[] = [];
+		receiver.receive('exec-1', (response) => seen.push(response));
+		const payload = {
+			body: { [ENCODED_BUFFER_KEY]: Buffer.from([0x00, 0xff, 0x10]).toString('base64') },
+			headers: { 'content-type': 'application/octet-stream' },
+			statusCode: 200,
+		};
+
+		sender.emitterFor('exec-1').send(payload);
+
+		expect(seen).toEqual([{ type: 'response', executionId: 'exec-1', payload }]);
+	});
+
 	it('validates a response before delivering it', () => {
 		const channel = new InMemoryExecutionResponseChannel();
 		const receiver = new InMemoryExecutionResponseReceiver(channel, mockLogger());

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

@@ -1,6 +1,7 @@
 import type { Logger } from '@n8n/backend-common';
 import type { EngineConfig } from '@n8n/config';
 import type { ExecutionResponse } from '@n8n/engine';
+import { ENCODED_BUFFER_KEY } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
 
 import { createExecutionIdV2 } from '@/executions/execution-id';
@@ -151,6 +152,29 @@ describe('EngineV2WebhookResponder', () => {
 		});
 	});
 
+	it('restores a Buffer body the data plane sent as a base64 envelope', async () => {
+		const pending = responder.waitForResponse(createExecutionIdV2(), true);
+		const bytes = Buffer.from([0x00, 0xff, 0x10]);
+		const headers = { 'content-type': 'application/octet-stream', 'content-length': 3 };
+
+		deliver({
+			type: 'response',
+			executionId: pending.executionId,
+			payload: {
+				body: { [ENCODED_BUFFER_KEY]: bytes.toString('base64') },
+				headers,
+				statusCode: 201,
+			},
+		});
+
+		const outcome = await pending.settled;
+		expect(outcome).toEqual({
+			status: 'response',
+			response: { body: bytes, headers, statusCode: 201 },
+		});
+		expect(Buffer.isBuffer((outcome as { response: { body: unknown } }).response.body)).toBe(true);
+	});
+
 	it('keeps the first terminal outcome', async () => {
 		const pending = responder.waitForResponse(createExecutionIdV2(), true);
 

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

@@ -2,6 +2,7 @@ import { Logger } from '@n8n/backend-common';
 import { EngineConfig } from '@n8n/config';
 import { Service } from '@n8n/di';
 import type { EndedMessage, ExecutionResponse } from '@n8n/engine';
+import { decodeBufferBody } from 'n8n-core';
 import { OperationalError, UnexpectedError } from 'n8n-workflow';
 
 import type { ExecutionIdV2 } from '@/executions/execution-id';
@@ -108,7 +109,8 @@ export class EngineV2WebhookResponder {
 				return;
 
 			case 'response':
-				response.resolveResponse(received.payload);
+				// A Buffer body arrives base64-encoded, because the channel is JSON.
+				response.resolveResponse(decodeBufferBody(received.payload));
 				return;
 
 			case 'ended':

packages/core/src/webhook-response/buffer-body-codec.ts

@@ -10,13 +10,14 @@ import { BINARY_ENCODING } from 'n8n-workflow';
 export const ENCODED_BUFFER_KEY = '__@N8nEncodedBuffer@__';
 
 /**
- * Replaces a Buffer body with its base64 envelope. Any other body passes through.
+ * Replaces a Buffer body with its base64 envelope. Every other value passes
+ * through, a payload that is not a full response included.
  *
- * @param response Full response. Mutated and returned.
+ * @param response Response to send. Mutated and returned.
  * @returns The same `response`.
  */
-export function encodeBufferBody<T extends IN8nHttpFullResponse>(response: T): T {
-	if (Buffer.isBuffer(response.body)) {
+export function encodeBufferBody<T>(response: T): T {
+	if (hasResponseBody(response) && Buffer.isBuffer(response.body)) {
 		response.body = { [ENCODED_BUFFER_KEY]: response.body.toString(BINARY_ENCODING) };
 	}