feat(ai) - add observability (#20850)

**AI Chat - Tool Executions (counters, tagged with model)**
ai-chat/tool-execution-succeeded: number of tool calls invoked by the AI
that completed without error
ai-chat/tool-execution-failed: number of tool calls invoked by the AI
that threw an error
**AI Chat - Token Usage (counters, tagged with model)**
ai-chat/input-tokens: total input tokens sent to the model across all
turns
ai-chat/output-tokens: total output tokens generated by the model
ai-chat/cache-read-tokens: input tokens served from the model's prompt
cache (cheaper)
ai-chat/cache-write-tokens: input tokens written into the prompt cache
for future reuse
**AI Chat - Latency (histograms in ms, tagged with model)**
ai-chat/turn-latency-ms: total duration of a full chat turn (from stream
start to stream end)
ai-chat/step-latency-ms: duration of a single reasoning/tool-call step
within a turn
ai-chat/ttft-ms: time-to-first-token, i.e. how long until the model
starts streaming output
**MCP - Tool Executions (counters)**
mcp/tool-execution-succeeded: number of MCP tool calls that completed
successfully
mcp/tool-execution-failed: number of MCP tool calls that threw an error

packages/twenty-server/src/engine/api/common/common-query-runners/common-base-query-runner.service.ts

@@ -385,7 +385,7 @@ export abstract class CommonBaseQueryRunnerService<
         longConfig.timeWindow,
       );
     } catch (error) {
-      await this.metricsService.incrementCounter({
+      await this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.CommonApiQueryRateLimited,
         shouldStoreInCache: false,
       });

packages/twenty-server/src/engine/api/mcp/mcp.module.ts

@@ -5,6 +5,7 @@ import { McpAuthGuard } from 'src/engine/api/mcp/guards/mcp-auth.guard';
 import { McpProtocolService } from 'src/engine/api/mcp/services/mcp-protocol.service';
 import { McpToolExecutorService } from 'src/engine/api/mcp/services/mcp-tool-executor.service';
 import { ApiKeyModule } from 'src/engine/core-modules/api-key/api-key.module';
+import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { TokenModule } from 'src/engine/core-modules/auth/token/token.module';
 import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
 import { ToolProviderModule } from 'src/engine/core-modules/tool-provider/tool-provider.module';
@@ -17,6 +18,7 @@ import { WorkspaceCacheStorageModule } from 'src/engine/workspace-cache-storage/
 @Module({
   imports: [
     ApiKeyModule,
+    MetricsModule,
     TokenModule,
     WorkspaceCacheStorageModule,
     UserRoleModule,

packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-tool-executor.service.spec.ts

@@ -4,13 +4,18 @@ import {
   MCP_PROGRESS_NOTIFICATION_METHOD,
   TOOL_CALL_PROGRESS_TOKEN_PREFIX,
 } from 'src/engine/api/mcp/constants/mcp-progress-notification.const';
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
 import { McpToolExecutorService } from 'src/engine/api/mcp/services/mcp-tool-executor.service';
 
 describe('McpToolExecutorService', () => {
   let service: McpToolExecutorService;
 
   beforeEach(() => {
-    service = new McpToolExecutorService();
+    const metricsService = {
+      incrementCounterBy: jest.fn().mockResolvedValue(undefined),
+    } as unknown as MetricsService;
+
+    service = new McpToolExecutorService(metricsService);
   });
 
   describe('handleToolsListing', () => {

packages/twenty-server/src/engine/api/mcp/services/mcp-tool-executor.service.ts

@@ -3,6 +3,9 @@ import { Injectable } from '@nestjs/common';
 import { type ToolSet } from 'ai';
 import { isDefined } from 'twenty-shared/utils';
 
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+
 import { JSON_RPC_ERROR_CODE } from 'src/engine/api/mcp/constants/json-rpc-error-code.const';
 import {
   MCP_PROGRESS_NOTIFICATION_METHOD,
@@ -22,6 +25,8 @@ const unwrapJsonSchema = (schema: unknown) =>
 
 @Injectable()
 export class McpToolExecutorService {
+  constructor(private readonly metricsService: MetricsService) {}
+
   async handleToolCall(
     id: string | number,
     toolSet: ToolSet,
@@ -58,6 +63,11 @@ export class McpToolExecutorService {
         messages: [],
       });
 
+      void this.metricsService.incrementCounterBy({
+        key: MetricsKeys.McpToolExecutionSucceeded,
+        amount: 1,
+      });
+
       return wrapJsonRpcResponse(id, {
         result: {
           content: [{ type: 'text', text: JSON.stringify(result) }],
@@ -65,6 +75,11 @@ export class McpToolExecutorService {
         },
       });
     } catch (executionError) {
+      void this.metricsService.incrementCounterBy({
+        key: MetricsKeys.McpToolExecutionFailed,
+        amount: 1,
+      });
+
       return wrapJsonRpcResponse(id, {
         result: {
           content: [

packages/twenty-server/src/engine/core-modules/auth/services/sign-in-up.service.spec.ts

@@ -86,7 +86,7 @@ const createSignInUpServiceForTests = () => {
       markEmailAsVerified: jest.fn(),
     } as any,
     {
-      incrementCounter: jest.fn(),
+      incrementCounterForEvent: jest.fn(),
     } as any,
     {
       invalidateAndRecompute: jest.fn(),

packages/twenty-server/src/engine/core-modules/auth/services/sign-in-up.service.ts

@@ -384,7 +384,7 @@ export class SignInUpService {
       undefined,
     );
 
-    void this.metricsService.incrementCounter({
+    void this.metricsService.incrementCounterForEvent({
       key: MetricsKeys.SignUpSuccess,
       shouldStoreInCache: false,
     });

packages/twenty-server/src/engine/core-modules/captcha/captcha.guard.ts

@@ -32,7 +32,7 @@ export class CaptchaGuard implements CanActivate {
     if (result.success) {
       return true;
     } else {
-      await this.metricsService.incrementCounter({
+      await this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.InvalidCaptcha,
         eventId: token || '',
         ...(result.error ? { attributes: { error: result.error } } : {}),

packages/twenty-server/src/engine/core-modules/graphql/hooks/use-graphql-error-handler.hook.ts

@@ -116,7 +116,7 @@ export const useGraphQLErrorHandlerHook = <
             setResult,
           }) => {
             if (!result.errors || result.errors.length === 0) {
-              void options.metricsService.incrementCounter({
+              void options.metricsService.incrementCounterForEvent({
                 key: MetricsKeys.GraphqlOperation200,
               });
 
@@ -178,11 +178,11 @@ export const useGraphQLErrorHandlerHook = <
               }
 
               if (metricKey) {
-                void options.metricsService.incrementCounter({
+                void options.metricsService.incrementCounterForEvent({
                   key: metricKey,
                 });
               } else {
-                void options.metricsService.incrementCounter({
+                void options.metricsService.incrementCounterForEvent({
                   key: MetricsKeys.GraphqlOperationUnknown,
                 });
               }
@@ -282,7 +282,7 @@ export const useGraphQLErrorHandlerHook = <
           isDefined(currentMetadataVersion) &&
           requestMetadataVersion !== `${currentMetadataVersion}`
         ) {
-          void options.metricsService.incrementCounter({
+          void options.metricsService.incrementCounterForEvent({
             key: MetricsKeys.SchemaVersionMismatch,
           });
 
@@ -312,7 +312,7 @@ export const useGraphQLErrorHandlerHook = <
           isDefined(backendMajor) &&
           frontEndMajor < backendMajor
         ) {
-          void options.metricsService.incrementCounter({
+          void options.metricsService.incrementCounterForEvent({
             key: MetricsKeys.AppVersionMismatch,
           });
           throw new GraphQLError(APP_VERSION_MISMATCH_ERROR, {

packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts

@@ -136,7 +136,7 @@ export class BullMQDriver
     );
 
     this.workerMap[queueName].on('completed', (job) => {
-      void this.metricsService.incrementCounter({
+      void this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.JobCompleted,
         attributes: { queue: queueName, job_name: job?.name ?? '' },
         shouldStoreInCache: false,
@@ -148,7 +148,7 @@ export class BullMQDriver
         return;
       }
 
-      void this.metricsService.incrementCounter({
+      void this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.JobFailed,
         attributes: {
           queue: queueName,

packages/twenty-server/src/engine/core-modules/metrics/metrics.service.ts

@@ -80,7 +80,7 @@ export class MetricsService {
     return gauge;
   }
 
-  async incrementCounter({
+  async incrementCounterForEvent({
     key,
     eventId,
     attributes,
@@ -110,7 +110,7 @@ export class MetricsService {
     }
   }
 
-  async batchIncrementCounter({
+  async incrementCounterForEvents({
     key,
     eventIds,
     attributes,
@@ -130,6 +130,32 @@ export class MetricsService {
     }
   }
 
+  incrementCounterBy({
+    key,
+    amount,
+    attributes,
+  }: {
+    key: MetricsKeys;
+    amount: number;
+    attributes?: Attributes;
+  }): void {
+    this.getMeter().createCounter(key).add(amount, attributes);
+  }
+
+  recordHistogram({
+    key,
+    value,
+    unit,
+    attributes,
+  }: {
+    key: MetricsKeys;
+    value: number;
+    unit?: string;
+    attributes?: Attributes;
+  }): void {
+    this.getMeter().createHistogram(key, { unit }).record(value, attributes);
+  }
+
   async groupMetrics(
     metrics: { name: string; cacheKey: MetricsKeys }[],
   ): Promise<Record<string, number>> {

packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts

@@ -23,8 +23,14 @@ export enum MetricsKeys {
   WorkflowRunThrottled = 'workflow-run/throttled',
   WorkflowRunFailedToEnqueue = 'workflow-run/failed/to-enqueue',
   WorkflowRunSystemError = 'workflow-run/system-error',
-  AiToolExecutionFailed = 'ai-tool-execution/failed',
-  AiToolExecutionSucceeded = 'ai-tool-execution/succeeded',
+  AiChatToolExecutionSucceeded = 'ai-chat/tool-execution-succeeded',
+  AiChatToolExecutionFailed = 'ai-chat/tool-execution-failed',
+  McpToolExecutionSucceeded = 'mcp/tool-execution-succeeded',
+  McpToolExecutionFailed = 'mcp/tool-execution-failed',
+  AiChatInputTokens = 'ai-chat/input-tokens',
+  AiChatOutputTokens = 'ai-chat/output-tokens',
+  AiChatCacheReadTokens = 'ai-chat/cache-read-tokens',
+  AiChatCacheWriteTokens = 'ai-chat/cache-write-tokens',
   SchemaVersionMismatch = 'schema-version/mismatch',
   AppVersionMismatch = 'app-version/mismatch',
   CronJobDeletedWorkspace = 'cron-job/deleted-workspace',
@@ -34,4 +40,7 @@ export enum MetricsKeys {
   JobCompleted = 'job/completed',
   JobFailed = 'job/failed',
   JobWaiting = 'job/waiting',
+  AiChatTurnLatencyMs = 'ai-chat/turn-latency-ms',
+  AiChatStepLatencyMs = 'ai-chat/step-latency-ms',
+  AiChatTtftMs = 'ai-chat/ttft-ms',
 }

packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/ai-chat.module.ts

@@ -12,6 +12,7 @@ import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user
 import { UserWorkspaceModule } from 'src/engine/core-modules/user-workspace/user-workspace.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { AiAgentExecutionModule } from 'src/engine/metadata-modules/ai/ai-agent-execution/ai-agent-execution.module';
+import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { AiBillingModule } from 'src/engine/metadata-modules/ai/ai-billing/ai-billing.module';
 import { AiGraphqlApiExceptionInterceptor } from 'src/engine/metadata-modules/ai/interceptors/ai-graphql-api-exception.interceptor';
 import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permissions.module';
@@ -56,6 +57,7 @@ import { SystemPromptBuilderService } from './services/system-prompt-builder.ser
     TokenModule,
     UserWorkspaceModule,
     AiBillingModule,
+    MetricsModule,
     ToolProviderModule,
     DashboardToolsModule,
     WorkflowToolsModule,

packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/chat-execution.service.ts

@@ -15,6 +15,9 @@ import {
 import { AppPath } from 'twenty-shared/types';
 import { getAppPath, isDefined } from 'twenty-shared/utils';
 
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { type ToolOutput } from 'src/engine/core-modules/tool/types/tool-output.type';
 import { UsageOperationType } from 'src/engine/core-modules/usage/enums/usage-operation-type.enum';
 
 import { type CodeExecutionStreamEmitter } from 'src/engine/core-modules/tool-provider/interfaces/code-execution-stream-emitter.type';
@@ -95,6 +98,7 @@ export class ChatExecutionService {
     private readonly exceptionHandlerService: ExceptionHandlerService,
     private readonly nativeToolBinder: NativeToolBinderService,
     private readonly messagePruningService: MessagePruningService,
+    private readonly metricsService: MetricsService,
   ) {}
 
   async streamChat({
@@ -275,6 +279,9 @@ export class ChatExecutionService {
     const modelMessages = pruningResult.messages;
 
     let hasNoMoreAvailableCredits = false;
+    const streamStartedAt = performance.now();
+    let stepStartedAt = streamStartedAt;
+    let ttftRecorded = false;
 
     const emitTurnUsageEvent = async (steps: StepResult<ToolSet>[]) => {
       const usage = steps.reduce<LanguageModelUsage>(
@@ -347,6 +354,35 @@ export class ChatExecutionService {
         workspace.id,
         userWorkspaceId,
       );
+
+      const modelAttr = { model: registeredModel.modelId };
+
+      this.metricsService.incrementCounterBy({
+        key: MetricsKeys.AiChatInputTokens,
+        amount: usage.inputTokens ?? 0,
+        attributes: modelAttr,
+      });
+      this.metricsService.incrementCounterBy({
+        key: MetricsKeys.AiChatOutputTokens,
+        amount: usage.outputTokens ?? 0,
+        attributes: modelAttr,
+      });
+      this.metricsService.incrementCounterBy({
+        key: MetricsKeys.AiChatCacheReadTokens,
+        amount: usage.inputTokenDetails?.cacheReadTokens ?? 0,
+        attributes: modelAttr,
+      });
+      this.metricsService.incrementCounterBy({
+        key: MetricsKeys.AiChatCacheWriteTokens,
+        amount: cacheCreationTokens,
+        attributes: modelAttr,
+      });
+      this.metricsService.recordHistogram({
+        key: MetricsKeys.AiChatTurnLatencyMs,
+        value: performance.now() - streamStartedAt,
+        unit: 'ms',
+        attributes: modelAttr,
+      });
     };
 
     const stream = streamText({
@@ -360,10 +396,35 @@ export class ChatExecutionService {
       providerOptions: getCallLevelCacheProviderOptions(
         registeredModel.sdkPackage,
       ),
-      prepareStep: ({ messages }) => ({
-        messages: injectCacheBreakpoint(messages, registeredModel.sdkPackage),
-      }),
+      prepareStep: ({ messages }) => {
+        stepStartedAt = performance.now();
+
+        return {
+          messages: injectCacheBreakpoint(messages, registeredModel.sdkPackage),
+        };
+      },
+      onChunk: ({ chunk }) => {
+        if (
+          !ttftRecorded &&
+          (chunk.type === 'text-delta' || chunk.type === 'tool-call')
+        ) {
+          ttftRecorded = true;
+          this.metricsService.recordHistogram({
+            key: MetricsKeys.AiChatTtftMs,
+            value: performance.now() - streamStartedAt,
+            unit: 'ms',
+            attributes: { model: registeredModel.modelId },
+          });
+        }
+      },
       onStepFinish: async (step) => {
+        this.metricsService.recordHistogram({
+          key: MetricsKeys.AiChatStepLatencyMs,
+          value: performance.now() - stepStartedAt,
+          unit: 'ms',
+          attributes: { model: registeredModel.modelId },
+        });
+
         const { hasNoMoreAvailableCredits: stepHasNoMoreAvailableCredits } =
           await this.aiBillingService.decrementAndCheckAvailableCredits(
             registeredModel.modelId,
@@ -379,6 +440,22 @@ export class ChatExecutionService {
         if (stepHasNoMoreAvailableCredits) {
           hasNoMoreAvailableCredits = true;
         }
+
+        for (const toolResult of step.toolResults) {
+          const output = toolResult.output as ToolOutput | undefined;
+
+          if (!isDefined(output?.success)) {
+            continue;
+          }
+
+          void this.metricsService.incrementCounterForEvent({
+            key: output.success
+              ? MetricsKeys.AiChatToolExecutionSucceeded
+              : MetricsKeys.AiChatToolExecutionFailed,
+            attributes: { model: registeredModel.modelId },
+            shouldStoreInCache: false,
+          });
+        }
       },
       onAbort: async ({ steps }) => {
         await emitTurnUsageEvent(steps);

packages/twenty-server/src/engine/metadata-modules/webhook/jobs/call-webhook.job.ts

@@ -95,7 +95,7 @@ export class CallWebhookJob {
         ...commonPayload,
       });
 
-      void this.metricsService.incrementCounter({
+      void this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.JobWebhookCallCompleted,
         shouldStoreInCache: false,
       });

packages/twenty-server/src/engine/workspace-manager/workspace-cleaner/services/cleaner.workspace-service.ts

@@ -378,7 +378,7 @@ export class CleanerWorkspaceService {
     }
 
     await this.workspaceService.deleteWorkspace(workspace.id);
-    void this.metricsService.incrementCounter({
+    void this.metricsService.incrementCounterForEvent({
       key: MetricsKeys.CronJobDeletedWorkspace,
       shouldStoreInCache: false,
     });

packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts

@@ -239,7 +239,7 @@ export class CalendarChannelSyncStatusService {
       workspaceId,
     );
 
-    await this.metricsService.batchIncrementCounter({
+    await this.metricsService.incrementCounterForEvents({
       key: MetricsKeys.CalendarEventSyncJobActive,
       eventIds: calendarChannelIds,
     });
@@ -275,7 +275,7 @@ export class CalendarChannelSyncStatusService {
       { lite: true },
     );
 
-    await this.metricsService.batchIncrementCounter({
+    await this.metricsService.incrementCounterForEvents({
       key: MetricsKeys.CalendarEventSyncJobFailedUnknown,
       eventIds: calendarChannelIds,
     });
@@ -333,7 +333,7 @@ export class CalendarChannelSyncStatusService {
       { lite: true },
     );
 
-    await this.metricsService.batchIncrementCounter({
+    await this.metricsService.incrementCounterForEvents({
       key: MetricsKeys.CalendarEventSyncJobFailedInsufficientPermissions,
       eventIds: calendarChannelIds,
     });

packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts

@@ -245,7 +245,7 @@ export class MessageChannelSyncStatusService {
       { lite: true },
     );
 
-    await this.metricsService.batchIncrementCounter({
+    await this.metricsService.incrementCounterForEvents({
       key: MetricsKeys.MessageChannelSyncJobActive,
       eventIds: messageChannelIds,
     });
@@ -331,7 +331,7 @@ export class MessageChannelSyncStatusService {
             ? MetricsKeys.MessageChannelSyncJobFailedInsufficientPermissions
             : MetricsKeys.MessageChannelSyncJobFailedUnknown;
 
-        await this.metricsService.batchIncrementCounter({
+        await this.metricsService.incrementCounterForEvents({
           key: metricsKey,
           eventIds: messageChannelIds,
         });

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts

@@ -86,7 +86,7 @@ describe('WorkflowExecutorWorkspaceService', () => {
   };
 
   const mockMetricsService = {
-    incrementCounter: jest.fn(),
+    incrementCounterForEvent: jest.fn(),
   };
 
   const mockMessageQueueService = {

packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts

@@ -519,7 +519,7 @@ export class WorkflowExecutorWorkspaceService {
           workspace: { id: workspaceId },
         });
 
-        await this.metricsService.incrementCounter({
+        await this.metricsService.incrementCounterForEvent({
           key: MetricsKeys.WorkflowRunSystemError,
           eventId: workflowRunId,
           debugLog: `[Workflow Run System Error] Workflow run ${workflowRunId} in workspace ${workspaceId} ended with system error`,

packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts

@@ -236,7 +236,7 @@ export class RunWorkflowJob {
         throw new Error('Invalid trigger type');
     }
 
-    await this.metricsService.incrementCounter({
+    await this.metricsService.incrementCounterForEvent({
       key,
       eventId: workflowRunId,
     });

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts

@@ -154,7 +154,7 @@ export class WorkflowRunEnqueueWorkspaceService {
       );
     } catch (error) {
       try {
-        await this.metricsService.incrementCounter({
+        await this.metricsService.incrementCounterForEvent({
           key: MetricsKeys.WorkflowRunFailedToEnqueue,
           eventId: workspaceId,
         });

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts

@@ -225,13 +225,13 @@ export class WorkflowRunWorkspaceService {
           ? MetricsKeys.WorkflowRunStopped
           : MetricsKeys.WorkflowRunFailed;
 
-    await this.metricsService.incrementCounter({
+    await this.metricsService.incrementCounterForEvent({
       key: metricKey,
       eventId: workflowRunId,
     });
 
     if (isSystemError) {
-      await this.metricsService.incrementCounter({
+      await this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.WorkflowRunSystemError,
         eventId: workflowRunId,
         debugLog: `[Workflow Run System Error] Workflow run ${workflowRunId} in workspace ${workspaceId} ended with system error`,

packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

@@ -269,7 +269,7 @@ export class WorkflowRunnerWorkspaceService {
 
       return false;
     } catch {
-      void this.metricsService.incrementCounter({
+      void this.metricsService.incrementCounterForEvent({
         key: MetricsKeys.WorkflowRunThrottled,
         eventId: workspaceId,
       });