import type { Logger } from '@n8n/backend-common';
import type { ExecutionsConfig } from '@n8n/config';
import type { IExecutionResponse, ExecutionRepository, Project } from '@n8n/db';
import { WorkflowPublishHistoryRepository } from '@n8n/db';
import type {
	WorkflowExecute as ActualWorkflowExecute,
	BinaryDataConfig,
	BinaryDataService,
	InstanceSettings,
} from 'n8n-core';
import { ExternalSecretsProxy } from 'n8n-core';
import { mockInstance } from 'n8n-core/test/utils';
import {
	type IPinData,
	type IRun,
	type ISupplyDataFunctions,
	type ITaskData,
	type IDataObject,
	type INodeExecutionData,
	type IWorkflowExecuteAdditionalData,
	Workflow,
	NodeConnectionTypes,
	NodeOperationError,
	createRunExecutionData,
	type IRunExecutionData,
	type WorkflowExecuteMode,
	type ExecutionError,
	WorkflowExpression,
} from 'n8n-workflow';
import type { Mock, MockedClass, MockInstance } from 'vitest';
import { mock } from 'vitest-mock-extended';

import { CredentialsHelper } from '@/credentials-helper';
import { VariablesService } from '@/environments.ee/variables/variables.service.ee';
import { WebhookResponseTooLargeError } from '@/errors/webhook-response-too-large.error';
import { ExternalHooks } from '@/external-hooks';
import type { ExecutionPersistence } from '@/executions/execution-persistence';
import type { ManualExecutionService } from '@/manual-execution.service';
import { DataTableProxyService } from '@/modules/data-table/data-table-proxy.service';
import type { NodeTypes } from '@/node-types';
import { OwnershipService } from '@/services/ownership.service';
import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
import { WorkflowHookContextService } from '@/workflow-hook-context.service';
import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';

import { JobProcessor } from '../job-processor';
import type { Job } from '../scaling.types';
import { ENCODED_BUFFER_KEY, WebhookResponseRelay } from '../webhook-response-relay';

mockInstance(WorkflowPublishHistoryRepository);
mockInstance(VariablesService, {
	getAllCached: vi.fn().mockResolvedValue([]),
});
mockInstance(CredentialsHelper);
mockInstance(ExternalSecretsProxy);
mockInstance(WorkflowStaticDataService);
mockInstance(WorkflowStatisticsService);
mockInstance(ExternalHooks);
mockInstance(WorkflowHookContextService);
mockInstance(DataTableProxyService);
mockInstance(OwnershipService, {
	getWorkflowProjectCached: vi.fn().mockResolvedValue(mock<Project>({ id: 'project-id' })),
});

const processRunExecutionDataMock = vi.fn();
vi.mock('n8n-core', async () => {
	const original = await vi.importActual<typeof import('n8n-core')>('n8n-core');

	// Mock class constructor and prototype methods
	return {
		...original,
		WorkflowExecute: vi.fn(function () {
			return { processRunExecutionData: processRunExecutionDataMock };
		}),
	};
});

const logger = mock<Logger>({
	scoped: vi.fn().mockImplementation(() => logger),
});

const executionsConfig = mock<ExecutionsConfig>({
	timeout: -1,
	maxTimeout: 3600,
});

const successRun = (): IRun =>
	mock<IRun>({
		status: 'success',
		stoppedAt: new Date(),
		data: mock<IRunExecutionData>({
			resultData: { runData: {}, error: undefined },
			executionData: undefined,
		}),
	});

const errorRun = (error: ExecutionError): IRun =>
	mock<IRun>({
		status: 'error',
		stoppedAt: new Date(),
		data: mock<IRunExecutionData>({
			resultData: { runData: {}, error },
			executionData: undefined,
		}),
	});

const createManualExecutionServiceMock = (run: IRun = successRun()): ManualExecutionService => {
	const svc = mock<ManualExecutionService>();
	svc.runManually.mockReturnValue(Promise.resolve(run) as ReturnType<typeof svc.runManually>);
	return svc;
};

describe('JobProcessor', () => {
	beforeEach(() => {
		processRunExecutionDataMock.mockReset();
		processRunExecutionDataMock.mockResolvedValue(successRun());
	});

	it('should refrain from processing a crashed execution', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(
			mock<IExecutionResponse>({ status: 'crashed' }),
		);
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			mock(),
			executionsConfig,
			mock(),
			mock(),
		);

		const result = await jobProcessor.processJob(mock<Job>());

		expect(result).toEqual({ success: false });
	});

	it('should throw a descriptive error when the execution has no run data', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(
			mock<IExecutionResponse>({
				id: 'execution-id',
				mode: 'trigger',
				workflowData: { nodes: [] },
				data: undefined,
			}),
		);

		const manualExecutionService = createManualExecutionServiceMock();
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const job = mock<Job>({ data: { executionId: 'execution-id', loadStaticData: false } });

		await expect(jobProcessor.processJob(job)).rejects.toThrow(/without run data/);
		expect(manualExecutionService.runManually).not.toHaveBeenCalled();
	});

	it.each(['manual', 'evaluation'] satisfies WorkflowExecuteMode[])(
		'should use manualExecutionService to process a job in %p mode',
		async (mode) => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValue(
				mock<IExecutionResponse>({
					mode,
					workflowData: { nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mock(),
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();

			await jobProcessor.processJob(job);

			expect(manualExecutionService.runManually).toHaveBeenCalledTimes(1);

			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'job-finished',
					success: true,
				}),
			);
		},
	);

	it('should remove the running job entry when the workflow run rejects', async () => {
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(
			mock<IExecutionResponse>({
				mode: 'manual',
				workflowData: { nodes: [], staticData: {} },
				data: mock<IRunExecutionData>({
					executionData: undefined,
				}),
			}),
		);

		const manualExecutionService = mock<ManualExecutionService>();
		manualExecutionService.runManually.mockReturnValue(
			Promise.reject(new Error('workflow run rejected')) as ReturnType<
				ManualExecutionService['runManually']
			>,
		);

		const jobProcessor = new JobProcessor(
			logger,
			mock<ExecutionRepository>(),
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const job = mock<Job>({ id: 'job-1' });

		await expect(jobProcessor.processJob(job)).rejects.toThrow('workflow run rejected');

		expect(jobProcessor.getRunningJobIds()).toEqual([]);
	});

	it('should send job-finished with success=false when execution has errors', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValueOnce(
			mock<IExecutionResponse>({
				mode: 'manual',
				workflowData: { nodes: [], staticData: {} },
				data: mock<IRunExecutionData>({
					executionData: undefined,
				}),
			}),
		);

		const manualExecutionService = createManualExecutionServiceMock(
			errorRun(mock<ExecutionError>()),
		);
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const job = mock<Job>();

		await jobProcessor.processJob(job);

		expect(job.progress).toHaveBeenCalledWith(
			expect.objectContaining({
				kind: 'job-finished',
				success: false,
			}),
		);
	});

	it('should pass additional data for partial executions to run', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const pinData: IPinData = { pinned: [] };
		const execution = mock<IExecutionResponse>({
			mode: 'manual',
			workflowData: { id: 'workflow-id', nodes: [], pinData, staticData: {} },
			data: mock<IRunExecutionData>({
				resultData: {
					runData: {
						trigger: [mock<ITaskData>({ executionIndex: 1 })],
						node: [mock<ITaskData>({ executionIndex: 3 }), mock<ITaskData>({ executionIndex: 4 })],
					},
					pinData,
				},
				executionData: undefined,
			}),
		});
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(execution);

		const additionalData = mock<IWorkflowExecuteAdditionalData>();
		vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);

		const manualExecutionService = createManualExecutionServiceMock();
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const executionId = 'execution-id';
		await jobProcessor.processJob(mock<Job>({ data: { executionId, loadStaticData: false } }));

		expect(WorkflowExecuteAdditionalData.getBase).toHaveBeenCalledWith({
			workflowId: execution.workflowData.id,
			executionTimeoutTimestamp: undefined,
			workflowSettings: execution.workflowData.settings,
		});

		expect(manualExecutionService.runManually).toHaveBeenCalledWith(
			expect.objectContaining({
				executionMode: 'manual',
			}),
			expect.any(Workflow),
			additionalData,
			executionId,
			pinData,
		);
	});

	it('should set restartExecutionId on additionalData when provided in job data', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const execution = mock<IExecutionResponse>({
			mode: 'manual',
			workflowData: { id: 'workflow-id', nodes: [], staticData: {} },
			data: mock<IRunExecutionData>({
				resultData: { runData: {} },
				executionData: undefined,
			}),
		});
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(execution);

		const additionalData = mock<IWorkflowExecuteAdditionalData>();
		vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);

		const manualExecutionService = createManualExecutionServiceMock();
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const executionId = 'execution-id';
		const restartExecutionId = 'restart-execution-id';
		await jobProcessor.processJob(
			mock<Job>({ data: { executionId, loadStaticData: false, restartExecutionId } }),
		);

		expect(additionalData.restartExecutionId).toBe(restartExecutionId);
	});

	it('should rehydrate the manual-execution identity onto additionalData from job data', async () => {
		const executionRepository = mock<ExecutionRepository>();
		const execution = mock<IExecutionResponse>({
			mode: 'manual',
			workflowData: { id: 'workflow-id', nodes: [], staticData: {} },
			data: mock<IRunExecutionData>({
				resultData: { runData: {} },
				executionData: undefined,
			}),
		});
		const executionPersistence = mock<ExecutionPersistence>();
		executionPersistence.findSingleExecution.mockResolvedValue(execution);

		const additionalData = mock<IWorkflowExecuteAdditionalData>();
		vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);

		const manualExecutionService = createManualExecutionServiceMock();
		const jobProcessor = new JobProcessor(
			logger,
			executionRepository,
			executionPersistence,
			mock(),
			mock(),
			mock(),
			manualExecutionService,
			executionsConfig,
			mock(),
			mock(),
		);

		const executionId = 'execution-id';
		const encryptedRunnerIdentity = 'encrypted-identity-blob';
		await jobProcessor.processJob(
			mock<Job>({ data: { executionId, loadStaticData: false, encryptedRunnerIdentity } }),
		);

		expect(additionalData.encryptedRunnerIdentity).toBe(encryptedRunnerIdentity);
	});

	it.each(['manual', 'evaluation', 'trigger'] satisfies WorkflowExecuteMode[])(
		'should use workflowExecute to process a job with mode %p with execution data',
		async (mode) => {
			const { WorkflowExecute } = await import('n8n-core');
			// Type it correctly so we can use mock methods later
			const MockedWorkflowExecute = WorkflowExecute as MockedClass<typeof ActualWorkflowExecute>;
			MockedWorkflowExecute.mockClear();

			const executionRepository = mock<ExecutionRepository>();
			const executionData = mock<IRunExecutionData>({
				startData: undefined,
				executionData: {
					nodeExecutionStack: [
						{
							node: { name: 'node-name' },
						},
					],
				},
			});
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValue(
				mock<IExecutionResponse>({
					mode,
					workflowData: { nodes: [], staticData: {} },
					data: executionData,
				}),
			);

			const additionalData = mock<IWorkflowExecuteAdditionalData>();
			vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);

			const manualExecutionService = createManualExecutionServiceMock();
			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mock(),
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			await jobProcessor.processJob(mock<Job>());

			// Assert the constructor and method were called
			expect(MockedWorkflowExecute).toHaveBeenCalledWith(additionalData, mode, executionData);
			expect(processRunExecutionDataMock).toHaveBeenCalled();
		},
	);

	describe('MCP execution support', () => {
		it('should send mcp-response message for MCP executions after job completion', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			// Second call for checking errors
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(), // workflowRepository
				mock(), // nodeTypes
				mcpInstanceSettings, // instanceSettings
				manualExecutionService,
				executionsConfig,
				mock(), // eventService
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-123',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'service',
				mcpSessionId: 'session-456',
				mcpMessageId: 'msg-789',
			};

			await jobProcessor.processJob(job);

			// Should have called progress with mcp-response
			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'mcp-response',
					executionId: 'exec-mcp-123',
					mcpType: 'service',
					sessionId: 'session-456',
					messageId: 'msg-789',
					workerId: 'worker-host-123',
				}),
			);
		});

		it('should not send mcp-response for non-MCP executions', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mock(),
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-regular-123',
				loadStaticData: false,
				isMcpExecution: undefined,
				mcpSessionId: undefined,
			};

			await jobProcessor.processJob(job);

			const progressCalls = (job.progress as Mock).mock.calls;
			const mcpResponseCalls = progressCalls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			expect(mcpResponseCalls).toHaveLength(0);
		});

		it('should include success=false in mcp-response when execution has errors', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock(
				errorRun({ message: 'Test error' } as ExecutionError),
			);
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(), // workflowRepository
				mock(), // nodeTypes
				mcpInstanceSettings, // instanceSettings
				manualExecutionService,
				executionsConfig,
				mock(), // eventService
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-error',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'service',
				mcpSessionId: 'session-456',
				mcpMessageId: 'msg-789',
			};

			await jobProcessor.processJob(job);

			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'mcp-response',
					response: expect.objectContaining({
						success: false,
					}),
				}),
			);
		});

		it('should send mcp-response for MCP Trigger executions without tool call', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			// Second call for fetching result
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(), // workflowRepository
				mock(), // nodeTypes
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(), // eventService
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-trigger',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger', // MCP Trigger type without tool call
				mcpSessionId: 'session-789',
				mcpMessageId: 'msg-456',
				mcpToolCall: undefined, // No tool call
			};

			await jobProcessor.processJob(job);

			// Should send mcp-response even for trigger type without tool call
			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'mcp-response',
					executionId: 'exec-mcp-trigger',
					mcpType: 'trigger',
					sessionId: 'session-789',
					messageId: 'msg-456',
					workerId: 'worker-host-123',
				}),
			);
		});

		it('should execute tool node and send result for MCP Trigger with tool call', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = { name: 'tool-node', type: 'n8n-nodes-base.tool' };
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: {
						id: 'wf-1',
						nodes: [toolNode],
						staticData: {},
					},
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			// Second call for fetching result
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			// Mock the processRunExecutionData to return tool result
			processRunExecutionDataMock.mockResolvedValue({
				data: {
					resultData: {
						runData: {
							'tool-node': [
								{
									data: {
										main: [[{ json: { toolResult: 'success' } }]],
									},
								},
							],
						},
					},
				},
			});

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(), // workflowRepository
				mock(), // nodeTypes
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(), // eventService
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-tool',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-tool',
				mcpMessageId: 'msg-tool',
				mcpToolCall: {
					toolName: 'test-tool',
					arguments: { arg1: 'value1' },
					sourceNodeName: 'tool-node',
				},
			};

			await jobProcessor.processJob(job);

			// Should send mcp-response with tool result
			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			expect(mcpResponseCalls.length).toBeGreaterThan(0);
			expect(mcpResponseCalls[mcpResponseCalls.length - 1][0]).toMatchObject({
				kind: 'mcp-response',
				executionId: 'exec-mcp-tool',
				mcpType: 'trigger',
				sessionId: 'session-tool',
				messageId: 'msg-tool',
				workerId: 'worker-host-123',
			});
		});

		it('should handle tool execution errors gracefully', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: {
						id: 'wf-1',
						nodes: [], // Missing tool node will cause error
						staticData: {},
					},
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-error',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-error',
				mcpMessageId: 'msg-error',
				mcpToolCall: {
					toolName: 'missing-tool',
					arguments: {},
					sourceNodeName: 'nonexistent-node',
				},
			};

			await jobProcessor.processJob(job);

			// Should send mcp-response with error (error object without stack trace)
			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'mcp-response',
					response: expect.objectContaining({
						error: expect.objectContaining({
							message: expect.any(String),
						}),
					}),
				}),
			);

			// Verify no stack trace is exposed
			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			const lastResponse = mcpResponseCalls[mcpResponseCalls.length - 1][0] as {
				response: { error?: { stack?: string } };
			};
			expect(lastResponse.response.error?.stack).toBeUndefined();
		});

		it('should invoke tool via execute for tool wrapper nodes without supplyData', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'HTTP Request',
				type: 'n8n-nodes-base.httpRequestTool',
				typeVersion: 4.4,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: {
						id: 'wf-1',
						nodes: [toolNode],
						staticData: {},
					},
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			// Mock nodeTypes to return a node without supplyData but with AiTool output
			const mockExecuteFn = vi
				.fn()
				.mockResolvedValue([[{ json: { result: 'tool response data' } }]]);
			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'httpRequestTool',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: mockExecuteFn,
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-tool-wrapper',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-tool',
				mcpMessageId: 'msg-tool',
				mcpToolCall: {
					toolName: 'HTTP Request',
					arguments: { url: 'https://example.com' },
					sourceNodeName: 'HTTP Request',
				},
			};

			await jobProcessor.processJob(job);

			// execute should have been called (not supplyData)
			expect(mockExecuteFn).toHaveBeenCalled();

			// Should send mcp-response with tool result
			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			expect(mcpResponseCalls.length).toBeGreaterThan(0);
			const lastResponse = mcpResponseCalls[mcpResponseCalls.length - 1][0] as {
				kind: string;
				mcpType: string;
				response: unknown;
			};
			expect(lastResponse).toMatchObject({
				kind: 'mcp-response',
				mcpType: 'trigger',
			});
			// Response should contain the tool's output data
			expect(lastResponse.response).toEqual([{ result: 'tool response data' }]);
		});

		it('should relay a tool result too large for the queue as an error', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'HTTP Request',
				type: 'n8n-nodes-base.httpRequestTool',
				typeVersion: 4.4,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'httpRequestTool',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockResolvedValue([[{ json: { blob: 'x'.repeat(64) } }]]),
			} as never);

			const relay = mock<WebhookResponseRelay>();
			relay.assertFitsInline.mockImplementation(() => {
				throw new WebhookResponseTooLargeError(
					'The response is too large to be sent back from the worker (limit is 1 MiB)',
					{ description: 'Raise N8N_WEBHOOK_RESPONSE_RELAY_SIZE_MAX.' },
				);
			});

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				{ hostId: 'worker-host-123' } as unknown as InstanceSettings,
				createManualExecutionServiceMock(),
				executionsConfig,
				mock(),
				relay,
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-tool-oversized',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-tool',
				mcpMessageId: 'msg-tool',
				mcpToolCall: {
					toolName: 'HTTP Request',
					arguments: { url: 'https://example.com' },
					sourceNodeName: 'HTTP Request',
				},
			};

			await jobProcessor.processJob(job);

			expect(relay.assertFitsInline).toHaveBeenCalledWith([{ blob: 'x'.repeat(64) }]);

			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			const lastResponse = mcpResponseCalls[mcpResponseCalls.length - 1][0] as {
				response: { error?: { message?: string } };
			};
			expect(lastResponse.response.error?.message).toContain('too large');
		});

		describe('expression isolate for tool calls', () => {
			const toolNode = {
				name: 'HTTP Request',
				type: 'n8n-nodes-base.httpRequestTool',
				typeVersion: 4.4,
				parameters: {},
				position: [0, 0] as [number, number],
			};

			const setupToolJob = (executeFn: Mock) => {
				const executionPersistence = mock<ExecutionPersistence>();
				executionPersistence.findSingleExecution.mockResolvedValueOnce(
					mock<IExecutionResponse>({
						mode: 'trigger',
						workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
						data: mock<IRunExecutionData>({ executionData: undefined }),
					}),
				);
				executionPersistence.findSingleExecution.mockResolvedValueOnce(
					mock<IExecutionResponse>({
						status: 'success',
						workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
						data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
					}),
				);

				const nodeTypes = mock<NodeTypes>();
				nodeTypes.getByNameAndVersion.mockReturnValue({
					description: {
						name: 'httpRequestTool',
						outputs: [NodeConnectionTypes.AiTool],
						properties: [],
					},
					execute: executeFn,
				} as never);

				const jobProcessor = new JobProcessor(
					logger,
					mock(), // executionRepository
					executionPersistence,
					mock(), // workflowRepository
					nodeTypes,
					{ hostId: 'worker-host-123' } as unknown as InstanceSettings,
					createManualExecutionServiceMock(),
					executionsConfig,
					mock(), // eventService
					mock(),
				);

				const job = mock<Job>();
				job.data = {
					workflowId: 'wf-1',
					executionId: 'exec-mcp-isolate',
					loadStaticData: false,
					isMcpExecution: true,
					mcpType: 'trigger',
					mcpSessionId: 'session-isolate',
					mcpMessageId: 'msg-isolate',
					mcpToolCall: {
						toolName: 'HTTP Request',
						arguments: { url: 'https://example.com' },
						sourceNodeName: 'HTTP Request',
					},
				};

				return { jobProcessor, job };
			};

			let acquireSpy: MockInstance;
			let releaseSpy: MockInstance;

			beforeEach(() => {
				acquireSpy = vi
					.spyOn(WorkflowExpression.prototype, 'acquireIsolate')
					.mockResolvedValue(true);
				releaseSpy = vi.spyOn(WorkflowExpression.prototype, 'releaseIsolate').mockResolvedValue();
			});

			afterEach(() => {
				acquireSpy.mockRestore();
				releaseSpy.mockRestore();
			});

			it('should acquire an isolate around the tool invocation and release it after', async () => {
				const executeFn = vi.fn().mockResolvedValue([[{ json: { ok: true } }]]);
				const { jobProcessor, job } = setupToolJob(executeFn);

				await jobProcessor.processJob(job);

				expect(acquireSpy).toHaveBeenCalledTimes(1);
				expect(releaseSpy).toHaveBeenCalledTimes(1);
				// The tool must run inside the acquire/release window
				expect(acquireSpy.mock.invocationCallOrder[0]).toBeLessThan(
					executeFn.mock.invocationCallOrder[0],
				);
				expect(releaseSpy.mock.invocationCallOrder[0]).toBeGreaterThan(
					executeFn.mock.invocationCallOrder[0],
				);
			});

			it('should release the isolate when the tool invocation throws', async () => {
				const executeFn = vi.fn().mockRejectedValue(new Error('tool failed'));
				const { jobProcessor, job } = setupToolJob(executeFn);

				await jobProcessor.processJob(job);

				expect(acquireSpy).toHaveBeenCalledTimes(1);
				expect(releaseSpy).toHaveBeenCalledTimes(1);
			});

			it('should not release an isolate it did not newly acquire', async () => {
				acquireSpy.mockResolvedValue(false);
				const executeFn = vi.fn().mockResolvedValue([[{ json: { ok: true } }]]);
				const { jobProcessor, job } = setupToolJob(executeFn);

				await jobProcessor.processJob(job);

				expect(acquireSpy).toHaveBeenCalledTimes(1);
				expect(releaseSpy).not.toHaveBeenCalled();
			});
		});

		it('should invoke tool via supplyData for nodes with supplyData method', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Tool HTTP Request',
				type: '@n8n/n8n-nodes-langchain.toolHttpRequest',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: {
						id: 'wf-1',
						nodes: [toolNode],
						staticData: {},
					},
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = {
				hostId: 'worker-host-123',
			} as unknown as InstanceSettings;

			// Mock a tool that supplyData returns
			const mockTool = { invoke: vi.fn().mockResolvedValue('supply data tool result') };
			const mockSupplyData = vi.fn().mockResolvedValue({ response: mockTool });

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolHttpRequest',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				supplyData: mockSupplyData,
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-supply-data',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-supply',
				mcpMessageId: 'msg-supply',
				mcpToolCall: {
					toolName: 'Tool HTTP Request',
					arguments: { query: 'test' },
					sourceNodeName: 'Tool HTTP Request',
				},
			};

			await jobProcessor.processJob(job);

			// supplyData path should have been used
			expect(mockSupplyData).toHaveBeenCalled();
			expect(mockTool.invoke).toHaveBeenCalledWith({ query: 'test' });

			// Should send mcp-response with tool result
			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			expect(mcpResponseCalls.length).toBeGreaterThan(0);
			const lastResponse = mcpResponseCalls[mcpResponseCalls.length - 1][0] as {
				response: unknown;
			};
			expect(lastResponse.response).toBe('supply data tool result');
		});

		describe('MCP request context on the worker', () => {
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};

			const mcpToolInput = { method: 'tools/call', headers: { 'x-user-id': 'user-1' } };

			/** `triggerNames` is in connection order; only `triggerThatRan` gets run data. */
			const runToolCall = async ({
				triggerNames,
				triggerThatRan,
				toolInput,
			}: {
				triggerNames: string[];
				triggerThatRan: string;
				toolInput?: IDataObject;
			}) => {
				const workflowData = {
					id: 'wf-1',
					nodes: [
						...triggerNames.map((name) => ({
							name,
							type: '@n8n/n8n-nodes-langchain.mcpTrigger',
							typeVersion: 2,
							parameters: {},
							position: [0, 0] as [number, number],
						})),
						toolNode,
					],
					connections: {
						[toolNode.name]: {
							[NodeConnectionTypes.AiTool]: [
								triggerNames.map((node) => ({ node, type: NodeConnectionTypes.AiTool, index: 0 })),
							],
						},
					},
					staticData: {},
				};

				const executionPersistence = mock<ExecutionPersistence>();
				executionPersistence.findSingleExecution.mockResolvedValueOnce(
					mock<IExecutionResponse>({
						mode: 'trigger',
						workflowData,
						data: mock<IRunExecutionData>({ executionData: undefined }),
					}),
				);
				executionPersistence.findSingleExecution.mockResolvedValueOnce(
					mock<IExecutionResponse>({ status: 'success', workflowData }),
				);

				// Real (mutable) run-execution-data, not a mock, so the tool node's own run
				// lands in `resultData.runData` next to the trigger's.
				const run: IRun = {
					mode: 'trigger',
					status: 'success',
					finished: true,
					startedAt: new Date(),
					stoppedAt: new Date(),
					storedAt: 'db',
					data: createRunExecutionData({
						resultData: { runData: { [triggerThatRan]: [mock<ITaskData>()] } },
					}),
				};

				let toolNodeInput: INodeExecutionData[] | undefined;
				const nodeTypes = mock<NodeTypes>();
				nodeTypes.getByNameAndVersion.mockReturnValue({
					description: {
						name: 'toolCalculator',
						outputs: [NodeConnectionTypes.AiTool],
						properties: [],
					},
					execute: vi.fn().mockImplementation(function (this: ISupplyDataFunctions) {
						toolNodeInput = this.getInputData(0, NodeConnectionTypes.Main);
						return [[{ json: { response: 42 } }]];
					}),
				} as never);

				const jobProcessor = new JobProcessor(
					logger,
					mock(), // executionRepository
					executionPersistence,
					mock(), // workflowRepository
					nodeTypes,
					{ hostId: 'worker-host-123' } as unknown as InstanceSettings,
					createManualExecutionServiceMock(run),
					executionsConfig,
					mock(), // eventService
					mock(), // webhookResponseRelay
				);

				const job = mock<Job>();
				job.data = {
					workflowId: 'wf-1',
					executionId: 'exec-mcp-input',
					loadStaticData: false,
					isMcpExecution: true,
					mcpType: 'trigger',
					mcpSessionId: 'session-input',
					mcpMessageId: 'msg-1',
					mcpToolCall: {
						toolName: toolNode.name,
						arguments: { input: '2 + 2' },
						sourceNodeName: toolNode.name,
					},
					mcpToolInput: toolInput,
				};

				await jobProcessor.processJob(job);

				return { toolNodeInput, toolRun: run.data.resultData.runData[toolNode.name]?.[0] };
			};

			it.each([
				{ toolInput: mcpToolInput, expected: mcpToolInput },
				{ toolInput: undefined, expected: {} },
			])('should give the tool node $toolInput as input data', async ({ toolInput, expected }) => {
				const { toolNodeInput } = await runToolCall({
					triggerNames: ['MCP Server Trigger'],
					triggerThatRan: 'MCP Server Trigger',
					toolInput,
				});

				expect(toolNodeInput).toEqual([{ json: expected }]);
			});

			it.each(['MCP Trigger A', 'MCP Trigger B'])(
				'should record the tool run against the trigger that ran (%s)',
				async (triggerThatRan) => {
					const { toolRun } = await runToolCall({
						triggerNames: ['MCP Trigger A', 'MCP Trigger B'],
						triggerThatRan,
						toolInput: mcpToolInput,
					});

					expect(toolRun?.source?.[0]?.previousNode).toBe(triggerThatRan);
				},
			);
		});

		it('should expose the established execution context to the tool node', async () => {
			// Regression: in queue mode the tool runs on the worker, where dynamic
			// (private) credentials resolve from the node's execution context. The
			// context established on the main travels with the loaded execution and
			// must be threaded into the tool node, or resolution fails with
			// MissingExecutionContextError.
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Tool HTTP Request',
				type: '@n8n/n8n-nodes-langchain.toolHttpRequest',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};

			const runtimeContext = {
				version: 1,
				establishedAt: 0,
				source: 'webhook',
				credentials: 'encrypted-identity-blob',
			};

			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: { runtimeData: runtimeContext } as never,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			let capturedContext: unknown;
			const mockTool = { invoke: vi.fn().mockResolvedValue('ok') };
			const mockSupplyData = vi.fn().mockImplementation(function (this: {
				getExecutionContext: () => unknown;
			}) {
				capturedContext = this.getExecutionContext();
				return { response: mockTool };
			});

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolHttpRequest',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				supplyData: mockSupplyData,
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-ctx',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-ctx',
				mcpMessageId: 'msg-ctx',
				mcpToolCall: {
					toolName: 'Tool HTTP Request',
					arguments: { query: 'x' },
					sourceNodeName: 'Tool HTTP Request',
				},
			};

			await jobProcessor.processJob(job);

			expect(mockSupplyData).toHaveBeenCalled();
			expect(capturedContext).toEqual(runtimeContext);
		});

		it('persists the tool node run data into the execution after a tool call', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			const triggerNode = {
				name: 'MCP Server Trigger',
				type: '@n8n/n8n-nodes-langchain.mcpTrigger',
				typeVersion: 1,
				parameters: {},
				position: [200, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: {
						id: 'wf-1',
						nodes: [toolNode, triggerNode],
						connections: {
							Calculator: {
								[NodeConnectionTypes.AiTool]: [
									[{ node: 'MCP Server Trigger', type: NodeConnectionTypes.AiTool, index: 0 }],
								],
							},
						},
						staticData: {},
					},
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			// The completed workflow run must carry a real (mutable) run-execution-data
			// object, so the tool node's input/output actually land in resultData.runData.
			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({}),
			};
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockResolvedValue([[{ json: { response: 42 } }]]),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-record',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-record',
				mcpMessageId: 'msg-record',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: '2 + 2' },
					sourceNodeName: 'Calculator',
				},
			};

			await jobProcessor.processJob(job);

			expect(run.data.resultData.runData.Calculator?.[0]?.executionStatus).toBe('success');
			expect(run.data.resultData.runData.Calculator?.[0]?.source?.[0]?.previousNode).toBe(
				'MCP Server Trigger',
			);
			expect(executionPersistence.updateExistingExecution).toHaveBeenCalledWith(
				'exec-mcp-record',
				expect.objectContaining({
					data: expect.objectContaining({
						resultData: expect.objectContaining({
							runData: expect.objectContaining({ Calculator: expect.any(Array) }),
						}),
					}),
					usedPrivateCredentials: false,
				}),
				// A cancel racing the tool call must keep its status.
				{ requireNotCanceled: true },
			);
		});

		it('records the tool node as errored when the tool call fails', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({}),
			};
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockRejectedValue(new Error('boom')),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-error',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-error',
				mcpMessageId: 'msg-error',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: 'bad' },
					sourceNodeName: 'Calculator',
				},
			};

			await jobProcessor.processJob(job);

			expect(run.data.resultData.runData.Calculator?.[0]?.executionStatus).toBe('error');
			expect(executionPersistence.updateExistingExecution).toHaveBeenCalledWith(
				'exec-mcp-error',
				expect.objectContaining({
					data: expect.objectContaining({
						resultData: expect.objectContaining({
							runData: expect.objectContaining({ Calculator: expect.any(Array) }),
						}),
					}),
				}),
				{ requireNotCanceled: true },
			);
		});

		it('persists run data for a supplyData tool that records via its context', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({}),
			};
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			// Native langchain tools return a `logWrapper`-wrapped tool from supplyData that
			// records into the bound context on invoke. Simulate that here to prove the
			// supplyData branch's recording lands in the threaded run data (not a throwaway).
			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				supplyData: vi.fn(async function (this: ISupplyDataFunctions) {
					return {
						response: {
							invoke: async (args: IDataObject) => {
								const { index } = this.addInputData(NodeConnectionTypes.AiTool, [[{ json: args }]]);
								this.addOutputData(NodeConnectionTypes.AiTool, index, [
									[{ json: { response: 4 } }],
								]);
								return '4';
							},
						},
					};
				}),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-supply',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-supply',
				mcpMessageId: 'msg-supply',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: '2 + 2' },
					sourceNodeName: 'Calculator',
				},
			};

			await jobProcessor.processJob(job);

			expect(run.data.resultData.runData.Calculator?.[0]?.executionStatus).toBe('success');
			expect(executionPersistence.updateExistingExecution).toHaveBeenCalledWith(
				'exec-mcp-supply',
				expect.objectContaining({
					data: expect.objectContaining({
						resultData: expect.objectContaining({
							runData: expect.objectContaining({ Calculator: expect.any(Array) }),
						}),
					}),
				}),
				{ requireNotCanceled: true },
			);
		});

		it('still returns the tool result to the client when persisting the run data fails', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);
			executionPersistence.updateExistingExecution.mockRejectedValueOnce(new Error('db down'));

			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({}),
			};
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockResolvedValue([[{ json: { response: 42 } }]]),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-persist-fail',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-persist-fail',
				mcpMessageId: 'msg-persist-fail',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: '2 + 2' },
					sourceNodeName: 'Calculator',
				},
			};

			await expect(jobProcessor.processJob(job)).resolves.toEqual({ success: true });

			expect(logger.error).toHaveBeenCalledWith(
				'Failed to persist tool call run data for MCP Trigger',
				expect.objectContaining({
					executionId: 'exec-mcp-persist-fail',
					sourceNodeName: 'Calculator',
				}),
			);

			// The mcp-response with the tool result must still reach the client.
			const mcpResponseCalls = (job.progress as Mock).mock.calls.filter(
				(call: unknown[]) => (call[0] as { kind: string }).kind === 'mcp-response',
			);
			expect(mcpResponseCalls[mcpResponseCalls.length - 1][0]).toMatchObject({
				kind: 'mcp-response',
				executionId: 'exec-mcp-persist-fail',
			});
		});

		it('initializes execution data when the completed run lacks it', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			// `executionData: null` leaves the run without an executionData object, forcing
			// invokeTool to initialize it before recording the tool call.
			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({ executionData: null }),
			};
			expect(run.data.executionData).toBeUndefined();
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockResolvedValue([[{ json: { response: 42 } }]]),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-init-data',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-init-data',
				mcpMessageId: 'msg-init-data',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: '2 + 2' },
					sourceNodeName: 'Calculator',
				},
			};

			await jobProcessor.processJob(job);

			expect(run.data.executionData).toBeDefined();
			expect(run.data.resultData.runData.Calculator?.[0]?.executionStatus).toBe('success');
		});

		it('records a NodeOperationError from the tool without re-wrapping it', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			const toolNode = {
				id: 'calc-node',
				name: 'Calculator',
				type: '@n8n/n8n-nodes-langchain.toolCalculator',
				typeVersion: 1,
				parameters: {},
				position: [0, 0] as [number, number],
			};
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'trigger',
					workflowData: { id: 'wf-1', nodes: [toolNode], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			const run: IRun = {
				mode: 'trigger',
				status: 'success',
				finished: true,
				startedAt: new Date(),
				stoppedAt: new Date(),
				storedAt: 'db',
				data: createRunExecutionData({}),
			};
			const manualExecutionService = createManualExecutionServiceMock(run);
			const mcpInstanceSettings = { hostId: 'worker-host-123' } as unknown as InstanceSettings;

			const toolError = new NodeOperationError(toolNode, 'boom');
			const nodeTypes = mock<NodeTypes>();
			nodeTypes.getByNameAndVersion.mockReturnValue({
				description: {
					name: 'toolCalculator',
					outputs: [NodeConnectionTypes.AiTool],
					properties: [],
				},
				execute: vi.fn().mockRejectedValue(toolError),
			} as never);

			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				nodeTypes,
				mcpInstanceSettings,
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-mcp-node-error',
				loadStaticData: false,
				isMcpExecution: true,
				mcpType: 'trigger',
				mcpSessionId: 'session-node-error',
				mcpMessageId: 'msg-node-error',
				mcpToolCall: {
					toolName: 'Calculator',
					arguments: { input: 'bad' },
					sourceNodeName: 'Calculator',
				},
			};

			await jobProcessor.processJob(job);

			expect(run.data.resultData.runData.Calculator?.[0]?.executionStatus).toBe('error');
			expect(run.data.resultData.runData.Calculator?.[0]?.error).toBe(toolError);
		});
	});

	describe('waitTill propagation', () => {
		it('carries waitTill on JobFinishedProps from the worker run (lightweight path)', () => {
			const waitTill = new Date(Date.now() + 60_000);
			const jobProcessor = new JobProcessor(
				logger,
				mock<ExecutionRepository>(),
				mock(),
				mock(),
				mock(),
				mock(),
				mock(),
				executionsConfig,
				mock(),
				mock(),
			);
			const run = mock<IRun>({
				status: 'waiting',
				stoppedAt: new Date(),
				data: mock<IRunExecutionData>({
					resultData: { runData: {}, error: undefined },
					executionData: undefined,
				}),
			});
			// set Date field after construction, else vitest-mock-extended serializes them otherwise.
			run.waitTill = waitTill;

			const props = jobProcessor['deriveJobFinishedProps'](run, new Date());

			expect(props.waitTill).toBe(waitTill);
			expect(props.status).toBe('waiting');
		});

		it('defaults waitTill to null on JobFinishedProps when the run is not waiting', () => {
			const jobProcessor = new JobProcessor(
				logger,
				mock<ExecutionRepository>(),
				mock(),
				mock(),
				mock(),
				mock(),
				mock(),
				executionsConfig,
				mock(),
				mock(),
			);
			const run = mock<IRun>({
				status: 'success',
				stoppedAt: new Date(),
				data: mock<IRunExecutionData>({
					resultData: { runData: {}, error: undefined },
					executionData: undefined,
				}),
			});
			run.waitTill = undefined;
			expect(jobProcessor['deriveJobFinishedProps'](run, new Date()).waitTill).toBeNull();
		});
	});

	describe('project info in log metadata', () => {
		beforeEach(() => {
			vi.clearAllMocks();
		});
		it('should include project info in log metadata when present in job data', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { id: 'wf-1', name: 'Test Workflow', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			// Second call for checking errors
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mock(),
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-1',
				loadStaticData: false,
				projectId: 'proj-123',
				projectName: 'My Project',
			};

			await jobProcessor.processJob(job);

			// "Worker started" log should include project info
			expect(logger.info).toHaveBeenCalledWith(
				expect.stringContaining('Worker started execution'),
				expect.objectContaining({
					workflowId: 'wf-1',
					workflowName: 'Test Workflow',
					projectId: 'proj-123',
					projectName: 'My Project',
				}),
			);

			// "Worker finished" log should include project info
			expect(logger.info).toHaveBeenCalledWith(
				expect.stringContaining('Worker finished execution'),
				expect.objectContaining({
					workflowId: 'wf-1',
					workflowName: 'Test Workflow',
					projectId: 'proj-123',
					projectName: 'My Project',
				}),
			);
		});

		it('should not include project info in log metadata when absent from job data', async () => {
			const executionRepository = mock<ExecutionRepository>();
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { id: 'wf-1', name: 'Test Workflow', nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({
						executionData: undefined,
					}),
				}),
			);
			executionPersistence.findSingleExecution.mockResolvedValueOnce(
				mock<IExecutionResponse>({
					status: 'success',
					data: mock<IRunExecutionData>({ resultData: { runData: {} } }),
				}),
			);

			const manualExecutionService = createManualExecutionServiceMock();
			const jobProcessor = new JobProcessor(
				logger,
				executionRepository,
				executionPersistence,
				mock(),
				mock(),
				mock(),
				manualExecutionService,
				executionsConfig,
				mock(),
				mock(),
			);

			const job = mock<Job>();
			job.data = {
				workflowId: 'wf-1',
				executionId: 'exec-1',
				loadStaticData: false,
			};

			await jobProcessor.processJob(job);

			// "Worker started" log should not include project fields
			const startedCall = (logger.info as Mock).mock.calls.find(
				(call: unknown[]) =>
					typeof call[0] === 'string' && call[0].includes('Worker started execution'),
			) as [string, Record<string, unknown>] | undefined;
			expect(startedCall).toBeDefined();
			expect(startedCall![1].workflowId).toBe('wf-1');
			expect(startedCall![1]).not.toHaveProperty('projectId');
			expect(startedCall![1]).not.toHaveProperty('projectName');
		});
	});

	describe('webhook response relay', () => {
		const buildRelay = (webhookResponseRelaySizeMaxMiB: number, mode: BinaryDataConfig['mode']) => {
			const binaryDataService = mock<BinaryDataService>();
			binaryDataService.store.mockImplementation(async (_location, _body, binaryData) => ({
				...binaryData,
				id: `${mode}:stored-file-id`,
			}));

			const relay = new WebhookResponseRelay(
				logger,
				binaryDataService,
				mock<BinaryDataConfig>({ mode }),
				mock<ExecutionsConfig>({
					webhookResponseRelaySizeMaxMiB,
					webhookResponseRelayOffloadEnabled: true,
				}),
			);

			return { relay, binaryDataService };
		};

		/** Runs a job and returns the hooks it registered, so `sendResponse` can be invoked. */
		const processJobAndCaptureHooks = async (
			webhookResponseRelaySizeMaxMiB: number,
			jobData: Partial<Job['data']> = {},
			mode: BinaryDataConfig['mode'] = 'default',
		) => {
			const { relay, binaryDataService } = buildRelay(webhookResponseRelaySizeMaxMiB, mode);
			const executionPersistence = mock<ExecutionPersistence>();
			executionPersistence.findSingleExecution.mockResolvedValue(
				mock<IExecutionResponse>({
					mode: 'manual',
					workflowData: { nodes: [], staticData: {} },
					data: mock<IRunExecutionData>({ executionData: undefined }),
				}),
			);

			const additionalData = mock<IWorkflowExecuteAdditionalData>();
			vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);

			const jobProcessor = new JobProcessor(
				logger,
				mock<ExecutionRepository>(),
				executionPersistence,
				mock(),
				mock(),
				mock(),
				createManualExecutionServiceMock(),
				executionsConfig,
				mock(),
				relay,
			);

			const job = mock<Job>({
				data: {
					workflowId: 'wf-1',
					executionId: 'exec-1',
					loadStaticData: false,
					isMcpExecution: undefined,
					mcpSessionId: undefined,
					...jobData,
				},
			});
			await jobProcessor.processJob(job);

			return { hooks: additionalData.hooks!, job, binaryDataService };
		};

		it('should relay a response within the size limit, encoding a buffer body', async () => {
			const { hooks, job } = await processJobAndCaptureHooks(1);

			await hooks.runHook('sendResponse', [
				{ body: Buffer.from('hello'), headers: {}, statusCode: 200 },
			]);

			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'respond-to-webhook',
					executionId: 'exec-1',
					response: expect.objectContaining({ body: { [ENCODED_BUFFER_KEY]: 'aGVsbG8=' } }),
				}),
			);
		});

		it('should offload a response over the size limit and relay a reference', async () => {
			const { hooks, job, binaryDataService } = await processJobAndCaptureHooks(1, {}, 'database');

			await hooks.runHook('sendResponse', [
				{ body: { blob: 'x'.repeat(2 * 1024 * 1024) }, headers: {}, statusCode: 200 },
			]);

			expect(binaryDataService.store).toHaveBeenCalledTimes(1);
			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'respond-to-webhook',
					response: expect.objectContaining({
						body: expect.objectContaining({
							binaryData: expect.objectContaining({ id: 'database:stored-file-id' }),
						}),
					}),
				}),
			);
		});

		it('should refuse to relay a response over the size limit without a store', async () => {
			const { hooks, job } = await processJobAndCaptureHooks(1);
			const relayedBefore = (job.progress as Mock).mock.calls.length;

			await expect(
				hooks.runHook('sendResponse', [
					{ body: { blob: 'x'.repeat(2 * 1024 * 1024) }, headers: {}, statusCode: 200 },
				]),
			).rejects.toThrow(WebhookResponseTooLargeError);

			expect((job.progress as Mock).mock.calls).toHaveLength(relayedBefore);
		});

		it('should offload an MCP response over the size limit and relay a reference', async () => {
			const { hooks, job, binaryDataService } = await processJobAndCaptureHooks(
				1,
				{ isMcpExecution: true, mcpSessionId: 'session-1', mcpType: 'trigger' },
				'database',
			);

			await hooks.runHook('sendResponse', [
				{ body: { blob: 'x'.repeat(2 * 1024 * 1024) }, headers: {}, statusCode: 200 },
			]);

			expect(binaryDataService.store).toHaveBeenCalledTimes(1);
			expect(job.progress).toHaveBeenCalledWith(
				expect.objectContaining({
					kind: 'mcp-response',
					sessionId: 'session-1',
					response: expect.objectContaining({
						body: expect.objectContaining({
							binaryData: expect.objectContaining({ id: 'database:stored-file-id' }),
						}),
					}),
				}),
			);
		});

		it('should not relay an MCP Service response, which main reads from stored data', async () => {
			const { hooks, job } = await processJobAndCaptureHooks(1, {
				isMcpExecution: true,
				mcpSessionId: 'session-1',
				mcpType: 'service',
			});
			const relayedBefore = (job.progress as Mock).mock.calls.length;

			await hooks.runHook('sendResponse', [
				{ body: Buffer.from('hello'), headers: {}, statusCode: 200 },
			]);

			expect((job.progress as Mock).mock.calls).toHaveLength(relayedBefore);
		});

		it('should not store an MCP Service response over the size limit', async () => {
			const { hooks, job, binaryDataService } = await processJobAndCaptureHooks(
				1,
				{ isMcpExecution: true, mcpSessionId: 'session-1', mcpType: 'service' },
				'database',
			);
			const relayedBefore = (job.progress as Mock).mock.calls.length;

			await hooks.runHook('sendResponse', [
				{ body: { blob: 'x'.repeat(2 * 1024 * 1024) }, headers: {}, statusCode: 200 },
			]);

			expect(binaryDataService.store).not.toHaveBeenCalled();
			expect((job.progress as Mock).mock.calls).toHaveLength(relayedBefore);
		});

		it('should refuse to relay an oversized MCP payload with no body to offload', async () => {
			const { hooks, job, binaryDataService } = await processJobAndCaptureHooks(
				1,
				{ isMcpExecution: true, mcpSessionId: 'session-1', mcpType: 'trigger' },
				'database',
			);
			const relayedBefore = (job.progress as Mock).mock.calls.length;

			await expect(
				hooks.runHook('sendResponse', [{ toolResult: 'x'.repeat(2 * 1024 * 1024) }]),
			).rejects.toThrow(WebhookResponseTooLargeError);

			expect((job.progress as Mock).mock.calls).toHaveLength(relayedBefore);
			expect(binaryDataService.store).not.toHaveBeenCalled();
		});
	});
});