refactor(core): Simplify MCP execution wait and progress plumbing

- Drop the transport flush wrapper: the SDK's SSE responses set
  Cache-Control: no-transform, so the compression middleware never
  buffers them and notifications already stream in real time.
- Reuse ExecutionStatusList from n8n-workflow instead of a hand-copied
  status list.
- Hoist the shared IRun -> {status, error} outcome derivation into
  execution-utils.
- Let waitForExecutionResult own the progress reporter lifecycle and
  make cancelOnTimeout explicit at every call site; remove the
  McpService.waitForExecutionResult forwarding method and the
  onExecutionStarted callback.
- Stop heartbeats on client abort and unref the interval.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VGSEzz3A8eQqbAtnqXpaFk

packages/cli/src/modules/mcp/__tests__/execute-workflow.tool.test.ts

@@ -15,9 +15,11 @@ import {
 import { v4 as uuid } from 'uuid';
 
 import { createWorkflow, createWorkflowHistoryVersion } from './mock.utils';
-import { McpExecutionTimeoutError, WorkflowAccessError } from '../mcp.errors';
+import { WorkflowAccessError } from '../mcp.errors';
 import { createExecuteWorkflowTool, executeWorkflow } from '../tools/execute-workflow.tool';
+import { WORKFLOW_EXECUTION_TIMEOUT_MS } from '../tools/execution-utils';
 
+import { ActiveExecutions } from '@/active-executions';
 import { McpService } from '@/modules/mcp/mcp.service';
 import { Telemetry } from '@/telemetry';
 import { WorkflowRunner } from '@/workflow-runner';
@@ -27,6 +29,7 @@ import { WorkflowPublishedDataService } from '@/workflows/workflow-published-dat
 describe('execute-workflow MCP tool', () => {
 	const user = Object.assign(new User(), { id: 'user-1' });
 	let workflowFinderService: WorkflowFinderService;
+	let activeExecutions: ActiveExecutions;
 	let workflowRunner: WorkflowRunner;
 	let telemetry: Telemetry;
 	let mcpService: McpService;
@@ -35,6 +38,7 @@ describe('execute-workflow MCP tool', () => {
 
 	beforeEach(() => {
 		workflowFinderService = mockInstance(WorkflowFinderService);
+		activeExecutions = mockInstance(ActiveExecutions);
 		workflowRunner = mockInstance(WorkflowRunner);
 		telemetry = mockInstance(Telemetry, {
 			track: vi.fn(),
@@ -52,6 +56,7 @@ describe('execute-workflow MCP tool', () => {
 			const tool = createExecuteWorkflowTool(
 				user,
 				workflowFinderService,
+				activeExecutions,
 				workflowRunner,
 				telemetry,
 				mcpService,
@@ -706,6 +711,7 @@ describe('execute-workflow MCP tool', () => {
 				createExecuteWorkflowTool(
 					user,
 					workflowFinderService,
+					activeExecutions,
 					workflowRunner,
 					telemetry,
 					mcpService,
@@ -750,26 +756,26 @@ describe('execute-workflow MCP tool', () => {
 
 				const result = await callHandler(undefined);
 
-				expect(mcpService.waitForExecutionResult).not.toHaveBeenCalled();
+				expect(activeExecutions.getPostExecutePromise).not.toHaveBeenCalled();
 				expect(result.structuredContent).toEqual({ executionId: 'exec-wait', status: 'started' });
 			});
 
 			test('returns the final status when the execution completes', async () => {
 				setupExecutableWorkflow();
-				(mcpService.waitForExecutionResult as Mock).mockResolvedValue({
+				(activeExecutions.getPostExecutePromise as Mock).mockResolvedValue({
 					status: 'success',
 					data: { resultData: {} },
 				});
 
 				const result = await callHandler(true);
 
-				expect(mcpService.waitForExecutionResult).toHaveBeenCalledWith('exec-wait');
+				expect(activeExecutions.getPostExecutePromise).toHaveBeenCalledWith('exec-wait');
 				expect(result.structuredContent).toEqual({ executionId: 'exec-wait', status: 'success' });
 			});
 
 			test('returns error status and message when the execution fails', async () => {
 				setupExecutableWorkflow();
-				(mcpService.waitForExecutionResult as Mock).mockResolvedValue({
+				(activeExecutions.getPostExecutePromise as Mock).mockResolvedValue({
 					status: 'error',
 					data: { resultData: { error: { message: 'node exploded' } } },
 				});
@@ -783,26 +789,39 @@ describe('execute-workflow MCP tool', () => {
 				});
 			});
 
-			test('returns running status when the wait times out', async () => {
-				setupExecutableWorkflow();
-				(mcpService.waitForExecutionResult as Mock).mockRejectedValue(
-					new McpExecutionTimeoutError('exec-wait', 300_000),
-				);
+			test('returns running status and leaves the execution running when the wait times out', async () => {
+				vi.useFakeTimers();
+				try {
+					setupExecutableWorkflow();
+					(activeExecutions.getPostExecutePromise as Mock).mockReturnValue(new Promise(() => {}));
 
-				const result = await callHandler(true);
+					const resultPromise = callHandler(true);
+					await vi.advanceTimersByTimeAsync(WORKFLOW_EXECUTION_TIMEOUT_MS);
+					const result = await resultPromise;
 
-				expect(result.structuredContent).toEqual({ executionId: 'exec-wait', status: 'running' });
+					expect(result.structuredContent).toEqual({
+						executionId: 'exec-wait',
+						status: 'running',
+					});
+					expect(activeExecutions.stopExecution).not.toHaveBeenCalled();
+				} finally {
+					vi.useRealTimers();
+				}
 			});
 
 			test('sends progress notifications when the client provides a progress token', async () => {
 				setupExecutableWorkflow();
-				(mcpService.waitForExecutionResult as Mock).mockResolvedValue({
+				(activeExecutions.getPostExecutePromise as Mock).mockResolvedValue({
 					status: 'success',
 					data: { resultData: {} },
 				});
 				const sendNotification = vi.fn().mockResolvedValue(undefined);
 
-				await callHandler(true, { _meta: { progressToken: 'token-1' }, sendNotification });
+				await callHandler(true, {
+					_meta: { progressToken: 'token-1' },
+					sendNotification,
+					signal: new AbortController().signal,
+				});
 
 				expect(sendNotification).toHaveBeenCalledWith(
 					expect.objectContaining({
@@ -817,7 +836,7 @@ describe('execute-workflow MCP tool', () => {
 
 			test('sends no notifications when the client provides no progress token', async () => {
 				setupExecutableWorkflow();
-				(mcpService.waitForExecutionResult as Mock).mockResolvedValue({
+				(activeExecutions.getPostExecutePromise as Mock).mockResolvedValue({
 					status: 'success',
 					data: { resultData: {} },
 				});
@@ -851,6 +870,7 @@ describe('execute-workflow MCP tool', () => {
 				const tool = createExecuteWorkflowTool(
 					user,
 					workflowFinderService,
+					activeExecutions,
 					workflowRunner,
 					telemetry,
 					mcpService,
@@ -896,6 +916,7 @@ describe('execute-workflow MCP tool', () => {
 				const tool = createExecuteWorkflowTool(
 					user,
 					workflowFinderService,
+					activeExecutions,
 					workflowRunner,
 					telemetry,
 					mcpService,
@@ -940,6 +961,7 @@ describe('execute-workflow MCP tool', () => {
 				const tool = createExecuteWorkflowTool(
 					user,
 					workflowFinderService,
+					activeExecutions,
 					workflowRunner,
 					telemetry,
 					mcpService,

packages/cli/src/modules/mcp/__tests__/execution-utils.test.ts

@@ -24,6 +24,7 @@ describe('createExecutionProgressReporter', () => {
 		({
 			_meta: progressToken === undefined ? undefined : { progressToken },
 			sendNotification: vi.fn().mockResolvedValue(undefined),
+			signal: new AbortController().signal,
 		}) as unknown as Parameters<typeof createExecutionProgressReporter>[0] & {
 			sendNotification: ReturnType<typeof vi.fn>;
 		};
@@ -59,12 +60,24 @@ describe('createExecutionProgressReporter', () => {
 		expect(extra.sendNotification).toHaveBeenCalledTimes(3);
 	});
 
+	test('stops heartbeats when the request is aborted', () => {
+		const abortController = new AbortController();
+		const extra = createExtra('token-1');
+		Object.defineProperty(extra, 'signal', { value: abortController.signal });
+		const reporter = createExecutionProgressReporter(extra, 'Execution 1');
+
+		reporter.start();
+		abortController.abort();
+		vi.advanceTimersByTime(PROGRESS_HEARTBEAT_INTERVAL_MS * 2);
+
+		expect(extra.sendNotification).toHaveBeenCalledTimes(1); // only the start message
+	});
+
 	test('is a no-op when the client sent no progress token', () => {
 		const extra = createExtra();
 		const reporter = createExecutionProgressReporter(extra, 'Execution 1');
 
 		reporter.start();
-		reporter.update('some update');
 		vi.advanceTimersByTime(PROGRESS_HEARTBEAT_INTERVAL_MS * 2);
 		reporter.stop();
 
@@ -89,10 +102,12 @@ describe('waitForExecutionResult timeout behavior', () => {
 		return { activeExecutions, mcpService };
 	};
 
-	test('cancels the execution on timeout by default', async () => {
+	test('cancels the execution on timeout when cancelOnTimeout is true', async () => {
 		const { activeExecutions, mcpService } = createMocks();
 
-		const waitPromise = waitForExecutionResult('exec-1', activeExecutions, mcpService);
+		const waitPromise = waitForExecutionResult('exec-1', activeExecutions, mcpService, {
+			cancelOnTimeout: true,
+		});
 		const assertion = expect(waitPromise).rejects.toThrow(McpExecutionTimeoutError);
 		await vi.advanceTimersByTimeAsync(WORKFLOW_EXECUTION_TIMEOUT_MS);
 		await assertion;

packages/cli/src/modules/mcp/__tests__/mcp.controller.test.ts

@@ -33,7 +33,6 @@ vi.mock('@modelcontextprotocol/sdk/server/streamableHttp.js', () => {
 	const StreamableHTTPServerTransport = vi.fn().mockImplementation(function (_opts) {
 		return {
 			handleRequest: mockHandleRequest,
-			send: vi.fn().mockResolvedValue(undefined),
 			close: vi.fn().mockResolvedValue(undefined),
 		};
 	});

packages/cli/src/modules/mcp/mcp.controller.ts

@@ -210,22 +210,12 @@ export class McpController {
 		const transport = new StreamableHTTPServerTransport({
 			sessionIdGenerator: undefined,
 		});
-		// The global compression() middleware buffers writes, so mid-request
-		// messages (e.g. progress notifications) would only arrive with the final
-		// response. Flush after every message so they stream in real time — same
-		// approach as the McpTrigger node's StreamableHttpTransport.
-		const originalSend = transport.send.bind(transport);
-		transport.send = async (...args: Parameters<typeof transport.send>) => {
-			await originalSend(...args);
-			res.flush?.();
-		};
 		res.on('close', () => {
 			void transport.close();
 			void server.close();
 		});
 		await server.connect(transport);
 		await transport.handleRequest(req, res, body);
-		res.flush?.();
 	}
 
 	private trackConnectionEvent(payload: UserConnectedToMCPEventPayload) {

packages/cli/src/modules/mcp/mcp.service.ts

@@ -77,7 +77,6 @@ import { WorkflowService } from '@/workflows/workflow.service';
 
 import { MCP_PREVIEW_RENDER_REQUESTED_EVENT } from './mcp.constants';
 import { getAllowedToolNames } from './mcp-scopes';
-import { waitForExecutionResult } from './tools/execution-utils';
 import type { McpAppsTelemetryVariant, McpClientInfo, RegisterToolFn } from './mcp.types';
 import {
 	createAddDataTableColumnTool,
@@ -265,6 +264,7 @@ export class McpService {
 		const executeWorkflowTool = createExecuteWorkflowTool(
 			user,
 			this.workflowFinderService,
+			this.activeExecutions,
 			this.workflowRunner,
 			this.telemetry,
 			this,
@@ -428,17 +428,6 @@ export class McpService {
 		return server;
 	}
 
-	/**
-	 * Wait for a workflow execution started by an MCP tool to complete.
-	 * A timed-out wait leaves the execution running — the wait is a
-	 * convenience and must not cancel a (possibly production) run.
-	 */
-	async waitForExecutionResult(executionId: string): Promise<IRun> {
-		return await waitForExecutionResult(executionId, this.activeExecutions, this, {
-			cancelOnTimeout: false,
-		});
-	}
-
 	private async registerBuilderTools(
 		server: InstanceType<typeof McpServer>,
 		user: User,

packages/cli/src/modules/mcp/tools/execute-workflow.tool.ts

@@ -3,6 +3,7 @@ import type { User } from '@n8n/db';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import {
 	CHAT_TRIGGER_NODE_TYPE,
+	ExecutionStatusList,
 	FORM_TRIGGER_NODE_TYPE,
 	MANUAL_TRIGGER_NODE_TYPE,
 	WEBHOOK_NODE_TYPE,
@@ -28,9 +29,10 @@ import type {
 	UserCalledMCPToolEventPayload,
 } from '../mcp.types';
 import { findMcpSupportedTrigger } from '../mcp.utils';
-import { createExecutionProgressReporter } from './execution-utils';
+import { getExecutionOutcome, waitForExecutionResult } from './execution-utils';
 import { getMcpWorkflow, type FoundWorkflow } from './workflow-validation.utils';
 
+import type { ActiveExecutions } from '@/active-executions';
 import type { McpService } from '@/modules/mcp/mcp.service';
 import type { Telemetry } from '@/telemetry';
 import type { WorkflowRunner } from '@/workflow-runner';
@@ -88,17 +90,7 @@ const inputSchema = z.object({
 		),
 });
 
-const EXECUTION_STATUS_VALUES = [
-	'started',
-	'success',
-	'error',
-	'running',
-	'waiting',
-	'canceled',
-	'crashed',
-	'new',
-	'unknown',
-] as const;
+const EXECUTION_STATUS_VALUES = ['started', ...ExecutionStatusList] as const;
 
 type ExecuteWorkflowOutput = {
 	executionId: string | null;
@@ -119,6 +111,7 @@ const outputSchema = {
 export const createExecuteWorkflowTool = (
 	user: User,
 	workflowFinderService: WorkflowFinderService,
+	activeExecutions: ActiveExecutions,
 	workflowRunner: WorkflowRunner,
 	telemetry: Telemetry,
 	mcpService: McpService,
@@ -162,19 +155,24 @@ export const createExecuteWorkflowTool = (
 			);
 
 			if (waitForCompletion && output.executionId) {
-				const progress = createExecutionProgressReporter(
-					extra,
-					`Execution ${output.executionId} of workflow ${workflowId}`,
-				);
-				progress.start();
 				try {
-					const data = await mcpService.waitForExecutionResult(output.executionId);
-					const hasError = data.status === 'error' || data.data.resultData?.error;
-					output.status = hasError ? 'error' : data.status;
-					if (hasError) {
-						output.error =
-							data.data.resultData?.error?.message ?? 'Execution completed with errors';
-					}
+					const data = await waitForExecutionResult(
+						output.executionId,
+						activeExecutions,
+						mcpService,
+						{
+							// The wait is a convenience and must not cancel a (possibly
+							// production) run as a side effect of timing out.
+							cancelOnTimeout: false,
+							progress: {
+								extra,
+								label: `Execution ${output.executionId} of workflow ${workflowId}`,
+							},
+						},
+					);
+					const { status, error } = getExecutionOutcome(data);
+					output.status = status;
+					output.error = error;
 				} catch (waitError) {
 					if (waitError instanceof McpExecutionTimeoutError) {
 						// The execution keeps running — only the wait timed out.
@@ -183,8 +181,6 @@ export const createExecuteWorkflowTool = (
 						output.status = 'error';
 						output.error = ensureError(waitError).message;
 					}
-				} finally {
-					progress.stop();
 				}
 			}
 

packages/cli/src/modules/mcp/tools/execution-utils.ts

@@ -2,7 +2,12 @@ import type { RequestHandlerExtra } from '@modelcontextprotocol/sdk/shared/proto
 import type { ServerNotification, ServerRequest } from '@modelcontextprotocol/sdk/types.js';
 import { Time } from '@n8n/constants';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
-import { UnexpectedError, TimeoutExecutionCancelledError, type IRun } from 'n8n-workflow';
+import {
+	UnexpectedError,
+	TimeoutExecutionCancelledError,
+	type ExecutionStatus,
+	type IRun,
+} from 'n8n-workflow';
 
 import type { ActiveExecutions } from '@/active-executions';
 import type { McpService } from '@/modules/mcp/mcp.service';
@@ -13,7 +18,9 @@ export const WORKFLOW_EXECUTION_TIMEOUT_MS = 5 * Time.minutes.toMilliseconds;
 
 export const PROGRESS_HEARTBEAT_INTERVAL_MS = 3 * Time.seconds.toMilliseconds;
 
-type ToolHandlerExtra = RequestHandlerExtra<ServerRequest, ServerNotification>;
+export type ToolHandlerExtra = RequestHandlerExtra<ServerRequest, ServerNotification>;
+
+const noop = () => {};
 
 /**
  * Streams MCP progress notifications while a workflow execution is awaited.
@@ -21,35 +28,51 @@ type ToolHandlerExtra = RequestHandlerExtra<ServerRequest, ServerNotification>;
  * `total` is deliberately omitted: execution duration is unknown, and a total
  * would make clients render a meaningless completion percentage.
  */
-export const createExecutionProgressReporter = (
-	extra: ToolHandlerExtra | undefined,
-	label: string,
-) => {
-	const progressToken = extra?._meta?.progressToken;
+export const createExecutionProgressReporter = (extra: ToolHandlerExtra, label: string) => {
+	const progressToken = extra._meta?.progressToken;
+	if (progressToken === undefined) return { start: noop, stop: noop };
+
 	let elapsedSeconds = 0;
 	let heartbeat: NodeJS.Timeout | undefined;
 
 	const send = (message: string) => {
-		if (progressToken === undefined) return;
 		void extra
-			?.sendNotification({
+			.sendNotification({
 				method: 'notifications/progress',
 				params: { progressToken, progress: elapsedSeconds, message },
 			})
 			.catch(() => {}); // Client may have disconnected — progress is best-effort
 	};
 
+	const stop = () => clearInterval(heartbeat);
+
 	return {
 		start: () => {
-			if (progressToken === undefined) return;
 			send(`${label} started`);
 			heartbeat = setInterval(() => {
 				elapsedSeconds += PROGRESS_HEARTBEAT_INTERVAL_MS * Time.milliseconds.toSeconds;
 				send(`${label} running — ${elapsedSeconds}s elapsed`);
 			}, PROGRESS_HEARTBEAT_INTERVAL_MS);
+			// Progress is best-effort: never keep the process alive, and stop as
+			// soon as the client disconnects or cancels the request.
+			heartbeat.unref();
+			extra.signal.addEventListener('abort', stop, { once: true });
 		},
-		update: send,
-		stop: () => clearInterval(heartbeat),
+		stop,
+	};
+};
+
+/**
+ * Derive the tool-facing outcome of a finished execution.
+ * Shared between execute_workflow and test_workflow tools.
+ */
+export const getExecutionOutcome = (data: IRun): { status: ExecutionStatus; error?: string } => {
+	const hasError = data.status === 'error' || data.data.resultData?.error;
+	return {
+		status: hasError ? 'error' : data.status,
+		error: hasError
+			? (data.data.resultData?.error?.message ?? 'Execution completed with errors')
+			: undefined,
 	};
 };
 
@@ -59,13 +82,26 @@ export const createExecutionProgressReporter = (
  * `cancelOnTimeout: false` leaves the execution running when the wait times
  * out — used by execute_workflow, where the wait is a convenience and must not
  * kill a (possibly production) run as a side effect.
+ * When `progress` is given, heartbeat progress notifications are streamed to
+ * the client for the duration of the wait.
  */
 export const waitForExecutionResult = async (
 	executionId: string,
 	activeExecutions: ActiveExecutions,
 	mcpService: McpService,
-	{ cancelOnTimeout = true }: { cancelOnTimeout?: boolean } = {},
+	{
+		cancelOnTimeout,
+		progress,
+	}: {
+		cancelOnTimeout: boolean;
+		progress?: { extra: ToolHandlerExtra; label: string };
+	},
 ): Promise<IRun> => {
+	const reporter = progress
+		? createExecutionProgressReporter(progress.extra, progress.label)
+		: undefined;
+	reporter?.start();
+
 	let timeoutId: NodeJS.Timeout | undefined;
 	const timeoutPromise = new Promise<never>((_, reject) => {
 		timeoutId = setTimeout(() => {
@@ -114,5 +150,7 @@ export const waitForExecutionResult = async (
 			}
 		}
 		throw error;
+	} finally {
+		reporter?.stop();
 	}
 };

packages/cli/src/modules/mcp/tools/test-workflow.tool.ts

@@ -24,9 +24,10 @@ import { USER_CALLED_MCP_TOOL_EVENT } from '../mcp.constants';
 import { McpExecutionTimeoutError, WorkflowAccessError } from '../mcp.errors';
 import type { ToolDefinition, UserCalledMCPToolEventPayload } from '../mcp.types';
 import {
-	createExecutionProgressReporter,
+	getExecutionOutcome,
 	waitForExecutionResult,
 	WORKFLOW_EXECUTION_TIMEOUT_MS,
+	type ToolHandlerExtra,
 } from './execution-utils';
 import { getMcpWorkflow } from './workflow-validation.utils';
 
@@ -93,9 +94,6 @@ export const createTestWorkflowTool = (
 			},
 		};
 
-		const progress = createExecutionProgressReporter(extra, `Test of workflow ${workflowId}`);
-		progress.start();
-
 		try {
 			const output = await testWorkflow(
 				user,
@@ -107,7 +105,7 @@ export const createTestWorkflowTool = (
 				workflowId,
 				pinData as IPinData,
 				triggerNodeName,
-				(executionId) => progress.update(`Execution ${executionId} running`),
+				extra,
 			);
 
 			telemetryPayload.results = {
@@ -147,8 +145,6 @@ export const createTestWorkflowTool = (
 				content: [{ type: 'text', text: jsonStringify(output) }],
 				structuredContent: output,
 			};
-		} finally {
-			progress.stop();
 		}
 	},
 });
@@ -167,7 +163,7 @@ export async function testWorkflow(
 	workflowId: string,
 	pinData: IPinData,
 	triggerNodeName?: string,
-	onExecutionStarted?: (executionId: string) => void,
+	extra?: ToolHandlerExtra,
 ): Promise<TestWorkflowOutput> {
 	const workflow = await getMcpWorkflow(
 		workflowId,
@@ -245,17 +241,12 @@ export async function testWorkflow(
 	};
 
 	const executionId = await workflowRunner.run(runData);
-	onExecutionStarted?.(executionId);
-	const data = await waitForExecutionResult(executionId, activeExecutions, mcpService);
-	const hasError = data.status === 'error' || data.data.resultData?.error;
+	const data = await waitForExecutionResult(executionId, activeExecutions, mcpService, {
+		cancelOnTimeout: true,
+		progress: extra && { extra, label: `Execution ${executionId} of workflow ${workflowId}` },
+	});
 
-	return {
-		executionId,
-		status: hasError ? 'error' : data.status,
-		error: hasError
-			? (data.data.resultData?.error?.message ?? 'Execution completed with errors')
-			: undefined,
-	};
+	return { executionId, ...getExecutionOutcome(data) };
 }
 
 /**