feat(engine): Receive engine lifecycle events in the control plane (#37014)

2026-08-28 17:22:01 +08:00 · 2026-08-27 13:53:18 +00:00
parent ab32282d16
commit 68afc97ac9
15 changed files with 881 additions and 1 deletions
@@ -50,4 +50,16 @@ export class EngineConfig {
 	 */
 	@Env('N8N_ENGINE_AUTH_SECRET', z.string().min(AUTH_SECRET_MIN_LENGTH))
 	authSecret: string = '';
+
+	/** Port the control plane server listens on. Its own, so it can be firewalled off from the editor API. */
+	@Env('N8N_ENGINE_CONTROL_PLANE_PORT')
+	controlPlanePort: number = 3001;
+
+	/** Bind address for the control plane server. Loopback by default: only a data plane calls it. */
+	@Env('N8N_ENGINE_CONTROL_PLANE_HOST')
+	controlPlaneHost: string = '127.0.0.1';
+
+	/** Where the engine dials the control plane server. Defaults to loopback; set it when that is not reachable. */
+	@Env('N8N_ENGINE_CONTROL_PLANE_BASE_URL')
+	controlPlaneBaseUrl: string = '';
 }
@@ -31,4 +31,31 @@ describe('EngineConfig', () => {
 		expect(Container.get(EngineConfig).authSecret).toBe('');
 		expect(consoleWarnMock).toHaveBeenCalledWith(expect.stringContaining('N8N_ENGINE_AUTH_SECRET'));
 	});
+
+	it('should leave the control plane base URL empty so the host picks the default', () => {
+		expect(Container.get(EngineConfig).controlPlaneBaseUrl).toBe('');
+	});
+
+	it('should bind the control plane server to loopback by default', () => {
+		const config = Container.get(EngineConfig);
+
+		expect(config.controlPlaneHost).toBe('127.0.0.1');
+		expect(config.controlPlanePort).toBe(3001);
+	});
+
+	it('should read the control plane server bind address', () => {
+		process.env.N8N_ENGINE_CONTROL_PLANE_HOST = '0.0.0.0';
+		process.env.N8N_ENGINE_CONTROL_PLANE_PORT = '4001';
+
+		const config = Container.get(EngineConfig);
+
+		expect(config.controlPlaneHost).toBe('0.0.0.0');
+		expect(config.controlPlanePort).toBe(4001);
+	});
+
+	it('should read the control plane base URL', () => {
+		process.env.N8N_ENGINE_CONTROL_PLANE_BASE_URL = 'http://cp.internal:5678';
+
+		expect(Container.get(EngineConfig).controlPlaneBaseUrl).toBe('http://cp.internal:5678');
+	});
 });
@@ -0,0 +1,118 @@
+import type { Logger } from '@n8n/backend-common';
+import type { EngineConfig } from '@n8n/config';
+import { ACTION_TOKEN, mintActionToken, mintIdentityToken } from '@n8n/engine';
+import type { NextFunction, Request, Response } from 'express';
+import jwt from 'jsonwebtoken';
+import type { Mocked } from 'vitest';
+import { mock } from 'vitest-mock-extended';
+
+import { createEngineControlPlaneAuthMiddleware } from '../engine-control-plane-auth.middleware';
+
+const authSecret = 'a'.repeat(32);
+
+describe('createEngineControlPlaneAuthMiddleware', () => {
+	let logger: Logger;
+	let next: NextFunction;
+
+	const newResponse = () => {
+		const res = { status: vi.fn(), json: vi.fn() };
+		res.status.mockReturnValue(res);
+		return res as unknown as Mocked<Response>;
+	};
+
+	const newRequest = (authorization?: string) =>
+		({
+			method: 'POST',
+			originalUrl: '/internal/status-callback',
+			header: vi.fn((name: string) =>
+				name.toLowerCase() === 'authorization' ? authorization : undefined,
+			),
+		}) as unknown as Request;
+
+	const authenticate = (authorization?: string, secret = authSecret) => {
+		const res = newResponse();
+		const middleware = createEngineControlPlaneAuthMiddleware(
+			mock<EngineConfig>({ authSecret: secret }),
+			logger,
+		);
+		middleware(newRequest(authorization), res, next);
+		return res;
+	};
+
+	const expectRejected = (res: Mocked<Response>) => {
+		expect(next).not.toHaveBeenCalled();
+		expect(res.status).toHaveBeenCalledWith(401);
+		expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('/internal/status-callback'));
+	};
+
+	beforeEach(() => {
+		logger = mock<Logger>();
+		next = vi.fn();
+	});
+
+	it('accepts a callback token scoped to status writes', () => {
+		const res = authenticate(`Bearer ${mintActionToken(authSecret, 'lifecycle-events:write')}`);
+
+		expect(next).toHaveBeenCalled();
+		expect(res.status).not.toHaveBeenCalled();
+	});
+
+	it('reads the secret per request, so one generated after startup still works', () => {
+		// The secret is set after construction.
+		const engineConfig = mock<EngineConfig>({ authSecret: '' });
+		const middleware = createEngineControlPlaneAuthMiddleware(engineConfig, logger);
+		engineConfig.authSecret = authSecret;
+
+		middleware(
+			newRequest(`Bearer ${mintActionToken(authSecret, 'lifecycle-events:write')}`),
+			newResponse(),
+			next,
+		);
+
+		expect(next).toHaveBeenCalled();
+	});
+
+	it.each([
+		['no authorization header', undefined],
+		['a non-bearer scheme', 'Basic abc'],
+		['a token that is not a JWT', 'Bearer not-a-jwt'],
+	])('rejects %s', (_label, authorization) => {
+		expectRejected(authenticate(authorization));
+	});
+
+	it('rejects an identity token minted for the control plane to data plane direction', () => {
+		// Same secret; only the issuer and audience stop the replay.
+		const token = mintIdentityToken(authSecret, { cpId: 'cp-1', tenantId: 'tenant-1' });
+
+		expectRejected(authenticate(`Bearer ${token}`));
+	});
+
+	it('rejects a token carrying a different scope', () => {
+		const token = jwt.sign({ scope: 'credential:read' }, authSecret, {
+			algorithm: 'HS256',
+			issuer: ACTION_TOKEN.issuer,
+			audience: ACTION_TOKEN.audience,
+			expiresIn: ACTION_TOKEN.ttlSeconds,
+		});
+
+		expectRejected(authenticate(`Bearer ${token}`));
+	});
+
+	it('rejects a token signed with a different secret', () => {
+		const token = mintActionToken('b'.repeat(32), 'lifecycle-events:write');
+
+		expectRejected(authenticate(`Bearer ${token}`));
+	});
+
+	it('rejects everything when no shared secret is configured', () => {
+		const token = mintActionToken(authSecret, 'lifecycle-events:write');
+
+		expectRejected(authenticate(`Bearer ${token}`, ''));
+	});
+
+	it('does not leak which check failed', () => {
+		const res = authenticate('Bearer not-a-jwt');
+
+		expect(res.json).toHaveBeenCalledWith({ code: 401, message: 'Unauthenticated' });
+	});
+});
@@ -0,0 +1,151 @@
+import type {
+	HttpRequestClient,
+	HttpRequestClientOptions,
+	OutboundHttp,
+} from '@n8n/backend-network';
+import type { EngineConfig } from '@n8n/config';
+import type { LifecycleEvent } from '@n8n/engine';
+import { InvalidActionTokenError, verifyActionToken } from '@n8n/engine';
+import { OperationalError } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import { EngineControlPlaneClient } from '../engine-control-plane-client';
+
+const authSecret = 'a'.repeat(32);
+
+const events: LifecycleEvent[] = [
+	{
+		type: 'execution:completed',
+		executionId: 'exec-1',
+		workflowId: 'wf-1',
+		at: '2026-08-24T10:00:00.000Z',
+	},
+];
+
+describe('EngineControlPlaneClient', () => {
+	let http: HttpRequestClient;
+	let clientOptions: HttpRequestClientOptions | undefined;
+	let client: EngineControlPlaneClient;
+	let signal: AbortSignal;
+
+	const respondWith = (statusCode: number) => {
+		vi.mocked(http.request).mockResolvedValue({ statusCode, body: '', headers: {} });
+	};
+
+	/** Rebuilds the client so each test can vary the config. */
+	const newClient = (engineConfig: Partial<EngineConfig> = {}) => {
+		http = mock<HttpRequestClient>();
+		const outboundHttp = mock<OutboundHttp>({
+			requests: vi.fn((options?: HttpRequestClientOptions) => {
+				clientOptions = options;
+				return http;
+			}),
+		});
+
+		return new EngineControlPlaneClient(
+			mock<EngineConfig>({
+				controlPlaneBaseUrl: '',
+				authSecret,
+				controlPlanePort: 3001,
+				...engineConfig,
+			}),
+			outboundHttp,
+		);
+	};
+
+	beforeEach(() => {
+		client = newClient();
+		signal = new AbortController().signal;
+	});
+
+	describe('sendLifecycleEvents', () => {
+		it('posts the batch to the control plane status-callback endpoint', async () => {
+			respondWith(204);
+
+			await client.sendLifecycleEvents(events, signal);
+
+			expect(http.request).toHaveBeenCalledWith(
+				expect.objectContaining({
+					url: '/internal/status-callback',
+					method: 'POST',
+					body: { events },
+					json: true,
+				}),
+			);
+		});
+
+		it('does not follow redirects, so the action token reaches only the configured host', async () => {
+			respondWith(204);
+
+			await client.sendLifecycleEvents(events, signal);
+
+			expect(http.request).toHaveBeenCalledWith(
+				expect.objectContaining({ disableFollowRedirect: true }),
+			);
+		});
+
+		it('dials the control plane server on the loopback, not n8n main', () => {
+			expect(clientOptions?.baseURL).toBe('http://127.0.0.1:3001');
+		});
+
+		it('dials the configured base URL when the control plane answers elsewhere', () => {
+			newClient({ controlPlaneBaseUrl: 'https://cp.internal:8443' });
+
+			expect(clientOptions?.baseURL).toBe('https://cp.internal:8443');
+		});
+
+		it('opts out of SSRF protection for the n8n-controlled host', () => {
+			expect(clientOptions?.useDefaultSsrfPolicy).toBe('unsafe');
+		});
+
+		it('leaves the send deadline to the engine, which owns it', () => {
+			// A client timeout would fire first and hide the engine's deadline.
+			expect(clientOptions?.timeout).toBeUndefined();
+		});
+
+		it("forwards the engine's abort signal, so an abandoned batch cancels its request", async () => {
+			respondWith(204);
+
+			await client.sendLifecycleEvents(events, signal);
+
+			expect(http.request).toHaveBeenCalledWith(expect.objectContaining({ abortSignal: signal }));
+		});
+
+		it('mints a fresh token per request', () => {
+			expect(typeof clientOptions?.headers).toBe('function');
+		});
+
+		it('sends an action token scoped to lifecycle-event writes that the control plane accepts', () => {
+			const headers = clientOptions?.headers;
+			const resolved = typeof headers === 'function' ? headers() : headers;
+			const authorization = resolved?.authorization ?? '';
+
+			expect(authorization).toMatch(/^Bearer .+/);
+
+			const token = authorization.replace('Bearer ', '');
+
+			expect(() => verifyActionToken(authSecret, token, 'lifecycle-events:write')).not.toThrow();
+			expect(() => verifyActionToken('b'.repeat(32), token, 'lifecycle-events:write')).toThrow(
+				InvalidActionTokenError,
+			);
+		});
+
+		it.each([302, 400, 401, 500])(
+			'rejects a batch the control plane answered %s',
+			async (statusCode) => {
+				respondWith(statusCode);
+
+				await expect(client.sendLifecycleEvents(events, signal)).rejects.toThrow(OperationalError);
+				await expect(client.sendLifecycleEvents(events, signal)).rejects.toThrow(
+					String(statusCode),
+				);
+			},
+		);
+
+		it('resolves on a 204', async () => {
+			respondWith(204);
+
+			await expect(client.sendLifecycleEvents(events, signal)).resolves.toBeUndefined();
+		});
+	});
+});
@@ -0,0 +1,142 @@
+import type { Logger } from '@n8n/backend-common';
+import type { EngineConfig } from '@n8n/config';
+import { mintActionToken, mintIdentityToken, type LifecycleEvent } from '@n8n/engine';
+import request from 'supertest';
+import { mock } from 'vitest-mock-extended';
+
+import { EngineControlPlaneServer } from '../engine-control-plane-server';
+import { EngineLifecycleEventController } from '../engine-lifecycle-event.controller';
+
+const authSecret = 'a'.repeat(32);
+
+const events: LifecycleEvent[] = [
+	{
+		type: 'execution:completed',
+		executionId: 'exec-1',
+		workflowId: 'wf-1',
+		at: '2026-08-25T10:00:00.000Z',
+	},
+];
+
+/** Binds for real, so these exercise the wiring rather than a mock app. */
+describe('EngineControlPlaneServer', () => {
+	let server: EngineControlPlaneServer;
+	let logger: Logger;
+	let serverLogger: Logger;
+	let baseUrl: string;
+
+	const engineConfig = (overrides: Partial<EngineConfig> = {}) =>
+		mock<EngineConfig>({
+			authSecret,
+			controlPlaneHost: '127.0.0.1',
+			// Port 0: the OS picks a free one, so parallel files cannot clash.
+			controlPlanePort: 0,
+			...overrides,
+		});
+
+	beforeEach(async () => {
+		logger = mock<Logger>();
+		serverLogger = mock<Logger>();
+		const controller = new EngineLifecycleEventController(
+			mock<Logger>({ scoped: vi.fn().mockReturnValue(logger) }),
+		);
+		server = new EngineControlPlaneServer(
+			engineConfig(),
+			controller,
+			mock<Logger>({ scoped: vi.fn().mockReturnValue(serverLogger) }),
+		);
+		await server.start();
+
+		baseUrl = `http://127.0.0.1:${server.port}`;
+	});
+
+	afterEach(async () => {
+		await server.stop();
+	});
+
+	const post = (body: unknown, token?: string) => {
+		const req = request(baseUrl).post('/internal/status-callback');
+		if (token) req.set('Authorization', `Bearer ${token}`);
+		return req.send(body as object);
+	};
+
+	it('logs the port it actually bound, not the configured one', () => {
+		// Configured as `0`, so the OS picked it: logging the configured value
+		// would tell an operator to dial port 0.
+		expect(serverLogger.info).toHaveBeenCalledWith(
+			expect.stringContaining(`http://127.0.0.1:${server.port}`),
+		);
+		expect(serverLogger.info).not.toHaveBeenCalledWith(expect.stringContaining(':0'));
+	});
+
+	it('logs a server error that is not a failure to bind', () => {
+		// Nothing else handles these, so an unlogged one is a silent failure.
+		const error = Object.assign(new Error('boom'), { code: 'ECONNRESET' });
+
+		// @ts-expect-error reaching for the server's own error handler
+		server.server.emit('error', error);
+
+		expect(serverLogger.error).toHaveBeenCalledWith('Engine 2.0 control plane server error', {
+			error,
+		});
+	});
+
+	it('serves an open healthcheck', async () => {
+		const response = await request(baseUrl).get('/healthz');
+
+		expect(response.status).toBe(200);
+		expect(response.body).toEqual({ status: 'ok' });
+	});
+
+	it('accepts a batch from an authenticated data plane', async () => {
+		const response = await post({ events }, mintActionToken(authSecret, 'lifecycle-events:write'));
+
+		expect(response.status).toBe(204);
+		expect(logger.debug).toHaveBeenCalledExactlyOnceWith(
+			'Engine lifecycle event: execution:completed',
+			events[0],
+		);
+	});
+
+	it.each([
+		['no token', undefined],
+		[
+			'an identity token minted for the other direction',
+			mintIdentityToken(authSecret, {
+				cpId: 'cp-1',
+				tenantId: 'tenant-1',
+			}),
+		],
+		[
+			'a token signed with a different secret',
+			mintActionToken('b'.repeat(32), 'lifecycle-events:write'),
+		],
+	])('rejects %s', async (_label, token) => {
+		const response = await post({ events }, token);
+
+		expect(response.status).toBe(401);
+		expect(response.body).toEqual({ code: 401, message: 'Unauthenticated' });
+	});
+
+	it('rejects a batch the engine schema does not accept', async () => {
+		const response = await post(
+			{ events: [{ type: 'nope' }] },
+			mintActionToken(authSecret, 'lifecycle-events:write'),
+		);
+
+		expect(response.status).toBe(400);
+	});
+
+	it('does not authenticate the request body before the caller', async () => {
+		// An unauthenticated caller must not learn whether its body would validate.
+		const response = await post({ events: [{ type: 'nope' }] });
+
+		expect(response.status).toBe(401);
+	});
+
+	it('stops listening when stopped', async () => {
+		await server.stop();
+
+		await expect(request(baseUrl).get('/healthz')).rejects.toThrow();
+	});
+});
@@ -0,0 +1,110 @@
+import type { Logger } from '@n8n/backend-common';
+import type { LifecycleEvent } from '@n8n/engine';
+import type { Request, Response } from 'express';
+import type { Mocked } from 'vitest';
+import { mock } from 'vitest-mock-extended';
+
+import { BadRequestError } from '@/errors/response-errors/bad-request.error';
+
+import { EngineLifecycleEventController } from '../engine-lifecycle-event.controller';
+
+const events: LifecycleEvent[] = [
+	{
+		type: 'execution:started',
+		executionId: 'exec-1',
+		workflowId: 'wf-1',
+		mode: 'manual',
+		at: '2026-08-24T10:00:00.000Z',
+	},
+	{
+		type: 'step:completed',
+		executionId: 'exec-1',
+		stepId: 'step-1',
+		nodeId: 'node-a',
+		nodeName: 'Edit Fields',
+		iteration: 0,
+		outputs: [[{ json: { greeting: 'hi' } }]],
+		at: '2026-08-24T10:00:01.000Z',
+	},
+];
+
+describe('EngineLifecycleEventController', () => {
+	// The controller scopes its logger, so assert on the scoped one.
+	let logger: Logger;
+	let controller: EngineLifecycleEventController;
+
+	const newResponse = () => {
+		const res = {
+			status: vi.fn(),
+			end: vi.fn(),
+			json: vi.fn(),
+			header: vi.fn(),
+		};
+		res.status.mockReturnValue(res);
+		return res as unknown as Mocked<Response>;
+	};
+
+	const newRequest = (body: unknown = { events }) => ({ body }) as unknown as Request;
+
+	beforeEach(() => {
+		logger = mock<Logger>();
+		controller = new EngineLifecycleEventController(
+			mock<Logger>({ scoped: vi.fn().mockReturnValue(logger) }),
+		);
+	});
+
+	describe('receiveLifecycleEvents', () => {
+		it('answers 204', async () => {
+			const res = newResponse();
+
+			await controller.receiveLifecycleEvents(newRequest(), res);
+
+			expect(res.status).toHaveBeenCalledExactlyOnceWith(204);
+			expect(res.end).toHaveBeenCalled();
+		});
+
+		it('logs every event in the batch, not just a count', async () => {
+			await controller.receiveLifecycleEvents(newRequest(), newResponse());
+
+			expect(logger.debug).toHaveBeenCalledTimes(2);
+			expect(logger.debug).toHaveBeenNthCalledWith(
+				1,
+				'Engine lifecycle event: execution:started',
+				events[0],
+			);
+		});
+
+		it('logs a completed step by its output slot count, never its contents', async () => {
+			// A log must not become a copy of a user's execution data.
+			await controller.receiveLifecycleEvents(newRequest(), newResponse());
+
+			const [message, metadata] = vi.mocked(logger.debug).mock.calls[1];
+
+			expect(message).toBe('Engine lifecycle event: step:completed');
+			expect(metadata).toEqual({
+				type: 'step:completed',
+				executionId: 'exec-1',
+				stepId: 'step-1',
+				nodeId: 'node-a',
+				nodeName: 'Edit Fields',
+				iteration: 0,
+				at: '2026-08-24T10:00:01.000Z',
+				outputSlots: 1,
+			});
+			expect(JSON.stringify(metadata)).not.toContain('greeting');
+		});
+
+		it.each([
+			['a batch the engine schema rejects', { events: [{ type: 'nope' }] }],
+			['an empty batch', { events: [] }],
+			['a body that is not a batch at all', { hello: 'world' }],
+		])('rejects %s without a reason', async (_label, body) => {
+			const res = newResponse();
+
+			await expect(controller.receiveLifecycleEvents(newRequest(body), res)).rejects.toThrow(
+				BadRequestError,
+			);
+			expect(res.status).not.toHaveBeenCalled();
+		});
+	});
+});
@@ -4,6 +4,7 @@ import { Container } from '@n8n/di';

 import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';

+import { EngineControlPlaneServer } from '../engine-control-plane-server';
 import { EngineDataPlaneClient } from '../engine-data-plane-client';
 import { EngineV2Module } from '../engine-v2.module';
 import { EngineV2Runtime } from '../engine-v2.runtime';
@@ -14,6 +15,7 @@ describe('EngineV2Module', () => {
 	let engineConfig: EngineConfig;
 	let runtime: EngineV2Runtime;
 	let client: EngineDataPlaneClient;
+	let controlPlaneServer: EngineControlPlaneServer;

 	beforeEach(() => {
 		vi.clearAllMocks();
@@ -22,6 +24,7 @@ describe('EngineV2Module', () => {
 		engineConfig = mockInstance(EngineConfig, { authSecret: '' });
 		runtime = mockInstance(EngineV2Runtime);
 		client = mockInstance(EngineDataPlaneClient);
+		controlPlaneServer = mockInstance(EngineControlPlaneServer);
 		Container.set(EngineDataPlaneProxyService, new EngineDataPlaneProxyService());

 		module = new EngineV2Module();
@@ -52,6 +55,15 @@ describe('EngineV2Module', () => {
 			expect(client.startExecution).toHaveBeenCalledWith(request);
 		});

+		it('starts the control plane server before the engine, so a report always lands', async () => {
+			await module.init();
+
+			expect(controlPlaneServer.start).toHaveBeenCalled();
+			expect(vi.mocked(controlPlaneServer.start).mock.invocationCallOrder[0]).toBeLessThan(
+				vi.mocked(runtime.init).mock.invocationCallOrder[0],
+			);
+		});
+
 		it('generates a secret when unset', async () => {
 			await module.init();

@@ -73,5 +85,14 @@ describe('EngineV2Module', () => {

 			expect(runtime.shutdown).toHaveBeenCalled();
 		});
+
+		it('stops the control plane server after the engine, so a final flush still lands', async () => {
+			await module.shutdown();
+
+			expect(controlPlaneServer.stop).toHaveBeenCalled();
+			expect(vi.mocked(controlPlaneServer.stop).mock.invocationCallOrder[0]).toBeGreaterThan(
+				vi.mocked(runtime.shutdown).mock.invocationCallOrder[0],
+			);
+		});
 	});
 });
@@ -4,6 +4,7 @@ import { mock } from 'vitest-mock-extended';

 import type { NodeTypes } from '@/node-types';

+import type { EngineControlPlaneClient } from '../engine-control-plane-client';
 import { EngineV2Runtime } from '../engine-v2.runtime';

 // Hoisted so the `vi.mock` factories below, which vitest lifts above the imports,
@@ -84,8 +85,10 @@ describe('EngineV2Runtime', () => {

 	const nodeTypes = mock<NodeTypes>();

+	let controlPlaneClient: EngineControlPlaneClient;
+
 	const newRuntime = (databaseUrl = 'postgres://engine') =>
-		new EngineV2Runtime(engineConfig(databaseUrl), nodeTypes, mockLogger());
+		new EngineV2Runtime(engineConfig(databaseUrl), nodeTypes, mockLogger(), controlPlaneClient);

 	/** The `externalDependencies` callback the runtime handed to the engine. */
 	const externalDependencies = (stores: { executionStore: unknown; stepStore: unknown }) => {
@@ -99,6 +102,7 @@ describe('EngineV2Runtime', () => {
 		vi.clearAllMocks();
 		mocks.dataSource.isInitialized = false;
 		mocks.listen.error = undefined;
+		controlPlaneClient = mock<EngineControlPlaneClient>();
 	});

 	describe('init', () => {
@@ -125,6 +129,30 @@ describe('EngineV2Runtime', () => {
 			expect(mocks.engine.start).toHaveBeenCalled();
 		});

+		it('injects a lifecycle event callback that reports to the control plane', async () => {
+			await newRuntime().init();
+
+			const events = [
+				{
+					type: 'execution:completed' as const,
+					executionId: 'exec-1',
+					workflowId: 'wf-1',
+					at: '2026-08-24T10:00:00.000Z',
+				},
+			];
+			const lifecycleEventCallback = externalDependencies({ executionStore: {}, stepStore: {} })
+				.lifecycleEventCallback as (events: unknown[], signal: AbortSignal) => Promise<void>;
+			const signal = new AbortController().signal;
+
+			await lifecycleEventCallback(events, signal);
+
+			// The signal rides along, so an abandoned batch cancels its request.
+			expect(controlPlaneClient.sendLifecycleEvents).toHaveBeenCalledExactlyOnceWith(
+				events,
+				signal,
+			);
+		});
+
 		it('injects the v1 step executor so v1-node steps can run', async () => {
 			await newRuntime().init();

@@ -0,0 +1,49 @@
+import type { Logger } from '@n8n/backend-common';
+import type { EngineConfig } from '@n8n/config';
+import { InvalidActionTokenError, verifyActionToken } from '@n8n/engine';
+import type { NextFunction, Request, Response } from 'express';
+
+import { UnauthenticatedError } from '@/errors/response-errors/unauthenticated.error';
+
+const BEARER_PREFIX = /^bearer /i;
+
+/** Express's `RequestHandler` may return a promise; verifying a token never does. */
+type SyncRequestHandler = (req: Request, res: Response, next: NextFunction) => void;
+
+/**
+ * Rejects a caller the shared secret does not vouch for. Reads the secret per
+ * request, because it is generated after this is constructed.
+ */
+export function createEngineControlPlaneAuthMiddleware(
+	engineConfig: EngineConfig,
+	logger: Logger,
+): SyncRequestHandler {
+	const reject = (req: Request, res: Response, reason: string): void => {
+		// Logged, never returned: an operator needs the reason, a caller must not.
+		logger.warn(`Rejected ${req.method} ${req.originalUrl} - ${reason}`);
+		const error = new UnauthenticatedError();
+		res.status(error.httpStatusCode).json({ code: error.errorCode, message: error.message });
+	};
+
+	return (req, res, next) => {
+		const header = req.header('authorization') ?? '';
+		if (!BEARER_PREFIX.test(header)) {
+			reject(req, res, 'no bearer token');
+			return;
+		}
+
+		try {
+			verifyActionToken(
+				engineConfig.authSecret,
+				header.replace(BEARER_PREFIX, ''),
+				'lifecycle-events:write',
+			);
+		} catch (error) {
+			if (!(error instanceof InvalidActionTokenError)) throw error;
+			reject(req, res, 'token rejected');
+			return;
+		}
+
+		next();
+	};
+}
@@ -0,0 +1,60 @@
+import type { HttpRequestClient } from '@n8n/backend-network';
+import { OutboundHttp } from '@n8n/backend-network';
+import { EngineConfig } from '@n8n/config';
+import { Service } from '@n8n/di';
+import type { LifecycleEvent } from '@n8n/engine';
+import { mintActionToken } from '@n8n/engine';
+import { OperationalError } from 'n8n-workflow';
+
+import { STATUS_CALLBACK_PATH } from './engine-v2.constants';
+
+/**
+ * Posts lifecycle events to the control plane server. Over HTTP even in-process,
+ * so the engine can move out of it without changing either side.
+ */
+@Service()
+export class EngineControlPlaneClient {
+	private readonly http: HttpRequestClient;
+
+	constructor(
+		private readonly engineConfig: EngineConfig,
+		outboundHttp: OutboundHttp,
+	) {
+		this.http = outboundHttp.requests({
+			// Fixed, n8n-controlled host.
+			useDefaultSsrfPolicy: 'unsafe',
+			// A bind address is not dialable, so default to loopback.
+			baseURL:
+				engineConfig.controlPlaneBaseUrl || `http://127.0.0.1:${engineConfig.controlPlanePort}`,
+			// A factory: each request needs a fresh token, and the secret is set later.
+			headers: () => ({
+				authorization: `Bearer ${mintActionToken(this.engineConfig.authSecret, 'lifecycle-events:write')}`,
+			}),
+		});
+	}
+
+	/** Throws on a refused batch; the engine decides what a failed delivery costs. */
+	async sendLifecycleEvents(events: LifecycleEvent[], signal: AbortSignal): Promise<void> {
+		const response = await this.http.request<unknown>({
+			url: STATUS_CALLBACK_PATH,
+			method: 'POST',
+			body: { events },
+			json: true,
+			returnFullResponse: true,
+			// Inspect the status here rather than catching a generic request error.
+			ignoreHttpStatusErrors: true,
+			// A redirect would forward the token to whatever host it names.
+			disableFollowRedirect: true,
+			// The engine owns the deadline and aborts on it. A client timeout would
+			// fire first and make that unreachable.
+			abortSignal: signal,
+		});
+
+		// 3xx too: redirects are not followed, so one is a misconfiguration.
+		if (response.statusCode >= 300) {
+			throw new OperationalError(
+				`Control plane refused a lifecycle event batch with ${response.statusCode}`,
+			);
+		}
+	}
+}
@@ -0,0 +1,104 @@
+import { inTest, Logger } from '@n8n/backend-common';
+import { EngineConfig } from '@n8n/config';
+import { Service } from '@n8n/di';
+import express, { type Application } from 'express';
+import { createServer, type Server } from 'node:http';
+import type { AddressInfo } from 'node:net';
+
+import { bodyParser, rawBodyReader } from '@/middlewares';
+import { send } from '@/response-helper';
+
+import { createEngineControlPlaneAuthMiddleware } from './engine-control-plane-auth.middleware';
+import { CONTROL_PLANE_PREFIX, STATUS_CALLBACK_PATH } from './engine-v2.constants';
+import { EngineLifecycleEventController } from './engine-lifecycle-event.controller';
+
+/**
+ * Receives lifecycle events from the data plane. Its own server, not a route on
+ * n8n's main one, so this surface can be isolated from the editor API.
+ */
+@Service()
+export class EngineControlPlaneServer {
+	private server: Server | undefined;
+
+	/** The bound port, which differs from the configured one when that is `0`. */
+	get port(): number | undefined {
+		return (this.server?.address() as AddressInfo | null)?.port;
+	}
+
+	constructor(
+		private readonly engineConfig: EngineConfig,
+		private readonly lifecycleEventController: EngineLifecycleEventController,
+		private readonly logger: Logger,
+	) {
+		this.logger = this.logger.scoped('engine-v2');
+	}
+
+	async start(): Promise<void> {
+		const app = express();
+		app.disable('x-powered-by');
+		this.configureRoutes(app);
+
+		const { controlPlaneHost: host, controlPlanePort: port } = this.engineConfig;
+
+		this.server = createServer(app);
+		this.server.on('error', (error: NodeJS.ErrnoException) => {
+			if (error.code !== 'EADDRINUSE') {
+				// Nothing else handles these, so an unlogged one is a silent failure.
+				this.logger.error('Engine 2.0 control plane server error', { error });
+				return;
+			}
+
+			this.logger.error(`Engine 2.0 control plane port ${port} is already in use`);
+			// Skipped in tests, where exiting would kill the vitest worker.
+			if (!inTest) process.exit(1);
+		});
+
+		await new Promise<void>((resolve, reject) => {
+			// Detached once listening: a later error is not a failed bind.
+			const onBindError = (error: Error) => reject(error);
+			this.server!.once('error', onBindError);
+			this.server!.listen(port, host, () => {
+				this.server!.off('error', onBindError);
+				resolve();
+			});
+		});
+
+		// An IPv6 literal needs brackets to read as a URL.
+		const shownHost = host.includes(':') ? `[${host}]` : host;
+		// The bound port, not the configured one, which is `0` when the OS picks it.
+		this.logger.info(`Engine 2.0 control plane listening on http://${shownHost}:${this.port}`);
+	}
+
+	async stop(): Promise<void> {
+		if (!this.server) return;
+
+		await new Promise<void>((resolve, reject) => {
+			this.server!.close((error) => (error ? reject(error) : resolve()));
+		});
+
+		// Dropped only on success, so a failed close is retried next time.
+		this.server = undefined;
+	}
+
+	private configureRoutes(app: Application): void {
+		// Open: a liveness probe reveals nothing.
+		app.get('/healthz', (_req, res) => {
+			res.status(200).json({ status: 'ok' });
+		});
+
+		// On the prefix, not the route, so a later route cannot forget either.
+		app.use(
+			CONTROL_PLANE_PREFIX,
+			createEngineControlPlaneAuthMiddleware(this.engineConfig, this.logger),
+		);
+		// n8n's parser bounds the body by `N8N_PAYLOAD_SIZE_MAX`.
+		app.use(CONTROL_PLANE_PREFIX, rawBodyReader, bodyParser);
+
+		app.post(
+			STATUS_CALLBACK_PATH,
+			send(
+				async (req, res) => await this.lifecycleEventController.receiveLifecycleEvents(req, res),
+			),
+		);
+	}
+}
@@ -0,0 +1,41 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+import type { LifecycleEvent } from '@n8n/engine';
+import { lifecycleEventBatchSchema } from '@n8n/engine';
+import type { Request, Response } from 'express';
+
+import { BadRequestError } from '@/errors/response-errors/bad-request.error';
+
+/** Handles `LifecycleEvent` batches from the engine 2.0 data plane. */
+@Service()
+export class EngineLifecycleEventController {
+	constructor(private readonly logger: Logger) {
+		this.logger = this.logger.scoped('engine-v2');
+	}
+
+	async receiveLifecycleEvents(req: Request, res: Response): Promise<void> {
+		const parsed = lifecycleEventBatchSchema.safeParse(req.body);
+		// The engine's own schema, so emitter and receiver cannot drift. The reason
+		// stays out of the response: only a data plane calls this.
+		if (!parsed.success) throw new BadRequestError('Invalid lifecycle event batch');
+
+		// TODO(CAT-2878): forward these to the editor and dispatch the error workflow.
+		for (const event of parsed.data.events) {
+			this.logger.debug(`Engine lifecycle event: ${event.type}`, toLogMetadata(event));
+		}
+
+		// Nothing to return, and re-delivery is harmless while this only logs.
+		res.status(204).end();
+	}
+}
+
+/**
+ * An event's identifiers, ready to log. Outputs become a slot count, so a log
+ * never becomes a copy of a user's execution data.
+ */
+function toLogMetadata(event: LifecycleEvent): Record<string, unknown> {
+	if (event.type !== 'step:completed') return { ...event };
+
+	const { outputs, ...rest } = event;
+	return { ...rest, outputSlots: outputs.length };
+}
@@ -0,0 +1,5 @@
+/** Prefix for every route the data plane calls. Auth and body limits mount on it. */
+export const CONTROL_PLANE_PREFIX = '/internal';
+
+/** Where the data plane posts lifecycle event batches. Shared, so the two sides cannot disagree. */
+export const STATUS_CALLBACK_PATH = `${CONTROL_PLANE_PREFIX}/status-callback`;
@@ -29,6 +29,10 @@ export class EngineV2Module implements ModuleInterface {
 			engineConfig.authSecret = randomBytes(32).toString('hex');
 		}

+		// Before the engine, so nothing is reported with no server to receive it.
+		const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js');
+		await Container.get(EngineControlPlaneServer).start();
+
 		const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
 		await Container.get(EngineV2Runtime).init();

@@ -45,5 +49,9 @@ export class EngineV2Module implements ModuleInterface {
 	async shutdown() {
 		const { EngineV2Runtime } = await import('./engine-v2.runtime.js');
 		await Container.get(EngineV2Runtime).shutdown();
+
+		// After the engine, so its final flush still has somewhere to land.
+		const { EngineControlPlaneServer } = await import('./engine-control-plane-server.js');
+		await Container.get(EngineControlPlaneServer).stop();
 	}
 }
@@ -14,6 +14,7 @@ import assert from 'node:assert';
 import type { Server } from 'node:http';

 import { NodeTypes } from '@/node-types';
+import { EngineControlPlaneClient } from './engine-control-plane-client';
 import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';

 /**
@@ -36,6 +37,7 @@ export class EngineV2Runtime {
 		private readonly engineConfig: EngineConfig,
 		private readonly nodeTypes: NodeTypes,
 		private readonly logger: Logger,
+		private readonly controlPlaneClient: EngineControlPlaneClient,
 	) {
 		this.logger = this.logger.scoped('engine-v2');
 	}
@@ -83,6 +85,8 @@ export class EngineV2Runtime {
 			admittance: new AllowAllAdmittance(),
 			identityVerifier: new SharedSecretIdentityVerifier(this.engineConfig.authSecret),
 			externalDependencies: ({ executionStore, stepStore }) => ({
+				lifecycleEventCallback: async (events, signal) =>
+					await this.controlPlaneClient.sendLifecycleEvents(events, signal),
 				v1StepExecutor: new V1StepExecutor({
 					nodeTypes: this.nodeTypes,
 					// TODO(CAT-2880): no credential access. A v1 node that needs credentials