feat(core): Add source breakdowns to agent telemetry (no-changelog) (#40726)

packages/@n8n/telemetry/src/events/agents.ts

@@ -15,6 +15,34 @@ const agentRunType = z
 	.enum(['test', 'production'])
 	.describe('production means the run executed the published snapshot; test means it ran a draft');
 
+const agentRunSource = z
+	.string()
+	.describe(
+		'Run source: instance-ai, mcp, chat, n8n_chat, n8n_chat_production, workflow, task, subagent, or a chat integration (slack, telegram, linear, discord, whatsapp, teams). Preserve the runtime source value. unknown means the source was unavailable.',
+	);
+
+const agentExecutionCounts = {
+	message_count: z.number().describe('Fresh top-level user turns; delegated child runs excluded'),
+	token_count: z
+		.number()
+		.describe(
+			'Includes LLM calls belonging to no turn (title generation, memory, embeddings), so runs higher than "Agent session metrics".token_count_sum',
+		),
+	tool_call_count: z.number(),
+};
+
+const agentSessionMetrics = {
+	session_count: z.number(),
+	turn_count: z.number(),
+	latency_ms_sum: z.number(),
+	cost_sum: z.number(),
+	token_count_sum: z
+		.number()
+		.describe('Recorded-turn tokens only; reconciles with cost_sum, unlike token_count'),
+	tool_call_count_sum: z.number(),
+	num_skills_sum: z.number(),
+};
+
 // Spread into the session-metrics payload, matching `IAgentConfigurationTelemetryProperties`.
 const agentConfigurationTelemetry = {
 	model: z.string().nullable(),
@@ -175,6 +203,10 @@ export const AGENTS_TELEMETRY = defineTelemetryEvents({
 			run_type: agentRunType,
 			approved: z.boolean(),
 			scope: z.enum(['once', 'session']),
+			counts_by_source: z
+				.record(agentRunSource, z.object({ count: z.literal(1) }))
+				.optional()
+				.describe('One entry for the run source, with count 1 for this approval response'),
 		}),
 	},
 	AGENT_SETUP_COMPLETED: {
@@ -214,15 +246,13 @@ export const AGENTS_TELEMETRY = defineTelemetryEvents({
 				.optional()
 				.describe('Present only for runs with an n8n user — absent for chat integrations and cron'),
 			run_type: agentRunType,
-			message_count: z
-				.number()
-				.describe('Fresh top-level user turns; delegated child runs excluded'),
-			token_count: z
-				.number()
+			...agentExecutionCounts,
+			counts_by_source: z
+				.record(agentRunSource, z.object(agentExecutionCounts))
+				.optional()
 				.describe(
-					'Includes LLM calls belonging to no turn (title generation, memory, embeddings), so runs higher than "Agent session metrics".token_count_sum',
+					'Counters by run source. Each counter sums to its top-level total. Delegated token and tool usage stays with the parent run source.',
 				),
-			tool_call_count: z.number(),
 		}),
 	},
 	AGENT_SESSION_METRICS: {
@@ -236,15 +266,13 @@ export const AGENTS_TELEMETRY = defineTelemetryEvents({
 			agent_type: z.literal('inline').optional(),
 			run_type: agentRunType,
 			turn_status: z.enum(['succeeded', 'failed']),
-			session_count: z.number(),
-			turn_count: z.number(),
-			latency_ms_sum: z.number(),
-			cost_sum: z.number(),
-			token_count_sum: z
-				.number()
-				.describe('Recorded-turn tokens only; reconciles with cost_sum, unlike token_count'),
-			tool_call_count_sum: z.number(),
-			num_skills_sum: z.number(),
+			...agentSessionMetrics,
+			counts_by_source: z
+				.record(agentRunSource, z.object(agentSessionMetrics))
+				.optional()
+				.describe(
+					'Metrics by run source. Turn metrics sum to the top-level totals. session_count and num_skills_sum count each session once per source, so they can overlap across sources.',
+				),
 			...agentConfigurationTelemetry,
 		}),
 	},

packages/cli/src/interfaces.ts

@@ -221,6 +221,7 @@ export interface IAgentExecutionTrackProperties extends ITelemetryTrackPropertie
 	 */
 	user_id?: string;
 	run_type: AgentRunTelemetryType;
+	source?: string;
 	/**
 	 * Fresh user turns only. Resume continuations and delegated child runs do not
 	 * increment this count.
@@ -258,6 +259,7 @@ export interface IAgentTurnFinishedTrackProperties extends ITelemetryTrackProper
 	/** Internal aggregation key only. This must never be emitted to telemetry. */
 	thread_id: string;
 	run_type: AgentRunTelemetryType;
+	source?: string;
 	/** Absent for saved agents; 'inline' for node-embedded agent definitions. */
 	agent_type?: 'inline';
 	turn_status: AgentTurnTelemetryStatus;

packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts

@@ -1900,6 +1900,7 @@ describe('AgentExecutionOrchestratorService', () => {
 			integrationMessageContextService,
 			agentRunTracingService,
 			externalHooks,
+			telemetry,
 		} = makeService();
 		const runtime = makeRuntime([{ type: 'finish', finishReason: 'stop' }]);
 		runtimeCacheService.getRuntime.mockResolvedValue(runtime);
@@ -1956,6 +1957,23 @@ describe('AgentExecutionOrchestratorService', () => {
 				modelId: 'anthropic/claude-sonnet-4-5',
 			}),
 		);
+		const options = runtime.agent.stream.mock.calls[0][1] as ExecutionOptions;
+		options.executionCounter?.incrementMessageCount();
+		options.executionCounter?.incrementToolCallCount();
+		expect(telemetry.trackAgentExecution).toHaveBeenCalledWith({
+			agent_id: agentId,
+			user_id: userId,
+			run_type: 'test',
+			source: 'instance-ai',
+			message_count: 1,
+		});
+		expect(telemetry.trackAgentExecution).toHaveBeenCalledWith({
+			agent_id: agentId,
+			user_id: userId,
+			run_type: 'test',
+			source: 'instance-ai',
+			tool_call_count: 1,
+		});
 	});
 
 	it.each(['chat', 'wake', 'approval'] as const)(
@@ -3911,47 +3929,62 @@ describe('AgentExecutionOrchestratorService', () => {
 		);
 	});
 
-	it('recovers the original run source from the latest suspended execution when resuming', async () => {
-		const {
-			service,
-			checkpointStorage,
-			runtimeCacheService,
-			executionService,
-			agentRunTracingService,
-		} = makeService();
-		const runtime = makeRuntime([{ type: 'finish', finishReason: 'stop' }]);
+	it.each([true, false])(
+		'recovers the original resume source with tracing enabled: %s',
+		async (tracingEnabled) => {
+			const {
+				service,
+				checkpointStorage,
+				runtimeCacheService,
+				executionService,
+				agentRunTracingService,
+				telemetry,
+			} = makeService();
+			const runtime = makeRuntime([{ type: 'finish', finishReason: 'stop' }]);
 
-		checkpointStorage.getStatus.mockResolvedValueOnce({
-			status: 'active',
-			checkpoint: { persistence: { threadId: 'thread-1', resourceId: 'platform-user-1' } },
-		} as never);
-		runtimeCacheService.getRuntime.mockResolvedValue(runtime);
-		executionService.findLatestSuspendedRun.mockResolvedValueOnce({ source: 'telegram' } as never);
+			Object.defineProperty(agentRunTracingService, 'enabled', { value: tracingEnabled });
+			checkpointStorage.getStatus.mockResolvedValueOnce({
+				status: 'active',
+				checkpoint: { persistence: { threadId: 'thread-1', resourceId: 'platform-user-1' } },
+			} as never);
+			runtimeCacheService.getRuntime.mockResolvedValue(runtime);
+			executionService.findLatestSuspendedRun.mockResolvedValueOnce({
+				source: 'telegram',
+			} as never);
 
-		await collect(
-			service.resumeForChat({
-				agentId,
-				projectId,
-				runId: 'run-1',
-				toolCallId: 'tc-1',
-				resumeData: { value: 'yes' },
-				integrationType: 'telegram',
-			}),
-		);
+			await collect(
+				service.resumeForChat({
+					agentId,
+					projectId,
+					runId: 'run-1',
+					toolCallId: 'tc-1',
+					resumeData: { value: 'yes' },
+					integrationType: 'telegram',
+				}),
+			);
 
-		expect(executionService.findLatestSuspendedRun).toHaveBeenCalledWith('thread-1');
-		expect(agentRunTracingService.build).toHaveBeenCalledWith(
-			expect.objectContaining({ source: 'telegram' }),
-		);
-		expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
-			expect.objectContaining({ source: 'telegram' }),
-			expect.any(Date),
-		);
-		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
-			'execution-1',
-			expect.objectContaining({ source: 'telegram' }),
-		);
-	});
+			expect(executionService.findLatestSuspendedRun).toHaveBeenCalledWith('thread-1');
+			expect(agentRunTracingService.build).toHaveBeenCalledWith(
+				expect.objectContaining({ source: 'telegram' }),
+			);
+			expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
+				expect.objectContaining({ source: 'telegram' }),
+				expect.any(Date),
+			);
+			expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+				'execution-1',
+				expect.objectContaining({ source: 'telegram' }),
+			);
+			const options = runtime.agent.resume.mock.calls[0][2] as ExecutionOptions;
+			options.executionCounter?.incrementTokenCount(15);
+			expect(telemetry.trackAgentExecution).toHaveBeenCalledExactlyOnceWith({
+				agent_id: agentId,
+				run_type: 'production',
+				source: 'telegram',
+				token_count: 15,
+			});
+		},
+	);
 
 	it('falls back to source "unknown" when no suspended execution is found on resume', async () => {
 		const {
@@ -4041,40 +4074,6 @@ describe('AgentExecutionOrchestratorService', () => {
 		);
 	});
 
-	it('skips the suspended-run lookup on resume when tracing is disabled', async () => {
-		const {
-			service,
-			checkpointStorage,
-			runtimeCacheService,
-			executionService,
-			agentRunTracingService,
-		} = makeService();
-		const runtime = makeRuntime([{ type: 'finish', finishReason: 'stop' }]);
-
-		Object.defineProperty(agentRunTracingService, 'enabled', { value: false });
-		checkpointStorage.getStatus.mockResolvedValueOnce({
-			status: 'active',
-			checkpoint: { persistence: { threadId: 'thread-1', resourceId: 'platform-user-1' } },
-		} as never);
-		runtimeCacheService.getRuntime.mockResolvedValue(runtime);
-
-		await collect(
-			service.resumeForChat({
-				agentId,
-				projectId,
-				runId: 'run-1',
-				toolCallId: 'tc-1',
-				resumeData: { value: 'yes' },
-				integrationType: 'slack',
-			}),
-		);
-
-		expect(executionService.findLatestSuspendedRun).not.toHaveBeenCalled();
-		expect(agentRunTracingService.build).toHaveBeenCalledWith(
-			expect.objectContaining({ source: 'unknown' }),
-		);
-	});
-
 	it('records resumed chat executions as suspended when they suspend again', async () => {
 		const { service, checkpointStorage, runtimeCacheService, executionService, wakeService } =
 			makeService();

packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts

@@ -1110,6 +1110,7 @@ describe('AgentExecutionService', () => {
 				agentName: 'Agent',
 				projectId: 'project-1',
 				userMessage: 'Run',
+				source: 'instance-ai',
 				record: makeMessageRecord({
 					usage: { promptTokens: 10, completionTokens: 5, totalTokens: 15 },
 					totalCost: 25,
@@ -1145,6 +1146,7 @@ describe('AgentExecutionService', () => {
 				agent_id: 'agent-1',
 				thread_id: 'thread-1',
 				run_type: 'test',
+				source: 'instance-ai',
 				turn_status: 'succeeded',
 				configuration: {
 					model: 'anthropic/claude-sonnet-4-5',

packages/cli/src/modules/agents/__tests__/agent-tool-approval.service.test.ts

@@ -86,15 +86,44 @@ it.each<{ name: string; entry?: ToolRegistryEntry; input: unknown; expected: unk
 	expect(entry).toEqual(originalEntry);
 });
 
-it.each<{ decision: ApprovalResumePayload; allowed: boolean; scope: 'once' | 'session' }>([
-	{ decision: { approved: true }, allowed: false, scope: 'once' },
-	{ decision: { approved: true, scope: 'once' }, allowed: false, scope: 'once' },
-	{ decision: { approved: false, scope: 'session' }, allowed: false, scope: 'session' },
-	{ decision: { approved: true, scope: 'session' }, allowed: true, scope: 'session' },
+it.each<{
+	decision: ApprovalResumePayload;
+	allowed: boolean;
+	scope: 'once' | 'session';
+	source?: string;
+	expectedSource: string;
+}>([
+	{
+		decision: { approved: true },
+		allowed: false,
+		scope: 'once',
+		source: 'subagent',
+		expectedSource: 'subagent',
+	},
+	{
+		decision: { approved: true, scope: 'once' },
+		allowed: false,
+		scope: 'once',
+		source: 'n8n_chat_production',
+		expectedSource: 'n8n_chat_production',
+	},
+	{
+		decision: { approved: false, scope: 'session' },
+		allowed: false,
+		scope: 'session',
+		source: 'slack',
+		expectedSource: 'slack',
+	},
+	{
+		decision: { approved: true, scope: 'session' },
+		allowed: true,
+		scope: 'session',
+		expectedSource: 'unknown',
+	},
 ])(
 	'records the decision and grants only session approvals: $decision',
-	async ({ decision, allowed, scope }) => {
-		const context = await service.createContext(params);
+	async ({ decision, allowed, scope, source, expectedSource }) => {
+		const context = await service.createContext({ ...params, source });
 		await context.onDecision(key, decision);
 		expect(context.approvedKeys.has(key)).toBe(allowed);
 		if (allowed) expect(grants.grant).toHaveBeenCalledWith('child-thread', key);
@@ -107,8 +136,14 @@ it.each<{ decision: ApprovalResumePayload; allowed: boolean; scope: 'once' | 'se
 				run_type: 'test',
 				approved: decision.approved,
 				scope,
+				counts_by_source: { [expectedSource]: { count: 1 } },
 			},
 		);
+		expect(
+			TELEMETRY_EVENT.AGENTS.USER_RESPONDED_TO_AGENT_TOOL_APPROVAL.getValidationError(
+				telemetry.track.mock.calls[0][1],
+			),
+		).toBeNull();
 	},
 );
 

packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts

@@ -308,6 +308,7 @@ describe('AgentWorkflowExecutionService', () => {
 			agent_id: agentId,
 			user_id: userId,
 			run_type: 'production',
+			source: 'workflow',
 			message_count: 1,
 		});
 		expect(executionService.recordTimelineSnapshot).toHaveBeenCalledWith(
@@ -976,6 +977,7 @@ describe('AgentWorkflowExecutionService', () => {
 					agent_id: 'inline:wf-1:Message an Agent',
 					agent_type: 'inline',
 					run_type: 'production',
+					source: 'workflow',
 					turn_status: 'succeeded',
 					configuration: expect.objectContaining({ model: 'anthropic/claude-sonnet-4-5' }),
 				}),

packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts

@@ -1309,7 +1309,7 @@ export class AgentExecutionOrchestratorService {
 			type: 'resume',
 			resumeData: config.resumeData,
 			options: withBudgetGuardrail(
-				this.createResumeOptions(config, memoryScope, messageContext, tracing),
+				this.createResumeOptions(config, memoryScope, messageContext, executionSource, tracing),
 				{
 					budget: runtime.budget,
 					sessionId: memoryScope.threadId,
@@ -1329,10 +1329,10 @@ export class AgentExecutionOrchestratorService {
 		runtime: AgentRuntime,
 	) {
 		const { agentId, projectId, source, user } = config;
-		// Recover the original source only when tracing needs it. `resumeForChat`
+		// Recover the original source for telemetry and tracing. `resumeForChat`
 		// already resolved the n8n Chat source, so this only runs for other callers.
 		const suspendedExecution =
-			this.agentRunTracingService.enabled && source === undefined
+			source === undefined
 				? await this.agentExecutionService.findLatestSuspendedRun(threadId)
 				: undefined;
 		const executionSource = source ?? suspendedExecution?.source ?? undefined;
@@ -1351,6 +1351,7 @@ export class AgentExecutionOrchestratorService {
 		config: ResumeChatConfig,
 		memoryScope: ResumeCheckpoint['memoryScope'],
 		selectedContext: IntegrationMessageContext | null,
+		source: string | undefined,
 		tracing: Awaited<ReturnType<AgentRunTracingService['build']>>,
 	): Extract<AgentTurnRequest, { type: 'resume' }>['options'] {
 		const { runId, toolCallId, agentId, user, usePublishedVersion, abortSignal } = config;
@@ -1374,6 +1375,7 @@ export class AgentExecutionOrchestratorService {
 				agentId,
 				userId: user?.id,
 				runType,
+				source,
 			}),
 			...modelStreamStallOptions(this.aiConfig),
 			...(tracing ? { telemetry: tracing } : {}),
@@ -1697,6 +1699,7 @@ export class AgentExecutionOrchestratorService {
 						agentId,
 						userId,
 						runType: telemetry.runType,
+						source,
 					}),
 					...modelStreamStallOptions(this.aiConfig),
 					...(tracing ? { telemetry: tracing } : {}),

packages/cli/src/modules/agents/agent-execution.service.ts

@@ -1265,6 +1265,7 @@ export class AgentExecutionService {
 				user_id: params.telemetry.userId,
 				thread_id: threadId,
 				run_type: params.telemetry.runType,
+				source: params.source,
 				turn_status: status === 'success' ? 'succeeded' : 'failed',
 				configuration: params.telemetry.configuration,
 				...buildAgentTurnMetrics(record),

packages/cli/src/modules/agents/agent-tool-approval.service.ts

@@ -22,7 +22,8 @@ export class AgentToolApprovalService {
 			threadId,
 			agentId,
 			telemetry,
-		}: Pick<StartExecutionParams, 'threadId' | 'agentId' | 'telemetry'>,
+			source,
+		}: Pick<StartExecutionParams, 'threadId' | 'agentId' | 'telemetry' | 'source'>,
 		toolRegistry?: ToolRegistry,
 	): Promise<ToolApprovalContext> {
 		const approvedKeys = await this.grants.findKeys(threadId);
@@ -37,6 +38,7 @@ export class AgentToolApprovalService {
 						run_type: telemetry.runType,
 						approved: decision.approved,
 						scope: decision.scope ?? 'once',
+						counts_by_source: { [source || 'unknown']: { count: 1 } },
 					});
 				}
 				if (!decision.approved || decision.scope !== 'session') return;

packages/cli/src/modules/agents/agent-workflow-execution.service.ts

@@ -447,6 +447,7 @@ export class AgentWorkflowExecutionService {
 					agentId: telemetryAgentId,
 					userId: telemetryUserId,
 					runType,
+					source: AGENT_WORKFLOW_TRIGGER_TYPE,
 				}),
 				...modelStreamStallOptions(this.aiConfig),
 				...(telemetry ? { telemetry } : {}),
@@ -902,6 +903,7 @@ export class AgentWorkflowExecutionService {
 				user_id: telemetryUserId,
 				thread_id: threadId,
 				run_type: runType,
+				source: AGENT_WORKFLOW_TRIGGER_TYPE,
 				agent_type: 'inline',
 				turn_status:
 					run.messageRecord.error !== null || run.messageRecord.finishReason === 'error'

packages/cli/src/modules/agents/utils/agent-execution-counter.ts

@@ -9,7 +9,8 @@ export function createAgentExecutionCounter(
 		agentId,
 		userId,
 		runType,
-	}: { agentId: string; userId?: string; runType: AgentRunTelemetryType },
+		source,
+	}: { agentId: string; userId?: string; runType: AgentRunTelemetryType; source?: string },
 ): AgentExecutionCounter {
 	const attribution = userId ? { user_id: userId } : {};
 	return {
@@ -18,6 +19,7 @@ export function createAgentExecutionCounter(
 				agent_id: agentId,
 				...attribution,
 				run_type: runType,
+				source,
 				message_count: 1,
 			}),
 		incrementTokenCount: (tokenCount) =>
@@ -25,6 +27,7 @@ export function createAgentExecutionCounter(
 				agent_id: agentId,
 				...attribution,
 				run_type: runType,
+				source,
 				token_count: tokenCount,
 			}),
 		incrementToolCallCount: () =>
@@ -32,6 +35,7 @@ export function createAgentExecutionCounter(
 				agent_id: agentId,
 				...attribution,
 				run_type: runType,
+				source,
 				tool_call_count: 1,
 			}),
 	};

packages/cli/src/telemetry/__tests__/telemetry.test.ts

@@ -483,6 +483,7 @@ describe('Telemetry', () => {
 					message_count: 1,
 					token_count: 15,
 					tool_call_count: 2,
+					counts_by_source: { unknown: { message_count: 1, token_count: 15, tool_call_count: 2 } },
 				},
 				'agent-2:test': {
 					agent_id: 'agent-2',
@@ -490,47 +491,90 @@ describe('Telemetry', () => {
 					message_count: 1,
 					token_count: 0,
 					tool_call_count: 0,
+					counts_by_source: { unknown: { message_count: 1, token_count: 0, tool_call_count: 0 } },
 				},
 			});
 		});
 
-		test('should flush agent execution counters and reset the buffer', () => {
-			telemetry.trackAgentExecution({ agent_id: 'agent-1', run_type: 'test', message_count: 1 });
-			telemetry.trackAgentExecution({ agent_id: 'agent-1', run_type: 'test', token_count: 15 });
-			telemetry.trackAgentExecution({ agent_id: 'agent-1', run_type: 'test', tool_call_count: 2 });
+		test('should flush source counters in one event and reset the buffer', () => {
+			const chatRun = { agent_id: 'agent-1', run_type: 'test' as const, source: 'chat' };
+			telemetry.trackAgentExecution({ ...chatRun, message_count: 2 });
+			telemetry.trackAgentExecution({ ...chatRun, token_count: 10 });
+			telemetry.trackAgentExecution({ ...chatRun, tool_call_count: 1 });
+			telemetry.trackAgentExecution({
+				...chatRun,
+				source: 'instance-ai',
+				message_count: 1,
+				token_count: 5,
+				tool_call_count: 1,
+			});
 
 			// @ts-expect-error Calling private method
 			telemetry.flushAgentExecutionCounts();
 
-			expect(spyTrack).toHaveBeenCalledWith(TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT, {
-				event_version: '2',
-				agent_id: 'agent-1',
-				run_type: 'test',
-				message_count: 1,
-				token_count: 15,
-				tool_call_count: 2,
-			});
+			expect(spyTrack).toHaveBeenCalledExactlyOnceWith(
+				TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT,
+				{
+					event_version: '2',
+					agent_id: 'agent-1',
+					run_type: 'test',
+					message_count: 3,
+					token_count: 15,
+					tool_call_count: 2,
+					counts_by_source: {
+						chat: { message_count: 2, token_count: 10, tool_call_count: 1 },
+						'instance-ai': { message_count: 1, token_count: 5, tool_call_count: 1 },
+					},
+				},
+			);
+			expect(
+				TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT.getValidationError(spyTrack.mock.calls[0][1]),
+			).toBeNull();
 			expect(telemetry.getAgentExecutionCountsBuffer()).toEqual({});
 		});
 
-		test('should bucket test and production runs of the same agent into separate events', () => {
-			telemetry.trackAgentExecution({ agent_id: 'agent-1', run_type: 'test', message_count: 1 });
+		test('should preserve source values within separate test and production events', () => {
+			telemetry.trackAgentExecution({
+				agent_id: 'agent-1',
+				run_type: 'test',
+				source: 'n8n_chat',
+				message_count: 1,
+			});
 			telemetry.trackAgentExecution({
 				agent_id: 'agent-1',
 				run_type: 'production',
-				message_count: 2,
+				source: 'n8n_chat_production',
+				message_count: 1,
+			});
+			telemetry.trackAgentExecution({
+				agent_id: 'agent-1',
+				run_type: 'production',
+				source: 'n8n_chat',
+				message_count: 1,
 			});
 
 			// @ts-expect-error Calling private method
 			telemetry.flushAgentExecutionCounts();
 
+			expect(spyTrack).toHaveBeenCalledTimes(2);
 			expect(spyTrack).toHaveBeenCalledWith(
 				TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT,
-				expect.objectContaining({ run_type: 'test', message_count: 1 }),
+				expect.objectContaining({
+					run_type: 'test',
+					message_count: 1,
+					counts_by_source: { n8n_chat: { message_count: 1, token_count: 0, tool_call_count: 0 } },
+				}),
 			);
 			expect(spyTrack).toHaveBeenCalledWith(
 				TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT,
-				expect.objectContaining({ run_type: 'production', message_count: 2 }),
+				expect.objectContaining({
+					run_type: 'production',
+					message_count: 2,
+					counts_by_source: {
+						n8n_chat_production: { message_count: 1, token_count: 0, tool_call_count: 0 },
+						n8n_chat: { message_count: 1, token_count: 0, tool_call_count: 0 },
+					},
+				}),
 			);
 		});
 
@@ -553,6 +597,7 @@ describe('Telemetry', () => {
 				message_count: 0,
 				token_count: 20,
 				tool_call_count: 0,
+				counts_by_source: { unknown: { message_count: 0, token_count: 20, tool_call_count: 0 } },
 			});
 			expect(telemetry.getAgentExecutionCountsBuffer()).toEqual({});
 		});
@@ -562,9 +607,15 @@ describe('Telemetry', () => {
 				agent_id: 'agent-1',
 				user_id: 'user-1',
 				run_type: 'test',
+				source: 'chat',
 				message_count: 1,
 			});
-			telemetry.trackAgentExecution({ agent_id: 'agent-1', run_type: 'test', token_count: 20 });
+			telemetry.trackAgentExecution({
+				agent_id: 'agent-1',
+				run_type: 'test',
+				source: 'slack',
+				token_count: 20,
+			});
 
 			// @ts-expect-error Calling private method
 			telemetry.flushAgentExecutionCounts();
@@ -577,6 +628,7 @@ describe('Telemetry', () => {
 				message_count: 1,
 				token_count: 0,
 				tool_call_count: 0,
+				counts_by_source: { chat: { message_count: 1, token_count: 0, tool_call_count: 0 } },
 			});
 			expect(spyTrack).toHaveBeenCalledWith(TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT, {
 				event_version: '2',
@@ -585,7 +637,18 @@ describe('Telemetry', () => {
 				message_count: 0,
 				token_count: 20,
 				tool_call_count: 0,
+				counts_by_source: { slack: { message_count: 0, token_count: 20, tool_call_count: 0 } },
 			});
+			expect(mockRudderStack.track).toHaveBeenCalledWith(
+				expect.objectContaining({
+					event: TELEMETRY_EVENT.AGENTS.AGENT_EXECUTION_COUNT.name,
+					userId: instanceId,
+					properties: expect.objectContaining({
+						user_id: undefined,
+						counts_by_source: { slack: { message_count: 0, token_count: 20, tool_call_count: 0 } },
+					}),
+				}),
+			);
 			expect(telemetry.getAgentExecutionCountsBuffer()).toEqual({});
 		});
 
@@ -643,17 +706,19 @@ describe('Telemetry', () => {
 			]);
 		});
 
-		test('should flush session metrics with additive sums', () => {
-			for (const [thread_id, latency_ms, cost, token_count, tool_call_count] of [
-				['thread-1', 100, 10, 1000, 1],
-				['thread-1', 200, 20, 2000, 3],
-				['thread-2', 400, 40, 4000, 5],
-				['thread-3', 800, 80, 8000, 7],
+		test('should split turn metrics by source and count distinct sessions', () => {
+			for (const [thread_id, source, latency_ms, cost, token_count, tool_call_count] of [
+				['thread-1', 'chat', 100, 10, 1000, 1],
+				['thread-1', 'instance-ai', 200, 20, 2000, 3],
+				['thread-2', 'chat', 400, 40, 4000, 5],
+				['thread-3', 'instance-ai', 800, 80, 8000, 7],
+				['thread-1', 'chat', 50, 5, 500, 2],
 			] as const) {
 				telemetry.trackAgentTurnFinished({
 					agent_id: 'agent-1',
 					user_id: 'user-1',
 					thread_id,
+					source,
 					run_type: 'test',
 					turn_status: 'succeeded',
 					configuration,
@@ -670,6 +735,7 @@ describe('Telemetry', () => {
 			const payload = spyTrack.mock.calls.find(
 				([eventName]) => eventName === TELEMETRY_EVENT.AGENTS.AGENT_SESSION_METRICS,
 			)?.[1];
+			expect(spyTrack).toHaveBeenCalledTimes(1);
 			expect(payload).toEqual({
 				event_version: '1',
 				agent_id: 'agent-1',
@@ -678,13 +744,34 @@ describe('Telemetry', () => {
 				run_type: 'test',
 				turn_status: 'succeeded',
 				session_count: 3,
-				turn_count: 4,
-				latency_ms_sum: 1500,
-				cost_sum: 150,
-				token_count_sum: 15000,
-				tool_call_count_sum: 16,
+				turn_count: 5,
+				latency_ms_sum: 1550,
+				cost_sum: 155,
+				token_count_sum: 15500,
+				tool_call_count_sum: 18,
 				num_skills_sum: 6,
+				counts_by_source: {
+					chat: {
+						session_count: 2,
+						turn_count: 3,
+						latency_ms_sum: 550,
+						cost_sum: 55,
+						token_count_sum: 5500,
+						tool_call_count_sum: 8,
+						num_skills_sum: 4,
+					},
+					'instance-ai': {
+						session_count: 2,
+						turn_count: 2,
+						latency_ms_sum: 1000,
+						cost_sum: 100,
+						token_count_sum: 10000,
+						tool_call_count_sum: 10,
+						num_skills_sum: 4,
+					},
+				},
 			});
+			expect(TELEMETRY_EVENT.AGENTS.AGENT_SESSION_METRICS.getValidationError(payload)).toBeNull();
 			expect(payload).not.toHaveProperty('latency_ms_avg');
 			expect(payload).not.toHaveProperty('latency_ms_p25');
 			expect(payload).not.toHaveProperty('cost_avg');
@@ -729,6 +816,7 @@ describe('Telemetry', () => {
 				agent_id: 'agent-1',
 				thread_id: 'thread-1',
 				run_type: 'test',
+				source: 'n8n_chat',
 				turn_status: 'succeeded',
 				configuration,
 				latency_ms: 100,
@@ -740,6 +828,7 @@ describe('Telemetry', () => {
 				agent_id: 'agent-1',
 				thread_id: 'thread-2',
 				run_type: 'production',
+				source: 'n8n_chat_production',
 				turn_status: 'failed',
 				configuration,
 				latency_ms: 200,
@@ -757,6 +846,7 @@ describe('Telemetry', () => {
 					run_type: 'test',
 					turn_status: 'succeeded',
 					latency_ms_sum: 100,
+					counts_by_source: { n8n_chat: expect.objectContaining({ turn_count: 1 }) },
 				}),
 			);
 			expect(spyTrack).toHaveBeenCalledWith(
@@ -765,6 +855,7 @@ describe('Telemetry', () => {
 					run_type: 'production',
 					turn_status: 'failed',
 					latency_ms_sum: 200,
+					counts_by_source: { n8n_chat_production: expect.objectContaining({ turn_count: 1 }) },
 				}),
 			);
 		});

packages/cli/src/telemetry/index.ts

@@ -71,14 +71,18 @@ interface IApiInvocationsBuffer {
 	[userId: string]: IApiInvocationsBufferEntry;
 }
 
+interface IAgentExecutionCounts {
+	message_count: number;
+	token_count: number;
+	tool_call_count: number;
+}
+
 interface IAgentExecutionCountsBuffer {
-	[bufferKey: string]: {
+	[bufferKey: string]: IAgentExecutionCounts & {
 		agent_id: string;
 		user_id?: string;
 		run_type: AgentRunTelemetryType;
-		message_count: number;
-		token_count: number;
-		tool_call_count: number;
+		counts_by_source: Record<string, IAgentExecutionCounts>;
 	};
 }
 
@@ -100,6 +104,7 @@ interface IAgentSessionMetricsBuffer {
 		turn_status: IAgentTurnFinishedTrackProperties['turn_status'];
 		configuration: IAgentConfigurationTelemetryProperties;
 		sessions: Record<string, IAgentSessionMetrics>;
+		sessions_by_source: Record<string, Record<string, IAgentSessionMetrics>>;
 	};
 }
 
@@ -343,18 +348,7 @@ export class Telemetry {
 
 	private flushAgentSessionMetrics() {
 		for (const bucket of Object.values(this.agentSessionMetricsBuffer)) {
-			const sessions = Object.values(bucket.sessions);
-			if (sessions.length === 0) continue;
-
-			const latencyMsSum = sessions.reduce((total, session) => total + session.latency_ms, 0);
-			const costSum = sessions.reduce((total, session) => total + session.cost, 0);
-			const tokenCountSum = sessions.reduce((total, session) => total + session.token_count, 0);
-			const toolCallCountSum = sessions.reduce(
-				(total, session) => total + session.tool_call_count,
-				0,
-			);
-			const numSkillsSum = sessions.reduce((total, session) => total + session.num_skills, 0);
-			const turnCount = sessions.reduce((total, session) => total + session.turn_count, 0);
+			if (Object.keys(bucket.sessions).length === 0) continue;
 
 			this.track(TELEMETRY_EVENT.AGENTS.AGENT_SESSION_METRICS, {
 				event_version: '1',
@@ -364,19 +358,32 @@ export class Telemetry {
 				...bucket.configuration,
 				run_type: bucket.run_type,
 				turn_status: bucket.turn_status,
-				session_count: sessions.length,
-				turn_count: turnCount,
-				latency_ms_sum: latencyMsSum,
-				cost_sum: costSum,
-				token_count_sum: tokenCountSum,
-				tool_call_count_sum: toolCallCountSum,
-				num_skills_sum: numSkillsSum,
+				...this.sumAgentSessionMetrics(bucket.sessions),
+				counts_by_source: Object.fromEntries(
+					Object.entries(bucket.sessions_by_source).map(([source, sessions]) => [
+						source,
+						this.sumAgentSessionMetrics(sessions),
+					]),
+				),
 			});
 		}
 
 		this.agentSessionMetricsBuffer = {};
 	}
 
+	private sumAgentSessionMetrics(sessionsById: Record<string, IAgentSessionMetrics>) {
+		const sessions = Object.values(sessionsById);
+		return {
+			session_count: sessions.length,
+			turn_count: sessions.reduce((total, session) => total + session.turn_count, 0),
+			latency_ms_sum: sessions.reduce((total, session) => total + session.latency_ms, 0),
+			cost_sum: sessions.reduce((total, session) => total + session.cost, 0),
+			token_count_sum: sessions.reduce((total, session) => total + session.token_count, 0),
+			tool_call_count_sum: sessions.reduce((total, session) => total + session.tool_call_count, 0),
+			num_skills_sum: sessions.reduce((total, session) => total + session.num_skills, 0),
+		};
+	}
+
 	trackWorkflowExecution(properties: IExecutionTrackProperties) {
 		if (this.rudderStack) {
 			const execTime = new Date();
@@ -457,12 +464,23 @@ export class Telemetry {
 			message_count: 0,
 			token_count: 0,
 			tool_call_count: 0,
+			counts_by_source: {},
 		};
 
 		const agentExecutionCounts = this.agentExecutionCountsBuffer[bufferKey];
 		agentExecutionCounts.message_count += message_count;
 		agentExecutionCounts.token_count += token_count;
 		agentExecutionCounts.tool_call_count += tool_call_count;
+
+		const source = properties.source || 'unknown';
+		const sourceCounts = (agentExecutionCounts.counts_by_source[source] ??= {
+			message_count: 0,
+			token_count: 0,
+			tool_call_count: 0,
+		});
+		sourceCounts.message_count += message_count;
+		sourceCounts.token_count += token_count;
+		sourceCounts.tool_call_count += tool_call_count;
 	}
 
 	trackAgentTurnFinished(properties: IAgentTurnFinishedTrackProperties) {
@@ -477,24 +495,30 @@ export class Telemetry {
 			turn_status: properties.turn_status,
 			configuration: properties.configuration,
 			sessions: {},
+			sessions_by_source: {},
 		};
 
 		const bucket = this.agentSessionMetricsBuffer[bufferKey];
-		const session = bucket.sessions[properties.thread_id] ?? {
-			latency_ms: 0,
-			cost: 0,
-			token_count: 0,
-			tool_call_count: 0,
-			num_skills: properties.configuration.num_skills,
-			turn_count: 0,
-		};
+		const source = properties.source || 'unknown';
+		const sourceSessions = (bucket.sessions_by_source[source] ??= {});
+		// A session can contain turns from several sources. Keep the total deduplicated.
+		for (const sessions of [bucket.sessions, sourceSessions]) {
+			const session = sessions[properties.thread_id] ?? {
+				latency_ms: 0,
+				cost: 0,
+				token_count: 0,
+				tool_call_count: 0,
+				num_skills: properties.configuration.num_skills,
+				turn_count: 0,
+			};
 
-		session.latency_ms += properties.latency_ms;
-		session.cost += properties.cost;
-		session.token_count += properties.token_count;
-		session.tool_call_count += properties.tool_call_count;
-		session.turn_count++;
-		bucket.sessions[properties.thread_id] = session;
+			session.latency_ms += properties.latency_ms;
+			session.cost += properties.cost;
+			session.token_count += properties.token_count;
+			session.tool_call_count += properties.tool_call_count;
+			session.turn_count++;
+			sessions[properties.thread_id] = session;
+		}
 	}
 
 	trackApiInvocation(properties: IApiInvocationProperties) {