fix(core): Count AI assistant workflow test runs as manual executions in statistics (backport to release-candidate/2.31.x) (#34152)

Co-authored-by: oleg <me@olegivaniv.com>

packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts

@@ -139,6 +139,7 @@ type HooksSetupParameters = {
 	pushRef?: string;
 	retryOf?: string;
 	parentExecution?: RelatedExecution;
+	source?: IWorkflowExecutionDataProcess['source'];
 };
 
 function hookFunctionsWorkflowEvents(
@@ -485,10 +486,13 @@ function hookFunctionsFinalizeExecutionStatus(hooks: ExecutionLifecycleHooks) {
 	});
 }
 
-function hookFunctionsStatistics(hooks: ExecutionLifecycleHooks) {
+function hookFunctionsStatistics(
+	hooks: ExecutionLifecycleHooks,
+	source?: IWorkflowExecutionDataProcess['source'],
+) {
 	const workflowStatisticsService = Container.get(WorkflowStatisticsService);
 	hooks.addHandler('nodeFetchedData', (workflowId, node) => {
-		workflowStatisticsService.emit('nodeFetchedData', { workflowId, node });
+		workflowStatisticsService.emit('nodeFetchedData', { workflowId, node, source });
 	});
 }
 
@@ -519,7 +523,7 @@ async function duplicateBinaryDataToParent(
  */
 function hookFunctionsSave(
 	hooks: ExecutionLifecycleHooks,
-	{ pushRef, retryOf, saveSettings, parentExecution }: HooksSetupParameters,
+	{ pushRef, retryOf, saveSettings, parentExecution, source }: HooksSetupParameters,
 ) {
 	const logger = Container.get(Logger);
 	const errorReporter = Container.get(ErrorReporter);
@@ -633,6 +637,7 @@ function hookFunctionsSave(
 			workflowStatisticsService.emit('workflowExecutionCompleted', {
 				workflowData: this.workflowData,
 				fullRunData,
+				source,
 			});
 		}
 	});
@@ -647,7 +652,7 @@ const DISCARDABLE_DATA_MODES: WorkflowExecuteMode[] = ['trigger', 'cli', 'error'
  */
 function hookFunctionsSaveWorker(
 	hooks: ExecutionLifecycleHooks,
-	{ pushRef, retryOf, saveSettings }: HooksSetupParameters,
+	{ pushRef, retryOf, saveSettings, source }: HooksSetupParameters,
 ) {
 	const logger = Container.get(Logger);
 	const errorReporter = Container.get(ErrorReporter);
@@ -722,6 +727,7 @@ function hookFunctionsSaveWorker(
 			workflowStatisticsService.emit('workflowExecutionCompleted', {
 				workflowData: this.workflowData,
 				fullRunData,
+				source,
 			});
 		}
 	});
@@ -760,7 +766,7 @@ export function getLifecycleHooksForScalingWorker(
 	data: IWorkflowExecutionDataProcess,
 	executionId: string,
 ): ExecutionLifecycleHooks {
-	const { pushRef, retryOf, executionMode, workflowData } = data;
+	const { pushRef, retryOf, executionMode, workflowData, source } = data;
 	const hooks = new ExecutionLifecycleHooks(
 		executionMode,
 		executionId,
@@ -768,12 +774,12 @@ export function getLifecycleHooksForScalingWorker(
 		retryOf ?? undefined,
 	);
 	const saveSettings = toSaveSettings(workflowData.settings);
-	const optionalParameters = { pushRef, retryOf: retryOf ?? undefined, saveSettings };
+	const optionalParameters = { pushRef, retryOf: retryOf ?? undefined, saveSettings, source };
 	hookFunctionsNodeEvents(hooks);
 	hookFunctionsFinalizeExecutionStatus(hooks);
 	hookFunctionsSaveWorker(hooks, optionalParameters);
 	hookFunctionsSaveProgress(hooks, optionalParameters);
-	hookFunctionsStatistics(hooks);
+	hookFunctionsStatistics(hooks, source);
 	hookFunctionsExternalHooks(hooks);
 
 	if (executionMode === 'manual' && Container.get(InstanceSettings).isWorker) {
@@ -898,14 +904,14 @@ export function getLifecycleHooksForRegularMain(
 		retryOf ?? undefined,
 	);
 	const saveSettings = toSaveSettings(workflowData.settings);
-	const optionalParameters = { pushRef, retryOf: retryOf ?? undefined, saveSettings };
+	const optionalParameters = { pushRef, retryOf: retryOf ?? undefined, saveSettings, source };
 	hookFunctionsWorkflowEvents(hooks, userId, projectId, projectName, source, telemetryMetadata);
 	hookFunctionsNodeEvents(hooks);
 	hookFunctionsFinalizeExecutionStatus(hooks);
 	hookFunctionsSave(hooks, optionalParameters);
 	hookFunctionsPush(hooks, optionalParameters, userId, source);
 	hookFunctionsSaveProgress(hooks, optionalParameters);
-	hookFunctionsStatistics(hooks);
+	hookFunctionsStatistics(hooks, source);
 	hookFunctionsExternalHooks(hooks);
 	Container.get(ModulesHooksRegistry).addHooks(hooks);
 	return hooks;

packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts

@@ -1121,6 +1121,17 @@ export class InstanceAiAdapterService {
 					mockDataSources: pinDataPlan.mockDataSources,
 				};
 
+				// In queue mode the worker rebuilds the run from persisted `execution.data`,
+				// where top-level `runData.source` doesn't survive. Persist it in
+				// `manualData` so worker-side statistics can classify IAI runs.
+				if (runData.executionData) {
+					runData.executionData.manualData = {
+						...runData.executionData.manualData,
+						userId: user.id,
+						source: runData.source,
+					};
+				}
+
 				// When manual executions are offloaded to workers (queue mode), the worker
 				// rebuilds the run from the persisted `execution.data`. The adapter's manual
 				// run details otherwise live in transient top-level fields that don't survive
@@ -1141,6 +1152,7 @@ export class InstanceAiAdapterService {
 						manualData: {
 							userId: runData.userId,
 							triggerToStartFrom: runData.triggerToStartFrom,
+							source: runData.source,
 						},
 						executionData: null,
 					});

packages/cli/src/scaling/job-processor.ts

@@ -183,6 +183,7 @@ export class JobProcessor {
 				retryOf: execution.retryOf,
 				pushRef,
 				userId: execution.data.manualData?.userId,
+				source: execution.data.manualData?.source,
 			},
 			executionId,
 		);

packages/cli/src/services/__tests__/workflow-statistics.service.integration.test.ts

@@ -24,6 +24,7 @@ import {
 	type INode,
 	type IRun,
 	type WorkflowExecuteMode,
+	type WorkflowExecutionSource,
 } from 'n8n-workflow';
 
 import { EventService } from '@/events/event.service';
@@ -81,8 +82,9 @@ describe('WorkflowStatisticsService', () => {
 			service: WorkflowStatisticsService,
 			workflowData: IWorkflowDb & WorkflowEntity,
 			runData: IRun,
+			source?: WorkflowExecutionSource,
 		) => {
-			await service.workflowExecutionCompleted(workflowData, runData);
+			await service.workflowExecutionCompleted(workflowData, runData, source);
 			await flushStats(service);
 		};
 
@@ -183,6 +185,113 @@ describe('WorkflowStatisticsService', () => {
 			}
 		});
 
+		test.each<WorkflowExecuteMode>(['trigger', 'webhook'])(
+			'should count successful instance_ai-sourced execution with mode %s as manual, not production',
+			async (mode) => {
+				// ARRANGE
+				const runData: IRun = {
+					finished: true,
+					status: 'success',
+					data: createEmptyRunExecutionData(),
+					mode,
+					startedAt: new Date(),
+					storedAt: 'db',
+				};
+
+				// ACT
+				await completeAndFlush(workflowStatisticsService, workflow, runData, 'instance_ai');
+
+				// ASSERT
+				const statistics = await workflowStatisticsRepository.find();
+				expect(statistics).toHaveLength(1);
+				expect(statistics[0]).toMatchObject({
+					count: 1,
+					rootCount: 0,
+					name: 'manual_success',
+					workflowId: workflow.id,
+				});
+			},
+		);
+
+		test('should count failing instance_ai-sourced execution as manual error and not emit instance-first-production-workflow-failed', async () => {
+			// ARRANGE
+			const runData: IRun = {
+				finished: false,
+				status: 'error',
+				data: createEmptyRunExecutionData(),
+				mode: 'trigger',
+				startedAt: new Date(),
+				storedAt: 'db',
+			};
+			const emitSpy = vi.spyOn(Container.get(EventService), 'emit');
+
+			// ACT
+			await completeAndFlush(workflowStatisticsService, workflow, runData, 'instance_ai');
+
+			// ASSERT
+			const statistics = await workflowStatisticsRepository.find();
+			expect(statistics).toHaveLength(1);
+			expect(statistics[0]).toMatchObject({
+				count: 1,
+				rootCount: 0,
+				name: 'manual_error',
+				workflowId: workflow.id,
+			});
+			expect(emitSpy).not.toHaveBeenCalledWith(
+				'instance-first-production-workflow-failed',
+				expect.anything(),
+			);
+		});
+
+		test('should not update user settings or emit first-production-workflow-succeeded for instance_ai-sourced executions', async () => {
+			// ARRANGE
+			const runData: IRun = {
+				finished: true,
+				status: 'success',
+				data: createEmptyRunExecutionData(),
+				mode: 'trigger',
+				startedAt: new Date(),
+				storedAt: 'db',
+			};
+			const emitSpy = vi.spyOn(Container.get(EventService), 'emit');
+			const updateSettingsSpy = vi.spyOn(userService, 'updateSettings');
+
+			// ACT
+			await completeAndFlush(workflowStatisticsService, workflow, runData, 'instance_ai');
+
+			// ASSERT
+			expect(updateSettingsSpy).not.toHaveBeenCalled();
+			expect(emitSpy).not.toHaveBeenCalledWith(
+				'first-production-workflow-succeeded',
+				expect.anything(),
+			);
+		});
+
+		test('should count user-sourced executions as production', async () => {
+			// ARRANGE
+			const runData: IRun = {
+				finished: true,
+				status: 'success',
+				data: createEmptyRunExecutionData(),
+				mode: 'trigger',
+				startedAt: new Date(),
+				storedAt: 'db',
+			};
+
+			// ACT
+			await completeAndFlush(workflowStatisticsService, workflow, runData, 'user');
+
+			// ASSERT
+			const statistics = await workflowStatisticsRepository.find();
+			expect(statistics).toHaveLength(1);
+			expect(statistics[0]).toMatchObject({
+				count: 1,
+				rootCount: 1,
+				name: 'production_success',
+				workflowId: workflow.id,
+			});
+		});
+
 		test.each<ExecutionStatus>(['success', 'crashed', 'error'])(
 			'should upsert `count` and `rootCount` for execution status %s',
 			async (status) => {
@@ -878,6 +987,21 @@ describe('WorkflowStatisticsService', () => {
 			});
 		});
 
+		test('should not record data-loaded statistics for instance_ai-sourced executions', async () => {
+			const workflowId = '1';
+			const node = {
+				id: 'abcde',
+				name: 'test node',
+				typeVersion: 1,
+				type: '',
+				position: [0, 0] as [number, number],
+				parameters: {},
+			};
+			await workflowStatisticsService.nodeFetchedData(workflowId, node, 'instance_ai');
+			expect(entityManager.insert).not.toHaveBeenCalled();
+			expect(eventService.emit).not.toHaveBeenCalled();
+		});
+
 		test('should emit event with no `userId` if workflow is owned by team project', async () => {
 			const workflowId = '123';
 			vi.mocked(ownershipService.getPersonalProjectOwnerCached).mockResolvedValueOnce(null);

packages/cli/src/services/workflow-statistics.service.ts

@@ -15,6 +15,7 @@ import {
 	type IRun,
 	type IWorkflowBase,
 	type WorkflowExecuteMode,
+	type WorkflowExecutionSource,
 } from 'n8n-workflow';
 
 import { EventService } from '@/events/event.service';
@@ -58,13 +59,19 @@ const isModeRootExecution = {
 	agent: false,
 } satisfies Record<WorkflowExecuteMode, boolean>;
 
-function getStatisticsNameForCompletedRun(runData: IRun): StatisticsNames | null {
+function getStatisticsNameForCompletedRun(
+	runData: IRun,
+	source?: WorkflowExecutionSource,
+): StatisticsNames | null {
 	const isChatExecution = runData.mode === 'chat';
 	if (isChatExecution || !isCompletedExecutionStatus(runData.status)) {
 		return null;
 	}
 
-	const isManualExecution = runData.mode === 'manual';
+	// Instance AI verification runs mimic the trigger's execution mode, but they
+	// are test runs on the user's behalf — count them as manual so they never
+	// land in production stats or fire first-production milestones.
+	const isManualExecution = runData.mode === 'manual' || source === 'instance_ai';
 	if (isManualExecution) {
 		return runData.status === 'success'
 			? StatisticsNames.manualSuccess
@@ -81,8 +88,12 @@ function isRootExecutionForRun(runData: IRun): boolean {
 }
 
 type WorkflowStatisticsEvents = {
-	nodeFetchedData: { workflowId: string; node: INode };
-	workflowExecutionCompleted: { workflowData: IWorkflowBase; fullRunData: IRun };
+	nodeFetchedData: { workflowId: string; node: INode; source?: WorkflowExecutionSource };
+	workflowExecutionCompleted: {
+		workflowData: IWorkflowBase;
+		fullRunData: IRun;
+		source?: WorkflowExecutionSource;
+	};
 };
 
 @Service()
@@ -102,24 +113,30 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 
 		this.on(
 			'nodeFetchedData',
-			async ({ workflowId, node }) => await this.nodeFetchedData(workflowId, node),
+			async ({ workflowId, node, source }) => await this.nodeFetchedData(workflowId, node, source),
 		);
 		this.on(
 			'workflowExecutionCompleted',
-			async ({ workflowData, fullRunData }) =>
-				await this.workflowExecutionCompleted(workflowData, fullRunData),
+			async ({ workflowData, fullRunData, source }) =>
+				await this.workflowExecutionCompleted(workflowData, fullRunData, source),
 		);
 	}
 
-	async workflowExecutionCompleted(workflowData: IWorkflowBase, runData: IRun): Promise<void> {
-		const statisticsName = getStatisticsNameForCompletedRun(runData);
+	async workflowExecutionCompleted(
+		workflowData: IWorkflowBase,
+		runData: IRun,
+		source?: WorkflowExecutionSource,
+	): Promise<void> {
+		const statisticsName = getStatisticsNameForCompletedRun(runData, source);
+
+		// Instance AI runs mimic trigger modes but are not root production runs.
+		const isRoot = source !== 'instance_ai' && isRootExecutionForRun(runData);
+
 		if (!statisticsName) return;
 
 		const workflowId = workflowData.id;
 		if (!workflowId) return;
 
-		const isRoot = isRootExecutionForRun(runData);
-
 		let upsertResult: Awaited<ReturnType<WorkflowStatisticsRepository['upsertWorkflowStatistics']>>;
 
 		try {
@@ -256,9 +273,17 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 		});
 	}
 
-	async nodeFetchedData(workflowId: string | undefined | null, node: INode): Promise<void> {
+	async nodeFetchedData(
+		workflowId: string | undefined | null,
+		node: INode,
+		source?: WorkflowExecutionSource,
+	): Promise<void> {
 		if (!workflowId) return;
 
+		// Instance AI verification runs must not claim a workflow's
+		// first-data-loaded milestone on the user's behalf.
+		if (source === 'instance_ai') return;
+
 		const insertResult = await this.repository.insertWorkflowStatistics(
 			StatisticsNames.dataLoaded,
 			workflowId,

packages/cli/src/workflows/workflow-execution.service.ts

@@ -264,6 +264,7 @@ export class WorkflowExecutionService {
 						userId: data.userId,
 						dirtyNodeNames: data.dirtyNodeNames,
 						triggerToStartFrom: data.triggerToStartFrom,
+						source: data.source,
 					},
 					// Set this to null so `createRunExecutionData` doesn't initialize it.
 					// Otherwise this would be treated as a resumed execution after waiting.

packages/workflow/src/run-execution-data/run-execution-data.v0.ts

@@ -51,6 +51,6 @@ export interface IRunExecutionDataV0 {
 	/** Data needed for a worker to run a manual execution. */
 	manualData?: Pick<
 		IWorkflowExecutionDataProcess,
-		'dirtyNodeNames' | 'triggerToStartFrom' | 'userId' | 'evaluationRunId'
+		'dirtyNodeNames' | 'triggerToStartFrom' | 'userId' | 'evaluationRunId' | 'source'
 	>;
 }

packages/workflow/src/run-execution-data/run-execution-data.v1.ts

@@ -66,7 +66,7 @@ export interface IRunExecutionDataV1 {
 	/** Data needed for a worker to run a manual execution. */
 	manualData?: Pick<
 		IWorkflowExecutionDataProcess,
-		'dirtyNodeNames' | 'triggerToStartFrom' | 'userId' | 'evaluationRunId'
+		'dirtyNodeNames' | 'triggerToStartFrom' | 'userId' | 'evaluationRunId' | 'source'
 	>;
 
 	/** Metadata about whether and how this execution's data was redacted. */