import type { Tool } from '@langchain/core/tools';
import type { RunningJobSummary } from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { EventService } from '@n8n/backend-services';
import { ExecutionsConfig } from '@n8n/config';
import { MAX_INTEGER_32BITS_SIGNED } from '@n8n/constants';
import type { IExecutionResponse } from '@n8n/db';
import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import {
	WorkflowHasIssuesError,
	InstanceSettings,
	WorkflowExecute,
	SupplyDataContext,
	StructuredToolkit,
} from 'n8n-core';
import {
	ManualExecutionCancelledError,
	NodeConnectionTypes,
	NodeOperationError,
	TimeoutExecutionCancelledError,
	Workflow,
	UnexpectedError,
	createRunExecutionData,
	CHAT_TRIGGER_NODE_TYPE,
	runDataAttemptedDynamicCredentials,
	runDataUsedDynamicCredentials,
} from 'n8n-workflow';
import type {
	CancellationReason,
	ExecutionStatus,
	WorkflowExecuteMode,
	IDataObject,
	IExecuteData,
	IExecuteFunctions,
	IExecutionContext,
	INodeExecutionData,
	IRun,
	IRunExecutionData,
	IWorkflowExecutionDataProcess,
	StructuredChunk,
	CloseFunction,
	GenericValue,
} from 'n8n-workflow';
import type PCancelable from 'p-cancelable';

import { getLifecycleHooksForScalingWorker } from '@/execution-lifecycle/execution-lifecycle-hooks';
import { prepareExecutionDataForDbUpdate } from '@/execution-lifecycle/shared/shared-hook-functions';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { getWorkflowActiveStatusFromWorkflowData } from '@/executions/execution.utils';
import { ManualExecutionService } from '@/manual-execution.service';
import { NodeTypes } from '@/node-types';
import { withExpressionIsolate } from '@/utils';
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';

import type {
	Job,
	JobFinishedMessage,
	JobFinishedProps,
	JobId,
	JobResult,
	RespondToWebhookMessage,
	McpResponseMessage,
	RunningJob,
	SendChunkMessage,
} from './scaling.types';
import { WebhookResponseRelay } from './webhook-response-relay';

function isInvokableTool(value: unknown): value is Pick<Tool, 'invoke'> {
	return (
		typeof value === 'object' &&
		value !== null &&
		'invoke' in value &&
		typeof value.invoke === 'function'
	);
}

/**
 * Runs `fn` at `timestamp`, returning a function to cancel it. `setTimeout` truncates delays past
 * its 32-bit signed limit (~24.8 days) and fires almost immediately, so longer waits are split
 * into bounded chunks.
 */
function scheduleAt(timestamp: number, fn: () => void): () => void {
	let timer: NodeJS.Timeout;

	const schedule = () => {
		const delay = Math.max(timestamp - Date.now(), 0);
		timer =
			delay > MAX_INTEGER_32BITS_SIGNED
				? setTimeout(schedule, MAX_INTEGER_32BITS_SIGNED)
				: setTimeout(fn, delay);
		timer.unref(); // A pending tick must not hold the process open on its own.
	};
	schedule();

	return () => clearTimeout(timer);
}

/** Production runs that may be parked at worker shutdown and resumed elsewhere. */
const SUSPENDABLE_EXECUTION_MODES = new Set<WorkflowExecuteMode>(['webhook', 'trigger', 'retry']);

/**
 * Responsible for processing jobs from the queue, i.e. running enqueued executions.
 */
@Service()
export class JobProcessor {
	private readonly runningJobs: Record<JobId, RunningJob> = {};

	/** Cause of the cancellation of each job cancelled so far, kept until its run settles. */
	private readonly cancellationReasons: Record<JobId, CancellationReason> = {};

	/** Set once at shutdown; jobs that register afterwards suspend right away. */
	private suspensionRequested = false;

	constructor(
		private readonly logger: Logger,
		private readonly executionRepository: ExecutionRepository,
		private readonly executionPersistence: ExecutionPersistence,
		private readonly workflowRepository: WorkflowRepository,
		private readonly nodeTypes: NodeTypes,
		private readonly instanceSettings: InstanceSettings,
		private readonly manualExecutionService: ManualExecutionService,
		private readonly executionsConfig: ExecutionsConfig,
		private readonly eventService: EventService,
		private readonly webhookResponseRelay: WebhookResponseRelay,
	) {
		this.logger = this.logger.scoped('scaling');
	}

	async processJob(job: Job): Promise<JobResult> {
		const { executionId, loadStaticData } = job.data;

		const execution = await this.executionPersistence.findSingleExecution(executionId, {
			includeData: true,
			unflattenData: true,
		});

		if (!execution) {
			throw new UnexpectedError(
				`Worker failed to find data for execution ${executionId} (job ${job.id})`,
			);
		}

		/**
		 * Bull's implicit retry mechanism and n8n's execution recovery mechanism may
		 * cause a crashed execution to be enqueued. We refrain from processing it,
		 * until we have reworked both mechanisms to prevent this scenario.
		 */
		if (execution.status === 'crashed') return { success: false };

		// A correctly enqueued execution always carries a run-data payload. A missing
		// one means the producer persisted no data, which would otherwise surface as an
		// opaque `Cannot read properties of undefined` deref further down. Fail with a
		// clear, attributable error instead.
		if (!execution.data) {
			throw new UnexpectedError(
				`Worker received execution ${executionId} without run data (job ${job.id})`,
			);
		}

		const workflowId = execution.workflowData.id;

		this.logger.info(`Worker started execution ${executionId} (job ${job.id})`, {
			executionId,
			workflowId,
			workflowName: execution.workflowData.name,
			jobId: job.id,
			...(job.data.projectId !== undefined && { projectId: job.data.projectId }),
			...(job.data.projectName !== undefined && { projectName: job.data.projectName }),
		});

		const startedAt = await this.executionRepository.setRunning(executionId);

		let { staticData } = execution.workflowData;

		if (loadStaticData) {
			const workflowData = await this.workflowRepository.findOne({
				select: ['id', 'staticData'],
				where: { id: workflowId },
			});

			if (workflowData === null) {
				throw new UnexpectedError(
					`Worker failed to find workflow ${workflowId} to run execution ${executionId} (job ${job.id})`,
				);
			}

			staticData = workflowData.staticData;
		}

		const workflowSettings = execution.workflowData.settings ?? {};

		let workflowTimeout = workflowSettings.executionTimeout ?? this.executionsConfig.timeout;

		let executionTimeoutTimestamp: number | undefined;

		if (workflowTimeout > 0) {
			workflowTimeout = Math.min(workflowTimeout, this.executionsConfig.maxTimeout);
			executionTimeoutTimestamp = Date.now() + workflowTimeout * 1000;
		}

		const workflow = new Workflow({
			id: workflowId,
			name: execution.workflowData.name,
			nodes: execution.workflowData.nodes,
			connections: execution.workflowData.connections,
			active: getWorkflowActiveStatusFromWorkflowData(execution.workflowData),
			nodeTypes: this.nodeTypes,
			staticData,
			settings: execution.workflowData.settings,
		});

		const additionalData = await WorkflowExecuteAdditionalData.getBase({
			workflowId,
			executionTimeoutTimestamp,
			workflowSettings: execution.workflowData.settings,
		});
		additionalData.streamingEnabled = job.data.streamingEnabled;
		additionalData.restartExecutionId = job.data.restartExecutionId;
		additionalData.evaluationRunId = execution.data.manualData?.evaluationRunId;
		// Rehydrate the manual-execution identity for private credential resolution.
		additionalData.encryptedRunnerIdentity = job.data.encryptedRunnerIdentity;

		const { pushRef } = job.data;

		const lifecycleHooks = getLifecycleHooksForScalingWorker(
			{
				executionMode: execution.mode,
				workflowData: execution.workflowData,
				retryOf: execution.retryOf,
				pushRef,
				userId: execution.data.manualData?.userId,
				source: execution.data.manualData?.source,
				suppressErrorWorkflow: execution.data.manualData?.suppressErrorWorkflow,
			},
			executionId,
		);
		additionalData.hooks = lifecycleHooks;

		if (pushRef) {
			additionalData.sendDataToUI = WorkflowExecuteAdditionalData.sendDataToUI.bind({
				pushRef,
			});
		}

		// An older main sends no value. Assume the strictest, so nothing is parked
		// until every main is upgraded.
		let callerAwaitsOutcome = job.data.callerAwaitsOutcome ?? 'completion';

		lifecycleHooks.addHandler('sendResponse', async (response): Promise<void> => {
			// An MCP Service call takes its result from the execution's stored data, so a
			// response relayed to main has no reader. Relaying one would also reach main
			// mid-execution, resolving the call on data the run has not finished writing.
			if (job.data.isMcpExecution && job.data.mcpSessionId && job.data.mcpType !== 'trigger') {
				return;
			}

			const relayed = await this.webhookResponseRelay.prepare(response, {
				workflowId: job.data.workflowId,
				executionId,
			});

			// Check if this is an MCP execution - broadcast response to all mains
			if (job.data.isMcpExecution && job.data.mcpSessionId) {
				const msg: McpResponseMessage = {
					kind: 'mcp-response',
					executionId,
					mcpType: 'trigger',
					sessionId: job.data.mcpSessionId,
					messageId: job.data.mcpMessageId ?? '',
					response: relayed,
					workerId: this.instanceSettings.hostId,
				};

				await job.progress(msg);
				return;
			}

			// Standard webhook response
			const msg: RespondToWebhookMessage = {
				kind: 'respond-to-webhook',
				executionId,
				response: relayed,
				workerId: this.instanceSettings.hostId,
			};

			await job.progress(msg);

			// Only a webhook caller is satisfied by a relayed response; a done promise waits for the end.
			if (callerAwaitsOutcome !== 'response') return;
			callerAwaitsOutcome = 'none';
			if (this.suspensionRequested) this.runningJobs[job.id]?.suspend?.();
		});

		lifecycleHooks.addHandler('sendChunk', async (chunk: StructuredChunk): Promise<void> => {
			const msg: SendChunkMessage = {
				kind: 'send-chunk',
				executionId,
				chunkText: chunk,
				workerId: this.instanceSettings.hostId,
			};

			await job.progress(msg);
		});

		additionalData.executionId = executionId;

		additionalData.setExecutionStatus = (status: ExecutionStatus) => {
			// Can't set the status directly in the queued worker, but it will happen in InternalHook.onWorkflowPostExecute
			this.logger.debug(
				`Queued worker execution status for execution ${executionId} (job ${job.id}) is "${status}"`,
				{
					executionId,
					workflowId,
					workflowName: execution.workflowData.name,
					jobId: job.id,
					...(job.data.projectId && { projectId: job.data.projectId }),
					...(job.data.projectName && { projectName: job.data.projectName }),
				},
			);
		};

		let workflowExecute: WorkflowExecute | undefined;
		let workflowRun: PCancelable<IRun>;

		const { startData, resultData, manualData } = execution.data;

		if (execution.data?.executionData) {
			workflowExecute = new WorkflowExecute(additionalData, execution.mode, execution.data);
			workflowRun = workflowExecute.processRunExecutionData(workflow);
		} else {
			const data: IWorkflowExecutionDataProcess = {
				executionMode: execution.mode,
				workflowData: execution.workflowData,
				destinationNode: startData?.destinationNode,
				startNodes: startData?.startNodes,
				runData: resultData.runData,
				pinData: resultData.pinData,
				dirtyNodeNames: manualData?.dirtyNodeNames,
				triggerToStartFrom: manualData?.triggerToStartFrom,
				userId: manualData?.userId,
				agentRequest: manualData?.agentRequest,
			};

			try {
				workflowRun = this.manualExecutionService.runManually(
					data,
					workflow,
					additionalData,
					executionId,
					resultData.pinData,
				);
			} catch (error) {
				if (error instanceof WorkflowHasIssuesError) {
					// execution did not even start, but we call `workflowExecuteAfter` to notify main

					const now = new Date();
					const runData: IRun = {
						mode: 'manual',
						status: 'error',
						finished: false,
						startedAt: now,
						stoppedAt: now,
						data: createRunExecutionData({ resultData: { error, runData: {} } }),
						storedAt: execution.storedAt,
					};

					await lifecycleHooks.runHook('workflowExecuteAfter', [runData]);
					return { success: false };
				}
				throw error;
			}
		}

		const runningJob: RunningJob = {
			run: workflowRun,
			executionId,
			workflowId: execution.workflowId,
			workflowName: execution.workflowData.name,
			mode: execution.mode,
			startedAt,
			retryOf: execution.retryOf ?? undefined,
			status: execution.status,
		};

		if (workflowExecute && this.isJobSuspendable(job, execution)) {
			const suspendable = workflowExecute;
			runningJob.suspend = () => {
				if (callerAwaitsOutcome !== 'none') return false;
				suspendable.suspend();
				return true;
			};
			// A job that was still in its preflight reads when shutdown asked the
			// running jobs to suspend would otherwise run to completion unasked.
			if (this.suspensionRequested) runningJob.suspend();
		}

		this.runningJobs[job.id] = runningJob;

		// The engine only checks `executionTimeoutTimestamp` between node executions, so it cannot
		// interrupt a node stuck mid-execution (e.g. a hanging HTTP call). This watchdog cancels
		// the job for abort-aware operations, mirroring the regular-process timeout in
		// `WorkflowRunner.runMainProcess`.
		const clearTimeoutWatchdog =
			executionTimeoutTimestamp !== undefined
				? scheduleAt(executionTimeoutTimestamp, () => this.cancelJob(job.id, 'timeout'))
				: undefined;

		let run: IRun;
		let cancellationReason: CancellationReason | undefined;
		try {
			run = await workflowRun;
		} finally {
			// A pending watchdog would cancel the job belatedly.
			clearTimeoutWatchdog?.();
			cancellationReason = this.cancellationReasons[job.id];
			delete this.cancellationReasons[job.id];
			// An entry left behind on rejection keeps the count of running jobs
			// above zero forever, which prevents shutdown from ever completing.
			delete this.runningJobs[job.id];
		}

		// A cancel this worker performed names its own cause. The engine cancels itself when its
		// between-node check finds the deadline passed, and records no reason, so read the
		// deadline the same way the engine does.
		const timedOut =
			cancellationReason === undefined
				? executionTimeoutTimestamp !== undefined && Date.now() >= executionTimeoutTimestamp
				: cancellationReason === 'timeout';

		// A cancelled job is already reported as cancelled through `execution-cancelled`, even
		// when the run itself ignored the cancel and ran to completion.
		if (run?.status === 'canceled' || cancellationReason !== undefined) {
			throw timedOut
				? new TimeoutExecutionCancelledError(executionId)
				: new ManualExecutionCancelledError(executionId);
		}

		const props = this.deriveJobFinishedProps(run, startedAt);

		this.logger.info(`Worker finished execution ${executionId} (job ${job.id})`, {
			executionId,
			workflowId,
			workflowName: execution.workflowData.name,
			jobId: job.id,
			success: props.success,
			...(job.data.projectId && { projectId: job.data.projectId }),
			...(job.data.projectName && { projectName: job.data.projectName }),
		});

		const msg: JobFinishedMessage = {
			kind: 'job-finished',
			version: 2,
			executionId,
			workerId: this.instanceSettings.hostId,
			...props,
		};

		await job.progress(msg);

		// For MCP Trigger executions with tool calls, execute the tool and send result
		if (
			job.data.isMcpExecution &&
			job.data.mcpType === 'trigger' &&
			job.data.mcpSessionId &&
			job.data.mcpToolCall?.sourceNodeName
		) {
			const { toolName, arguments: toolArgs, sourceNodeName } = job.data.mcpToolCall;

			let toolResult: unknown;
			try {
				// The execution's isolate window closed when the run finished, but the
				// tool's parameters may still contain expressions (e.g. $fromAI), so
				// the tool call needs its own isolate window.
				toolResult = await withExpressionIsolate(
					workflow,
					async () =>
						await this.invokeTool({
							workflow,
							sourceNodeName,
							toolName,
							toolArgs,
							toolInput: job.data.mcpToolInput,
							additionalData,
							runExecutionData: run.data,
							// The execution context (e.g. the OAuth identity for private credentials)
							// is established on the main and loaded with the execution here; pass it
							// through so the tool node can resolve dynamic credentials on the worker.
							executionContext: execution.data?.executionData?.runtimeData,
						}),
				);

				// A tool result is not a response, so it has no offload path: this limit
				// is all that keeps it from travelling through the queue unbounded.
				this.webhookResponseRelay.assertFitsInline(toolResult);
			} catch (error) {
				this.logger.error('Tool call failed for MCP Trigger', {
					executionId,
					toolName,
					sourceNodeName,
					error: error instanceof Error ? error.message : String(error),
				});
				toolResult = {
					error:
						error instanceof Error
							? { message: error.message, name: error.name }
							: { message: String(error) },
				};
			}

			// Persist the tool call's run data, since the save hook fired before it ran.
			try {
				const toolRunData = run.data.resultData?.runData;
				await this.executionPersistence.updateExistingExecution(
					executionId,
					{
						...prepareExecutionDataForDbUpdate({
							runData: run,
							workflowData: execution.workflowData,
							workflowStatusFinal: run.status,
							retryOf: execution.retryOf ?? undefined,
						}),
						// The save hook computed this marker before the tool ran, so recompute it now
						// that the tool task may carry dynamic-credential flags.
						usedPrivateCredentials:
							runDataUsedDynamicCredentials(toolRunData) ||
							runDataAttemptedDynamicCredentials(toolRunData),
					},
					// A cancel racing the tool call must keep its status; skip the tool-run persist
					// entirely rather than write over `canceled` (matches the completion hook).
					{ requireNotCanceled: true },
				);
			} catch (error) {
				this.logger.error('Failed to persist tool call run data for MCP Trigger', {
					executionId,
					sourceNodeName,
					error: error instanceof Error ? error.message : String(error),
				});
			}

			const mcpMsg: McpResponseMessage = {
				kind: 'mcp-response',
				executionId,
				mcpType: 'trigger',
				sessionId: job.data.mcpSessionId,
				messageId: job.data.mcpMessageId ?? '',
				response: toolResult, // Actual tool result
				workerId: this.instanceSettings.hostId,
			};

			await job.progress(mcpMsg);
		} else if (job.data.isMcpExecution && job.data.mcpSessionId) {
			// For MCP Service executions or MCP Trigger without tool call, send basic response
			const mcpMsg: McpResponseMessage = {
				kind: 'mcp-response',
				executionId,
				mcpType: job.data.mcpType ?? 'service',
				sessionId: job.data.mcpSessionId,
				messageId: job.data.mcpMessageId ?? '',
				response: { success: props.success },
				workerId: this.instanceSettings.hostId,
			};

			await job.progress(mcpMsg);
		}

		/**
		 * @important Do NOT call `workflowExecuteAfter` hook here.
		 * It is being called from processSuccessExecution() already.
		 */

		return { success: true };
	}

	private deriveJobFinishedProps(run: IRun, startedAt: Date): JobFinishedProps {
		return {
			success: run.status !== 'error' && run.data.resultData.error === undefined,
			status: run.status,
			error: run.data.resultData.error,
			startedAt,
			stoppedAt: run.stoppedAt!,
			lastNodeExecuted: run.data.resultData.lastNodeExecuted,
			usedDynamicCredentials: !!run.data.executionData?.runtimeData?.credentials,
			metadata: run.data.resultData.metadata,
			waitTill: run.waitTill ?? null,
		};
	}

	stopJob(jobId: JobId) {
		this.cancelJob(jobId, 'manual'); // Job stops via scaling service are always user-initiated
	}

	private cancelJob(jobId: JobId, reason: CancellationReason) {
		const runningJob = this.runningJobs[jobId];
		if (!runningJob) return;

		const { executionId, workflowId, workflowName } = runningJob;
		this.eventService.emit('execution-cancelled', {
			executionId,
			workflowId,
			workflowName,
			reason,
		});

		runningJob.run.cancel();
		delete this.runningJobs[jobId];
		this.cancellationReasons[jobId] = reason;
	}

	getRunningJobIds(): JobId[] {
		return Object.keys(this.runningJobs);
	}

	/**
	 * Whether a job may be suspended at worker shutdown. Only production
	 * executions qualify: manual and evaluation runs are tied to a session,
	 * streaming responses cannot migrate mid-stream, and MCP executions are
	 * pinned to their session, and so is a chat conversation, whose client polls
	 * the waiting state and would mistake a parked run for a reply. A pending
	 * webhook response is checked at suspend time, since the run may still send it.
	 */
	private isJobSuspendable(job: Job, execution: IExecutionResponse): boolean {
		const hasChatTrigger = execution.workflowData.nodes.some(
			(node) => node.type === CHAT_TRIGGER_NODE_TYPE && !node.disabled,
		);

		return (
			SUSPENDABLE_EXECUTION_MODES.has(execution.mode) &&
			!job.data.streamingEnabled &&
			!job.data.isMcpExecution &&
			!hasChatTrigger &&
			execution.data.executionData !== undefined
		);
	}

	/** Ask every suspendable running job to stop at its next node boundary. */
	suspendRunningJobs() {
		this.suspensionRequested = true;
		const executionIds: string[] = [];

		for (const runningJob of Object.values(this.runningJobs)) {
			if (!runningJob.suspend?.()) continue;
			executionIds.push(runningJob.executionId);
		}

		if (executionIds.length > 0) {
			this.logger.info(
				`Requested suspension of ${executionIds.length} execution(s) at the next node boundary (execution IDs: ${executionIds.join(', ')})`,
			);
		}
	}

	getRunningJobsSummary(): RunningJobSummary[] {
		return Object.values(this.runningJobs).map(({ run, ...summary }) => summary);
	}

	/**
	 * Invoke a tool directly for MCP Trigger in queue mode.
	 * For nodes with supplyData (e.g. native langchain tool nodes), creates a
	 * SupplyDataContext, calls supplyData to get the Tool, and invokes it.
	 * For tool wrapper nodes without supplyData (e.g. httpRequestTool), calls
	 * execute directly — mirroring the fallback in get-input-connection-data.ts.
	 */
	private async invokeTool({
		workflow,
		sourceNodeName,
		toolName,
		toolArgs,
		toolInput,
		additionalData,
		runExecutionData,
		executionContext,
	}: {
		workflow: Workflow;
		sourceNodeName: string;
		toolName: string;
		toolArgs: Record<string, unknown>;
		toolInput?: IDataObject;
		additionalData: Awaited<ReturnType<typeof WorkflowExecuteAdditionalData.getBase>>;
		runExecutionData: IRunExecutionData;
		executionContext?: IExecutionContext;
	}): Promise<unknown> {
		const toolNode = workflow.getNode(sourceNodeName);
		if (!toolNode) {
			throw new UnexpectedError(`Tool node "${sourceNodeName}" not found in workflow`);
		}

		// Get the node type
		const nodeType = this.nodeTypes.getByNameAndVersion(toolNode.type, toolNode.typeVersion);

		// Validate toolArgs is a proper object (not null/array) before using as input data
		const validatedToolArgs =
			typeof toolArgs === 'object' && toolArgs !== null && !Array.isArray(toolArgs) ? toolArgs : {};

		// A tool can feed several MCP triggers, and connection order says nothing about which
		// one this execution came from — only the trigger that ran has run data. Recording it
		// as parent points the tool's run data `source` at it, as in direct mode.
		const { runData } = runExecutionData.resultData;
		const triggerNames = workflow.getChildNodes(sourceNodeName, NodeConnectionTypes.AiTool, 1);
		const triggerName = triggerNames.find((name) => runData[name]) ?? triggerNames[0];
		const parentNode = triggerName ? (workflow.getNode(triggerName) ?? undefined) : undefined;

		const inputData: INodeExecutionData[][] = [[{ json: toolInput ?? {} }]];

		// `executionData` must exist for output recording; init it if the run lacks it.
		runExecutionData.executionData ??= {
			contextData: {},
			nodeExecutionStack: [],
			metadata: {},
			waitingExecution: {},
			waitingExecutionSource: {},
		};
		if (executionContext) {
			runExecutionData.executionData.runtimeData = executionContext;
		}

		// Create execute data for the tool node
		const executeData: IExecuteData = {
			node: toolNode,
			data: {
				main: inputData,
			},
			source: null,
		};

		const closeFunctions: CloseFunction[] = [];

		// Create SupplyDataContext for the tool node
		const context = new SupplyDataContext(
			workflow,
			toolNode,
			additionalData,
			'webhook',
			runExecutionData,
			0,
			inputData[0],
			{ main: inputData },
			NodeConnectionTypes.AiTool,
			executeData,
			closeFunctions,
			undefined,
			parentNode,
		);

		try {
			if (nodeType.supplyData) {
				const supplyDataResult = await nodeType.supplyData.call(context, 0);
				if (supplyDataResult.closeFunction) {
					closeFunctions.push(supplyDataResult.closeFunction);
				}

				let tool = supplyDataResult.response;
				if (tool instanceof StructuredToolkit) {
					tool = tool.tools.find((member) => member.name === toolName);
					if (!tool) {
						throw new UnexpectedError(
							`Tool "${toolName}" not found in toolkit from node "${sourceNodeName}"`,
						);
					}
				}

				if (!isInvokableTool(tool)) {
					throw new UnexpectedError(`Tool node "${sourceNodeName}" did not return a valid Tool`);
				}

				return await tool.invoke(validatedToolArgs);
			}

			if (nodeType.execute && nodeType.description.outputs.includes(NodeConnectionTypes.AiTool)) {
				// oxlint-disable-next-line typescript/no-deprecated
				context.addInputData(NodeConnectionTypes.AiTool, [
					[{ json: validatedToolArgs as INodeExecutionData['json'] }],
				]);

				let result: Awaited<ReturnType<NonNullable<typeof nodeType.execute>>>;
				try {
					result = await nodeType.execute.call(context as unknown as IExecuteFunctions);
				} catch (error) {
					// Record the failure so the tool node shows as errored, not stuck
					// "running"; rethrow so the caller returns an error to the client.
					// oxlint-disable-next-line typescript/no-deprecated
					context.addOutputData(
						NodeConnectionTypes.AiTool,
						0,
						error instanceof NodeOperationError
							? error
							: new NodeOperationError(toolNode, error as Error),
					);
					throw error;
				}

				let response: IDataObject | IDataObject[] | GenericValue | GenericValue[] = [];
				if (Array.isArray(result)) {
					response = result?.[0]?.flatMap((item: INodeExecutionData) => item.json);
				}

				// oxlint-disable-next-line typescript/no-deprecated
				context.addOutputData(NodeConnectionTypes.AiTool, 0, [[{ json: { response } }]]);

				return response;
			}

			throw new UnexpectedError(
				`Tool node "${sourceNodeName}" does not have supplyData or execute method`,
			);
		} finally {
			for (const closeFunction of closeFunctions) {
				try {
					await closeFunction();
				} catch (error) {
					this.logger.warn(`Error closing tool resource: ${error}`);
				}
			}
		}
	}
}