[AI] Add group_by_* database tools and centralize groupBy validation (#19406)

closes https://discord.com/channels/1130383047699738754/1488990242873806868 https://github.com/user-attachments/assets/2b2bbfba-3fa6-4114-9a26-96a61599d748 <img width="729" height="1283" alt="CleanShot 2026-04-07 at 20 43 06" src="https://github.com/user-attachments/assets/815efb97-81a0-44ea-8d79-b3ce7d5b00b6" /> <img width="708" height="1266" alt="CleanShot 2026-04-07 at 20 40 13" src="https://github.com/user-attachments/assets/692366bc-b629-4d9f-b6b8-ab670d5ad046" /> <img width="665" height="3524" alt="CleanShot 2026-04-07 at 20 42 00" src="https://github.com/user-attachments/assets/5e844e0f-7835-47a8-9d20-a5baddc0992d" />
2026-04-10 14:15:18 +05:30
parent 43ce396152
commit f13e7e01fe
32 changed files with 1504 additions and 258 deletions
@@ -1,9 +1,31 @@
 import { Injectable } from '@nestjs/common';
+import {
+  AggregateOperations,
+  type RestrictedFieldsPermissions,
+} from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+
+import { type GroupByField } from 'src/engine/api/common/common-query-runners/types/group-by-field.types';
+import {
+  CommonQueryRunnerException,
+  CommonQueryRunnerExceptionCode,
+} from 'src/engine/api/common/common-query-runners/errors/common-query-runner.exception';
+import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';
+import { validateAndTransformGroupByFieldsOrThrow } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/validate-and-transform-group-by-fields-or-throw.util';
+import {
+  type AggregationField,
+  getAvailableAggregationsFromObjectFields,
+} from 'src/engine/api/graphql/workspace-schema-builder/utils/get-available-aggregations-from-object-fields.util';
 import {
  ObjectRecordGroupByForAtomicField,
  ObjectRecordGroupByForCompositeField,
  ObjectRecordGroupByForDateField,
 } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';
+import { findManyFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-many-flat-entity-by-id-in-flat-entity-maps.util';
+import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type';
+import { resolveAggregateFieldKey } from 'src/engine/core-modules/record-crud/utils/resolve-aggregate-field-key.util';

@Injectable()
 export class GroupByArgProcessorService {
@@ -30,4 +52,113 @@ export class GroupByArgProcessorService {

    return [groupBy];
  }
+
+  validateAndTransformGroupByFieldsOrThrow({
+    groupBy,
+    flatObjectMetadata,
+    flatObjectMetadataMaps,
+    flatFieldMetadataMaps,
+  }: {
+    groupBy: Array<
+      | ObjectRecordGroupByForAtomicField
+      | ObjectRecordGroupByForCompositeField
+      | ObjectRecordGroupByForDateField
+    >;
+    flatObjectMetadata: FlatObjectMetadata;
+    flatObjectMetadataMaps: FlatEntityMaps<FlatObjectMetadata>;
+    flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+  }): GroupByField[] {
+    return validateAndTransformGroupByFieldsOrThrow({
+      groupBy,
+      flatObjectMetadata,
+      flatObjectMetadataMaps,
+      flatFieldMetadataMaps,
+    });
+  }
+
+  getAvailableAggregations({
+    flatObjectMetadata,
+    flatFieldMetadataMaps,
+    restrictedFields,
+  }: {
+    flatObjectMetadata: FlatObjectMetadata;
+    flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+    restrictedFields?: RestrictedFieldsPermissions;
+  }): Record<string, AggregationField> {
+    const objectFields = findManyFlatEntityByIdInFlatEntityMaps({
+      flatEntityIds: flatObjectMetadata.fieldIds,
+      flatEntityMaps: flatFieldMetadataMaps,
+    }).filter((field) => restrictedFields?.[field.id]?.canRead !== false);
+
+    return getAvailableAggregationsFromObjectFields(objectFields);
+  }
+
+  validateAggregateFieldKeysOrThrow({
+    aggregateFieldKeys,
+    availableAggregations,
+  }: {
+    aggregateFieldKeys: string[];
+    availableAggregations: Record<string, AggregationField>;
+  }): void {
+    const invalidAggregateFieldKeys = aggregateFieldKeys.filter(
+      (aggregateFieldKey) =>
+        !isDefined(availableAggregations[aggregateFieldKey]),
+    );
+
+    if (invalidAggregateFieldKeys.length === 0) {
+      return;
+    }
+
+    throw new CommonQueryRunnerException(
+      `Unknown aggregate field(s): ${invalidAggregateFieldKeys.join(', ')}`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  resolveToolAggregateFieldKeyOrThrow({
+    aggregateOperation,
+    aggregateFieldName,
+    availableAggregations,
+  }: {
+    aggregateOperation: keyof typeof AggregateOperations;
+    aggregateFieldName?: string;
+    availableAggregations: Record<string, AggregationField>;
+  }): string {
+    if (aggregateOperation === AggregateOperations.COUNT) {
+      if (aggregateFieldName) {
+        throw new CommonQueryRunnerException(
+          'aggregateFieldName is not supported for COUNT operation',
+          CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+          { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+        );
+      }
+
+      return 'totalCount';
+    }
+
+    if (!aggregateFieldName) {
+      throw new CommonQueryRunnerException(
+        `aggregateFieldName is required for ${aggregateOperation} operation`,
+        CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+        { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+      );
+    }
+
+    const aggregateFieldKey = resolveAggregateFieldKey(
+      aggregateOperation,
+      aggregateFieldName,
+      availableAggregations,
+    );
+
+    if (!aggregateFieldKey) {
+      throw new CommonQueryRunnerException(
+        `No aggregation available for ${aggregateOperation} on field "${aggregateFieldName}"`,
+        CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+        { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+      );
+    }
+
+    return aggregateFieldKey;
+  }
 }
@@ -1,6 +1,6 @@
 import {
-  type ObjectRecordGroupByDateGranularity,
  type FirstDayOfTheWeek,
+  type ObjectRecordGroupByDateGranularity,
 } from 'twenty-shared/types';

 export type DateFieldGroupByDefinition = {
@@ -1,5 +1,5 @@
-import { type CompositeFieldGroupByDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/types/composite-field-group-by-definition.type';
-import { type DateFieldGroupByDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/types/date-field-group-by-definition.type';
+import { type CompositeFieldGroupByDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/types/composite-field-group-by-definition.type';
+import { type DateFieldGroupByDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/types/date-field-group-by-definition.type';

 export type FieldGroupByDefinition =
  | boolean
@@ -0,0 +1,27 @@
+import { ObjectRecordGroupByDateGranularity } from 'twenty-shared/types';
+import { isPlainObject } from 'twenty-shared/utils';
+
+import { type DateFieldGroupByDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/types/date-field-group-by-definition.type';
+
+const GROUP_BY_DATE_GRANULARITIES = new Set<string>(
+  Object.values(ObjectRecordGroupByDateGranularity),
+);
+
+export const isGroupByDateFieldDefinition = (
+  fieldGroupByDefinition: unknown,
+): fieldGroupByDefinition is DateFieldGroupByDefinition => {
+  if (!isPlainObject(fieldGroupByDefinition)) {
+    return false;
+  }
+
+  if (!('granularity' in fieldGroupByDefinition)) {
+    return false;
+  }
+
+  const granularity = fieldGroupByDefinition.granularity;
+
+  return (
+    typeof granularity === 'string' &&
+    GROUP_BY_DATE_GRANULARITIES.has(granularity)
+  );
+};
@@ -0,0 +1,16 @@
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { isFlatFieldMetadataSupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-supported-in-group-by.util';
+
+export const isRelationNestedFieldSupportedInGroupBy = ({
+  nestedFieldName,
+  nestedFieldMetadata,
+}: {
+  nestedFieldName: string;
+  nestedFieldMetadata: FlatFieldMetadata;
+}): boolean => {
+  if (nestedFieldName === 'id') {
+    return true;
+  }
+
+  return isFlatFieldMetadataSupportedInGroupBy(nestedFieldMetadata);
+};
@@ -0,0 +1,271 @@
+import { FieldMetadataType } from 'twenty-shared/types';
+import { isDefined, isPlainObject } from 'twenty-shared/utils';
+
+import { type GroupByField } from 'src/engine/api/common/common-query-runners/types/group-by-field.types';
+import {
+  CommonQueryRunnerException,
+  CommonQueryRunnerExceptionCode,
+} from 'src/engine/api/common/common-query-runners/errors/common-query-runner.exception';
+import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';
+import { isGroupByDateFieldDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/is-group-by-date-field-definition.util';
+import { validateAndTransformRelationGroupByFieldOrThrow } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/validate-and-transform-relation-group-by-field-or-throw.util';
+import { validateSingleKeyForGroupByOrThrow } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/validate-single-key-for-group-by-or-throw.util';
+import {
+  ObjectRecordGroupByForAtomicField,
+  ObjectRecordGroupByForCompositeField,
+  ObjectRecordGroupByForDateField,
+} from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';
+import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
+import { findFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps.util';
+import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { getGroupableSubFieldsForCompositeType } from 'src/engine/metadata-modules/field-metadata/utils/get-groupable-sub-fields-for-composite-type.util';
+import { isFlatFieldMetadataSupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-supported-in-group-by.util';
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { buildFieldMapsFromFlatObjectMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/build-field-maps-from-flat-object-metadata.util';
+import { isMorphOrRelationFlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-morph-or-relation-flat-field-metadata.util';
+import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type';
+
+const getFieldMetadataForGroupByOrThrow = ({
+  fieldName,
+  fieldIdByName,
+  fieldIdByJoinColumnName,
+  flatFieldMetadataMaps,
+}: {
+  fieldName: string;
+  fieldIdByName: Record<string, string>;
+  fieldIdByJoinColumnName: Record<string, string>;
+  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+}): FlatFieldMetadata => {
+  const fieldMetadataId =
+    fieldIdByName[fieldName] || fieldIdByJoinColumnName[fieldName];
+  const fieldMetadata = fieldMetadataId
+    ? findFlatEntityByIdInFlatEntityMaps({
+        flatEntityId: fieldMetadataId,
+        flatEntityMaps: flatFieldMetadataMaps,
+      })
+    : undefined;
+
+  if (!isDefined(fieldMetadata) || !isDefined(fieldMetadataId)) {
+    throw new CommonQueryRunnerException(
+      `Unidentified field in groupBy: ${fieldName}`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  return fieldMetadata;
+};
+
+const validateAndTransformCompositeGroupByDefinitionOrThrow = ({
+  fieldName,
+  fieldMetadata,
+  fieldGroupByDefinition,
+  groupByFields,
+}: {
+  fieldName: string;
+  fieldMetadata: FlatFieldMetadata;
+  fieldGroupByDefinition: Record<string, unknown>;
+  groupByFields: GroupByField[];
+}) => {
+  if (!isCompositeFieldMetadataType(fieldMetadata.type)) {
+    throw new CommonQueryRunnerException(
+      `Field "${fieldName}" does not support nested subfields in groupBy`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  const supportedCompositeSubFields = getGroupableSubFieldsForCompositeType(
+    fieldMetadata.type,
+  );
+
+  validateSingleKeyForGroupByOrThrow({
+    groupByKeys: Object.keys(fieldGroupByDefinition),
+    errorMessage:
+      'You cannot provide multiple subfields in one GroupByInput, split them into multiple GroupByInput',
+  });
+
+  for (const subFieldName of Object.keys(fieldGroupByDefinition)) {
+    if (
+      isCompositeFieldMetadataType(fieldMetadata.type) &&
+      !supportedCompositeSubFields?.includes(subFieldName)
+    ) {
+      throw new CommonQueryRunnerException(
+        `Composite subfield "${subFieldName}" is not supported in groupBy for "${fieldName}"`,
+        CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+        { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+      );
+    }
+
+    if (fieldGroupByDefinition[subFieldName] !== true) {
+      throw new CommonQueryRunnerException(
+        `Composite subfield "${subFieldName}" must be set to true in groupBy for "${fieldName}"`,
+        CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+        { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+      );
+    }
+
+    groupByFields.push({
+      fieldMetadata,
+      subFieldName,
+    });
+  }
+};
+
+const validateAndTransformSingleGroupByFieldOrThrow = ({
+  fieldNames,
+  fieldName,
+  fieldIdByName,
+  fieldIdByJoinColumnName,
+  flatObjectMetadataMaps,
+  flatFieldMetadataMaps,
+  groupByFields,
+}: {
+  fieldNames: Record<string, unknown>;
+  fieldName: string;
+  fieldIdByName: Record<string, string>;
+  fieldIdByJoinColumnName: Record<string, string>;
+  flatObjectMetadataMaps: FlatEntityMaps<FlatObjectMetadata>;
+  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+  groupByFields: GroupByField[];
+}) => {
+  const fieldMetadata = getFieldMetadataForGroupByOrThrow({
+    fieldName,
+    fieldIdByName,
+    fieldIdByJoinColumnName,
+    flatFieldMetadataMaps,
+  });
+
+  if (!isFlatFieldMetadataSupportedInGroupBy(fieldMetadata)) {
+    throw new CommonQueryRunnerException(
+      `Field "${fieldName}" is not supported in groupBy`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  const fieldGroupByDefinition = fieldNames[fieldName];
+  const isObjectFieldGroupByDefinition = isPlainObject(fieldGroupByDefinition);
+  const isGroupByRelationField =
+    isMorphOrRelationFlatFieldMetadata(fieldMetadata) &&
+    isObjectFieldGroupByDefinition &&
+    !isGroupByDateFieldDefinition(fieldGroupByDefinition);
+  const isGroupByRelationJoinColumnField =
+    isMorphOrRelationFlatFieldMetadata(fieldMetadata) &&
+    fieldGroupByDefinition === true &&
+    isDefined(fieldIdByJoinColumnName[fieldName]);
+
+  if (isGroupByRelationField || isGroupByRelationJoinColumnField) {
+    const normalizedFieldNames = isGroupByRelationJoinColumnField
+      ? { ...fieldNames, [fieldName]: { id: true } }
+      : fieldNames;
+
+    validateAndTransformRelationGroupByFieldOrThrow({
+      fieldNames: normalizedFieldNames,
+      fieldName,
+      fieldMetadata,
+      flatObjectMetadataMaps,
+      flatFieldMetadataMaps,
+      groupByFields,
+    });
+
+    return;
+  }
+
+  if (
+    (fieldMetadata.type === FieldMetadataType.DATE ||
+      fieldMetadata.type === FieldMetadataType.DATE_TIME) &&
+    isGroupByDateFieldDefinition(fieldGroupByDefinition)
+  ) {
+    groupByFields.push({
+      fieldMetadata,
+      dateGranularity: fieldGroupByDefinition.granularity,
+      weekStartDay: fieldGroupByDefinition.weekStartDay,
+      timeZone: fieldGroupByDefinition.timeZone,
+    });
+
+    return;
+  }
+
+  if (isObjectFieldGroupByDefinition && 'unnest' in fieldGroupByDefinition) {
+    groupByFields.push({
+      fieldMetadata,
+      subFieldName: undefined,
+      shouldUnnest: true,
+    });
+
+    return;
+  }
+
+  if (fieldGroupByDefinition === true) {
+    groupByFields.push({
+      fieldMetadata,
+      subFieldName: undefined,
+    });
+
+    return;
+  }
+
+  if (isObjectFieldGroupByDefinition) {
+    validateAndTransformCompositeGroupByDefinitionOrThrow({
+      fieldName,
+      fieldMetadata,
+      fieldGroupByDefinition,
+      groupByFields,
+    });
+
+    return;
+  }
+
+  throw new CommonQueryRunnerException(
+    `Invalid groupBy definition for field "${fieldName}"`,
+    CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+    { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+  );
+};
+
+export const validateAndTransformGroupByFieldsOrThrow = ({
+  groupBy,
+  flatObjectMetadata,
+  flatObjectMetadataMaps,
+  flatFieldMetadataMaps,
+}: {
+  groupBy: Array<
+    | ObjectRecordGroupByForAtomicField
+    | ObjectRecordGroupByForCompositeField
+    | ObjectRecordGroupByForDateField
+  >;
+  flatObjectMetadata: FlatObjectMetadata;
+  flatObjectMetadataMaps: FlatEntityMaps<FlatObjectMetadata>;
+  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+}): GroupByField[] => {
+  const groupByFields: GroupByField[] = [];
+
+  const { fieldIdByName, fieldIdByJoinColumnName } =
+    buildFieldMapsFromFlatObjectMetadata(
+      flatFieldMetadataMaps,
+      flatObjectMetadata,
+    );
+
+  for (const fieldNames of groupBy) {
+    validateSingleKeyForGroupByOrThrow({
+      groupByKeys: Object.keys(fieldNames),
+      errorMessage:
+        'You cannot provide multiple fields in one GroupByInput, split them into multiple GroupByInput',
+    });
+
+    for (const fieldName of Object.keys(fieldNames)) {
+      validateAndTransformSingleGroupByFieldOrThrow({
+        fieldNames,
+        fieldName,
+        fieldIdByName,
+        fieldIdByJoinColumnName,
+        flatObjectMetadataMaps,
+        flatFieldMetadataMaps,
+        groupByFields,
+      });
+    }
+  }
+
+  return groupByFields;
+};
@@ -1,21 +1,20 @@
 import { FieldMetadataType } from 'twenty-shared/types';
-import { isDefined } from 'twenty-shared/utils';
+import { isDefined, isPlainObject } from 'twenty-shared/utils';

-import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';
+import { type CompositeFieldGroupByDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/types/composite-field-group-by-definition.type';
+import { isGroupByDateFieldDefinition } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/is-group-by-date-field-definition.util';
+import { isRelationNestedFieldSupportedInGroupBy } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/is-relation-nested-field-supported-in-group-by.util';
+import { validateSingleKeyForGroupByOrThrow } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/utils/validate-single-key-for-group-by-or-throw.util';
 import {
-  GraphqlQueryRunnerException,
-  GraphqlQueryRunnerExceptionCode,
-} from 'src/engine/api/graphql/graphql-query-runner/errors/graphql-query-runner.exception';
-import { type CompositeFieldGroupByDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/types/composite-field-group-by-definition.type';
-import { type DateFieldGroupByDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/types/date-field-group-by-definition.type';
+  CommonQueryRunnerException,
+  CommonQueryRunnerExceptionCode,
+} from 'src/engine/api/common/common-query-runners/errors/common-query-runner.exception';
+import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';
 import { type GroupByField } from 'src/engine/api/common/common-query-runners/types/group-by-field.types';
-import { isGroupByDateFieldDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/is-group-by-date-field-definition.util';
-import { validateSingleKeyForGroupByOrThrow } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/validate-single-key-for-group-by-or-throw.util';
-import { UserInputError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
 import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { getGroupableSubFieldsForCompositeType } from 'src/engine/metadata-modules/field-metadata/utils/get-groupable-sub-fields-for-composite-type.util';
 import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
 import { findFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps.util';
-import { findFlatEntityByIdInFlatEntityMapsOrThrow } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps-or-throw.util';
 import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
 import { buildFieldMapsFromFlatObjectMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/build-field-maps-from-flat-object-metadata.util';
 import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type';
@@ -33,22 +32,37 @@ const getNestedFieldMetadataDetails = ({
  flatObjectMetadataMaps: FlatEntityMaps<FlatObjectMetadata>;
  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
 }) => {
-  const nestedFieldGroupByDefinitions = fieldNames[fieldName] as
-    | Record<string, boolean>
-    | Record<string, CompositeFieldGroupByDefinition>
-    | Record<string, DateFieldGroupByDefinition>;
+  const nestedFieldGroupByDefinitions = fieldNames[fieldName];

-  if (!isDefined(fieldMetadata.relationTargetObjectMetadataId)) {
-    throw new UserInputError(
-      `Relation target object metadata id not found for field ${fieldMetadata.name}`,
+  if (!isPlainObject(nestedFieldGroupByDefinitions)) {
+    throw new CommonQueryRunnerException(
+      `Invalid groupBy definition for relation field "${fieldName}"`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
    );
  }

-  const targetObjectMetadata = findFlatEntityByIdInFlatEntityMapsOrThrow({
+  if (!isDefined(fieldMetadata.relationTargetObjectMetadataId)) {
+    throw new CommonQueryRunnerException(
+      `Relation target object metadata id not found for field ${fieldMetadata.name}`,
+      CommonQueryRunnerExceptionCode.INTERNAL_SERVER_ERROR,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  const targetObjectMetadata = findFlatEntityByIdInFlatEntityMaps({
    flatEntityId: fieldMetadata.relationTargetObjectMetadataId,
    flatEntityMaps: flatObjectMetadataMaps,
  });

+  if (!isDefined(targetObjectMetadata)) {
+    throw new CommonQueryRunnerException(
+      `Target object metadata not found for relation field ${fieldMetadata.name}`,
+      CommonQueryRunnerExceptionCode.INTERNAL_SERVER_ERROR,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
  const nestedFieldNames = Object.keys(nestedFieldGroupByDefinitions);

  validateSingleKeyForGroupByOrThrow({
@@ -70,16 +84,31 @@ const getNestedFieldMetadataDetails = ({
  });

  if (!isDefined(nestedFieldMetadata) || !isDefined(nestedFieldMetadataId)) {
-    throw new GraphqlQueryRunnerException(
+    throw new CommonQueryRunnerException(
      `Nested field "${nestedFieldName}" not found in target object "${targetObjectMetadata.nameSingular}"`,
-      GraphqlQueryRunnerExceptionCode.FIELD_NOT_FOUND,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  if (
+    !isRelationNestedFieldSupportedInGroupBy({
+      nestedFieldName,
+      nestedFieldMetadata,
+    })
+  ) {
+    throw new CommonQueryRunnerException(
+      `Nested field "${nestedFieldName}" is not supported in groupBy`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
    );
  }

  if (nestedFieldMetadata.type === FieldMetadataType.RELATION) {
-    throw new UserInputError(
+    throw new CommonQueryRunnerException(
      `Cannot group by a relation field of the relation field: "${nestedFieldName}" is a relation field of "${targetObjectMetadata.nameSingular}"`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
    );
  }

@@ -93,57 +122,59 @@ const getNestedFieldMetadataDetails = ({
  };
 };

-const handleNestedCompositeField = ({
+const validateAndTransformNestedCompositeFieldOrThrow = ({
  nestedFieldGroupByDefinition,
  nestedFieldName,
  fieldMetadata,
  nestedFieldMetadata,
  groupByFields,
 }: {
-  nestedFieldGroupByDefinition: CompositeFieldGroupByDefinition;
+  nestedFieldGroupByDefinition: unknown;
  nestedFieldName: string;
  fieldMetadata: FlatFieldMetadata;
  nestedFieldMetadata: FlatFieldMetadata;
  groupByFields: GroupByField[];
 }) => {
-  if (
-    typeof nestedFieldGroupByDefinition === 'object' &&
-    nestedFieldGroupByDefinition !== null
-  ) {
-    const compositeSubFields = Object.keys(
-      nestedFieldGroupByDefinition as Record<string, unknown>,
+  if (!isPlainObject(nestedFieldGroupByDefinition)) {
+    throw new CommonQueryRunnerException(
+      `Composite field "${nestedFieldName}" requires a subfield to be specified`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
    );
-
-    validateSingleKeyForGroupByOrThrow({
-      groupByKeys: compositeSubFields,
-      errorMessage:
-        'You cannot provide multiple composite subfields in one GroupByInput, split them into multiple GroupByInput',
-    });
-
-    const nestedSubFieldName = compositeSubFields[0];
-
-    if (
-      (nestedFieldGroupByDefinition as Record<string, boolean>)[
-        nestedSubFieldName
-      ] === true
-    ) {
-      groupByFields.push({
-        fieldMetadata,
-        nestedFieldMetadata,
-        nestedSubFieldName,
-      });
-
-      return;
-    }
  }
-  throw new GraphqlQueryRunnerException(
-    `Composite field "${nestedFieldName}" requires a subfield to be specified`,
-    GraphqlQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
-    { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+
+  const compositeSubFields = Object.keys(nestedFieldGroupByDefinition);
+
+  validateSingleKeyForGroupByOrThrow({
+    groupByKeys: compositeSubFields,
+    errorMessage:
+      'You cannot provide multiple composite subfields in one GroupByInput, split them into multiple GroupByInput',
+  });
+
+  const nestedSubFieldName = compositeSubFields[0];
+  const supportedCompositeSubFields = getGroupableSubFieldsForCompositeType(
+    nestedFieldMetadata.type,
  );
+
+  if (
+    !supportedCompositeSubFields?.includes(nestedSubFieldName) ||
+    nestedFieldGroupByDefinition[nestedSubFieldName] !== true
+  ) {
+    throw new CommonQueryRunnerException(
+      `Composite subfield "${nestedSubFieldName}" is not supported in groupBy for "${nestedFieldName}"`,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+    );
+  }
+
+  groupByFields.push({
+    fieldMetadata,
+    nestedFieldMetadata,
+    nestedSubFieldName,
+  });
 };

-export const parseGroupByRelationField = ({
+export const validateAndTransformRelationGroupByFieldOrThrow = ({
  fieldNames,
  fieldName,
  fieldMetadata,
@@ -167,7 +198,6 @@ export const parseGroupByRelationField = ({
      flatFieldMetadataMaps,
    });

-  // Handle date fields in nested relations
  if (
    (nestedFieldMetadata.type === FieldMetadataType.DATE ||
      nestedFieldMetadata.type === FieldMetadataType.DATE_TIME) &&
@@ -186,19 +216,18 @@ export const parseGroupByRelationField = ({
    return;
  }

-  // Handle composite fields in nested relations
  if (isCompositeFieldMetadataType(nestedFieldMetadata.type)) {
-    handleNestedCompositeField({
-      nestedFieldGroupByDefinition:
-        nestedFieldGroupByDefinition as CompositeFieldGroupByDefinition,
+    validateAndTransformNestedCompositeFieldOrThrow({
+      nestedFieldGroupByDefinition,
      nestedFieldName,
      fieldMetadata,
      nestedFieldMetadata,
      groupByFields,
    });
+
+    return;
  }

-  // Handle regular nested fields
  if (nestedFieldGroupByDefinition === true) {
    groupByFields.push({
      fieldMetadata,
@@ -207,4 +236,10 @@ export const parseGroupByRelationField = ({

    return;
  }
+
+  throw new CommonQueryRunnerException(
+    `Invalid groupBy definition for nested field "${fieldName}.${nestedFieldName}"`,
+    CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+    { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
+  );
 };
@@ -1,8 +1,8 @@
-import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';
 import {
-  GraphqlQueryRunnerException,
-  GraphqlQueryRunnerExceptionCode,
-} from 'src/engine/api/graphql/graphql-query-runner/errors/graphql-query-runner.exception';
+  CommonQueryRunnerException,
+  CommonQueryRunnerExceptionCode,
+} from 'src/engine/api/common/common-query-runners/errors/common-query-runner.exception';
+import { STANDARD_ERROR_MESSAGE } from 'src/engine/api/common/common-query-runners/errors/standard-error-message.constant';

 export const validateSingleKeyForGroupByOrThrow = ({
  groupByKeys,
@@ -12,9 +12,9 @@ export const validateSingleKeyForGroupByOrThrow = ({
  errorMessage: string;
 }): void => {
  if (groupByKeys.length > 1) {
-    throw new GraphqlQueryRunnerException(
+    throw new CommonQueryRunnerException(
      errorMessage,
-      GraphqlQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
+      CommonQueryRunnerExceptionCode.INVALID_QUERY_INPUT,
      { userFriendlyMessage: STANDARD_ERROR_MESSAGE },
    );
  }
@@ -41,7 +41,6 @@ import {
 import { CommonSelectedFieldsResult } from 'src/engine/api/common/types/common-selected-fields-result.type';
 import { GraphqlQueryParser } from 'src/engine/api/graphql/graphql-query-runner/graphql-query-parsers/graphql-query.parser';
 import { formatResultWithGroupByDimensionValues } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/format-result-with-group-by-dimension-values.util';
-import { parseGroupByArgs } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/parse-group-by-args.util';
 import { GroupByWithRecordsService } from 'src/engine/api/graphql/graphql-query-runner/group-by/services/group-by-with-records.service';
 import { getGroupLimit } from 'src/engine/api/graphql/graphql-query-runner/group-by/utils/get-group-limit.util';
 import { ProcessAggregateHelper } from 'src/engine/api/graphql/graphql-query-runner/helpers/process-aggregate.helper';
@@ -94,12 +93,13 @@ export class CommonGroupByQueryRunnerService extends CommonBaseQueryRunnerServic
      objectMetadataNameSingular,
    );

-    const groupByFields = parseGroupByArgs(
-      args,
-      flatObjectMetadata,
-      flatObjectMetadataMaps,
-      flatFieldMetadataMaps,
-    );
+    const groupByFields =
+      this.groupByArgProcessor.validateAndTransformGroupByFieldsOrThrow({
+        groupBy: args.groupBy,
+        flatObjectMetadata,
+        flatObjectMetadataMaps,
+        flatFieldMetadataMaps,
+      });

    const objectAlias = getObjectAlias(flatObjectMetadata);

@@ -393,9 +393,20 @@ export class CommonGroupByQueryRunnerService extends CommonBaseQueryRunnerServic
  }

  async validate(
-    _args: CommonInput<GroupByQueryArgs>,
-    _queryRunnerContext: CommonBaseQueryRunnerContext,
-  ): Promise<void> {}
+    args: CommonInput<GroupByQueryArgs>,
+    queryRunnerContext: CommonBaseQueryRunnerContext,
+  ): Promise<void> {
+    const normalizedGroupBy = this.groupByArgProcessor.process({
+      groupBy: args.groupBy,
+    });
+
+    this.groupByArgProcessor.validateAndTransformGroupByFieldsOrThrow({
+      groupBy: normalizedGroupBy,
+      flatObjectMetadata: queryRunnerContext.flatObjectMetadata,
+      flatObjectMetadataMaps: queryRunnerContext.flatObjectMetadataMaps,
+      flatFieldMetadataMaps: queryRunnerContext.flatFieldMetadataMaps,
+    });
+  }

  async computeArgs(
    args: CommonInput<GroupByQueryArgs>,
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';

 import { CommonArgsProcessors } from 'src/engine/api/common/common-args-processors/common-args-processors';
+import { GroupByArgProcessorService } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/group-by-arg-processor.service';
 import { ProcessNestedRelationsV2Helper } from 'src/engine/api/common/common-nested-relations-processor/process-nested-relations-v2.helper';
 import { ProcessNestedRelationsHelper } from 'src/engine/api/common/common-nested-relations-processor/process-nested-relations.helper';
 import { CommonQueryRunners } from 'src/engine/api/common/common-query-runners/common-query-runners';
@@ -53,6 +54,6 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache
    CommonResultGettersService,
    GroupByWithRecordsService,
  ],
-  exports: [...CommonQueryRunners],
+  exports: [...CommonQueryRunners, GroupByArgProcessorService],
 })
 export class CoreCommonApiModule {}
@@ -1,30 +0,0 @@
-import { isDefined } from 'class-validator';
-import { ObjectRecordGroupByDateGranularity } from 'twenty-shared/types';
-
-import { type FieldGroupByDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/types/field-group-by-definition.type';
-
-export const isGroupByDateFieldDefinition = (
-  fieldGroupByDefinition: FieldGroupByDefinition,
-): fieldGroupByDefinition is {
-  granularity: ObjectRecordGroupByDateGranularity;
-} => {
-  if (
-    typeof fieldGroupByDefinition !== 'object' ||
-    !isDefined(fieldGroupByDefinition)
-  ) {
-    return false;
-  }
-  if (!('granularity' in fieldGroupByDefinition)) {
-    return false;
-  }
-
-  const granularity = fieldGroupByDefinition.granularity;
-
-  return (
-    isDefined(granularity) &&
-    typeof granularity === 'string' &&
-    Object.values(ObjectRecordGroupByDateGranularity).includes(
-      granularity as ObjectRecordGroupByDateGranularity,
-    )
-  );
-};
@@ -1,141 +0,0 @@
-import { FieldMetadataType } from 'twenty-shared/types';
-import { isDefined } from 'twenty-shared/utils';
-
-import { type GroupByResolverArgs } from 'src/engine/api/graphql/workspace-resolver-builder/interfaces/workspace-resolvers-builder.interface';
-
-import { type GroupByField } from 'src/engine/api/common/common-query-runners/types/group-by-field.types';
-import { isGroupByDateFieldDefinition } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/is-group-by-date-field-definition.util';
-import { parseGroupByRelationField } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/parse-group-by-relation-field.util';
-import { validateSingleKeyForGroupByOrThrow } from 'src/engine/api/graphql/graphql-query-runner/group-by/resolvers/utils/validate-single-key-for-group-by-or-throw.util';
-import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
-import { findFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps.util';
-import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
-import { buildFieldMapsFromFlatObjectMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/build-field-maps-from-flat-object-metadata.util';
-import { isMorphOrRelationFlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-morph-or-relation-flat-field-metadata.util';
-import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type';
-
-export const parseGroupByArgs = (
-  args: GroupByResolverArgs,
-  flatObjectMetadata: FlatObjectMetadata,
-  flatObjectMetadataMaps: FlatEntityMaps<FlatObjectMetadata>,
-  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>,
-): GroupByField[] => {
-  const groupByFieldNames = args.groupBy;
-
-  const groupByFields: GroupByField[] = [];
-
-  const { fieldIdByName, fieldIdByJoinColumnName } =
-    buildFieldMapsFromFlatObjectMetadata(
-      flatFieldMetadataMaps,
-      flatObjectMetadata,
-    );
-
-  for (const fieldNames of groupByFieldNames) {
-    validateSingleKeyForGroupByOrThrow({
-      groupByKeys: Object.keys(fieldNames),
-      errorMessage:
-        'You cannot provide multiple fields in one GroupByInput, split them into multiple GroupByInput',
-    });
-
-    for (const fieldName of Object.keys(fieldNames)) {
-      const fieldMetadataId =
-        fieldIdByName[fieldName] || fieldIdByJoinColumnName[fieldName];
-      const fieldMetadata = fieldMetadataId
-        ? findFlatEntityByIdInFlatEntityMaps({
-            flatEntityId: fieldMetadataId,
-            flatEntityMaps: flatFieldMetadataMaps,
-          })
-        : undefined;
-
-      if (!isDefined(fieldMetadata) || !isDefined(fieldMetadataId)) {
-        throw new Error(`Unidentified field in groupBy: ${fieldName}`);
-      }
-
-      const isGroupByRelationField =
-        isMorphOrRelationFlatFieldMetadata(fieldMetadata) &&
-        typeof fieldNames[fieldName] === 'object' &&
-        fieldNames[fieldName] !== null &&
-        !isGroupByDateFieldDefinition(fieldNames[fieldName]);
-
-      // Handle relation fields
-      if (isGroupByRelationField) {
-        parseGroupByRelationField({
-          fieldNames,
-          fieldName,
-          fieldMetadata,
-          flatObjectMetadataMaps,
-          flatFieldMetadataMaps,
-          groupByFields,
-        });
-
-        continue;
-      }
-
-      // Handle date fields
-      if (
-        fieldMetadata.type === FieldMetadataType.DATE ||
-        fieldMetadata.type === FieldMetadataType.DATE_TIME
-      ) {
-        const fieldGroupByDefinition = fieldNames[fieldName];
-
-        const shouldGroupByDateGranularity = isGroupByDateFieldDefinition(
-          fieldGroupByDefinition,
-        );
-
-        if (shouldGroupByDateGranularity) {
-          groupByFields.push({
-            fieldMetadata,
-            dateGranularity: fieldGroupByDefinition.granularity,
-            weekStartDay: fieldGroupByDefinition.weekStartDay,
-            timeZone: fieldGroupByDefinition.timeZone,
-          });
-          continue;
-        }
-      }
-
-      // Handle array unnest fields
-      if (
-        typeof fieldNames[fieldName] === 'object' &&
-        fieldNames[fieldName] !== null &&
-        'unnest' in fieldNames[fieldName]
-      ) {
-        groupByFields.push({
-          fieldMetadata,
-          subFieldName: undefined,
-          shouldUnnest: true,
-        });
-        continue;
-      }
-
-      // Handle regular fields and composite fields
-      if (fieldNames[fieldName] === true) {
-        groupByFields.push({
-          fieldMetadata,
-          subFieldName: undefined,
-        });
-        continue;
-      } else if (typeof fieldNames[fieldName] === 'object') {
-        validateSingleKeyForGroupByOrThrow({
-          groupByKeys: Object.keys(fieldNames[fieldName]),
-          errorMessage:
-            'You cannot provide multiple subfields in one GroupByInput, split them into multiple GroupByInput',
-        });
-
-        for (const subFieldName of Object.keys(fieldNames[fieldName])) {
-          if (
-            (fieldNames[fieldName] as Record<string, boolean>)[subFieldName] ===
-            true
-          ) {
-            groupByFields.push({
-              fieldMetadata,
-              subFieldName,
-            });
-            continue;
-          }
-        }
-      }
-    }
-  }
-
-  return groupByFields;
-};
@@ -12,6 +12,7 @@ import { GqlInputTypeDefinitionKind } from 'src/engine/api/graphql/workspace-sch
 import { TypeMapperService } from 'src/engine/api/graphql/workspace-schema-builder/services/type-mapper.service';
 import { GqlTypesStorage } from 'src/engine/api/graphql/workspace-schema-builder/storages/gql-types.storage';
 import { computeCompositeFieldInputTypeKey } from 'src/engine/api/graphql/workspace-schema-builder/utils/compute-stored-gql-type-key-utils/compute-composite-field-input-type-key.util';
+import { isCompositePropertySupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-property-supported-in-group-by.util';
 import { isMorphOrRelationFieldMetadataType } from 'src/engine/utils/is-morph-or-relation-field-metadata-type.util';

@Injectable()
@@ -54,8 +55,7 @@ export class CompositeFieldMetadataGroupByGqlInputTypeGenerator {
        throw new Error('Relation fields are not supported in composite types');
      }

-      // Skip hidden fields
-      if (property.hidden === true) {
+      if (!isCompositePropertySupportedInGroupBy(property)) {
        continue;
      }

@@ -20,6 +20,7 @@ import { type SchemaGenerationContext } from 'src/engine/api/graphql/workspace-s
 import { computeCompositeFieldInputTypeKey } from 'src/engine/api/graphql/workspace-schema-builder/utils/compute-stored-gql-type-key-utils/compute-composite-field-input-type-key.util';
 import { computeObjectMetadataInputTypeKey } from 'src/engine/api/graphql/workspace-schema-builder/utils/compute-stored-gql-type-key-utils/compute-object-metadata-input-type.util';
 import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { isFlatFieldMetadataSupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-supported-in-group-by.util';
 import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
 import { isMorphOrRelationFlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-morph-or-relation-flat-field-metadata.util';
 import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type';
@@ -62,6 +63,10 @@ export class ObjectMetadataGroupByGqlInputTypeGenerator {
    const allGeneratedFields: GraphQLInputFieldConfigMap = {};

    for (const fieldMetadata of fields) {
+      if (!isFlatFieldMetadataSupportedInGroupBy(fieldMetadata)) {
+        continue;
+      }
+
      const generatedField = isMorphOrRelationFlatFieldMetadata(fieldMetadata)
        ? this.relationFieldMetadataGqlInputTypeGenerator.generateSimpleRelationFieldGroupByInputType(
            fieldMetadata,
@@ -1,2 +1,2 @@
 export const MCP_SERVER_INSTRUCTIONS =
-  'Twenty CRM MCP Server. Follow this workflow: (1) get_tool_catalog to discover tools, (2) learn_tools to get input schemas, (3) execute_tool to run them. Never guess tool names — always start with get_tool_catalog. Use load_skills for guidance on complex tasks like workflow or dashboard building.';
+  'Twenty CRM MCP Server. Follow this workflow: (1) get_tool_catalog to discover tools, (2) learn_tools to get input schemas, (3) execute_tool to run them. Never guess tool names — always start with get_tool_catalog. Use load_skills for guidance on complex tasks like workflow or dashboard building. For comparative/grouped analytics (by/per/top/most/least/average/total/ranking), use group_by tools, and use find tools for record retrieval; for multiple metrics, run multiple group_by calls and merge results.';
@@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common';

 import { DEFAULT_NUMBER_OF_GROUPS_LIMIT } from 'twenty-shared/constants';

+import { GroupByArgProcessorService } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/group-by-arg-processor.service';
 import { CommonGroupByQueryRunnerService } from 'src/engine/api/common/common-query-runners/common-group-by-query-runner.service';
 import { RestApiBaseHandler } from 'src/engine/api/rest/core/handlers/rest-api-base.handler';
 import { parseAggregateFieldsRestRequest } from 'src/engine/api/rest/input-request-parsers/aggregate-fields-parser-utils/parse-aggregate-fields-rest-request.util';
@@ -19,6 +20,7 @@ import { workspaceQueryRunnerRestApiExceptionHandler } from 'src/engine/api/rest
 export class RestApiGroupByHandler extends RestApiBaseHandler {
  constructor(
    private readonly commonGroupByQueryRunnerService: CommonGroupByQueryRunnerService,
+    private readonly groupByArgProcessor: GroupByArgProcessorService,
  ) {
    super();
  }
@@ -84,6 +86,17 @@ export class RestApiGroupByHandler extends RestApiBaseHandler {
    const groupBy = parseGroupByRestRequest(request);
    const includeRecords = parseIncludeRecordsSampleRestRequest(request);
    const aggregateFields = parseAggregateFieldsRestRequest(request);
+    const availableAggregations =
+      this.groupByArgProcessor.getAvailableAggregations({
+        flatObjectMetadata,
+        flatFieldMetadataMaps,
+      });
+
+    this.groupByArgProcessor.validateAggregateFieldKeysOrThrow({
+      aggregateFieldKeys: Object.keys(aggregateFields),
+      availableAggregations,
+    });
+
    const limit = parseLimitRestRequest(
      request,
      DEFAULT_NUMBER_OF_GROUPS_LIMIT,
@@ -7,6 +7,7 @@ import { CreateManyRecordsService } from 'src/engine/core-modules/record-crud/se
 import { CreateRecordService } from 'src/engine/core-modules/record-crud/services/create-record.service';
 import { DeleteRecordService } from 'src/engine/core-modules/record-crud/services/delete-record.service';
 import { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
+import { GroupByRecordsService } from 'src/engine/core-modules/record-crud/services/group-by-records.service';
 import { UpdateManyRecordsService } from 'src/engine/core-modules/record-crud/services/update-many-records.service';
 import { UpdateRecordService } from 'src/engine/core-modules/record-crud/services/update-record.service';
 import { UpsertRecordService } from 'src/engine/core-modules/record-crud/services/upsert-record.service';
@@ -30,6 +31,7 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache
    UpdateManyRecordsService,
    DeleteRecordService,
    FindRecordsService,
+    GroupByRecordsService,
    UpsertRecordService,
  ],
  exports: [
@@ -39,6 +41,7 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache
    UpdateManyRecordsService,
    DeleteRecordService,
    FindRecordsService,
+    GroupByRecordsService,
    UpsertRecordService,
  ],
 })
@@ -0,0 +1,194 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { QUERY_MAX_RECORDS } from 'twenty-shared/constants';
+import {
+  AggregateOperations,
+  OrderByDirection,
+  type OrderByWithGroupBy,
+} from 'twenty-shared/types';
+
+import { GroupByArgProcessorService } from 'src/engine/api/common/common-args-processors/group-by-arg-processor/group-by-arg-processor.service';
+import { CommonGroupByQueryRunnerService } from 'src/engine/api/common/common-query-runners/common-group-by-query-runner.service';
+import { CommonQueryRunnerException } from 'src/engine/api/common/common-query-runners/errors/common-query-runner.exception';
+import {
+  RecordCrudException,
+  RecordCrudExceptionCode,
+} from 'src/engine/core-modules/record-crud/exceptions/record-crud.exception';
+import { CommonApiContextBuilderService } from 'src/engine/core-modules/record-crud/services/common-api-context-builder.service';
+import { type GroupByRecordsParams } from 'src/engine/core-modules/record-crud/types/group-by-records-params.type';
+import { type GroupByRecordsResult } from 'src/engine/core-modules/record-crud/types/group-by-records-result.type';
+import { type ToolOutput } from 'src/engine/core-modules/tool/types/tool-output.type';
+
+@Injectable()
+export class GroupByRecordsService {
+  private readonly logger = new Logger(GroupByRecordsService.name);
+
+  constructor(
+    private readonly commonGroupByRunner: CommonGroupByQueryRunnerService,
+    private readonly commonApiContextBuilder: CommonApiContextBuilderService,
+    private readonly groupByArgProcessor: GroupByArgProcessorService,
+  ) {}
+
+  async execute(
+    params: GroupByRecordsParams,
+  ): Promise<ToolOutput<GroupByRecordsResult>> {
+    const {
+      objectName,
+      groupBy,
+      aggregateOperation = AggregateOperations.COUNT,
+      aggregateFieldName,
+      limit,
+      orderBy = 'DESC',
+      filter,
+      authContext,
+    } = params;
+
+    try {
+      const {
+        queryRunnerContext,
+        flatObjectMetadata,
+        flatFieldMetadataMaps,
+        objectsPermissions,
+      } = await this.commonApiContextBuilder.build({
+        authContext,
+        objectName,
+      });
+
+      const availableAggregations =
+        this.groupByArgProcessor.getAvailableAggregations({
+          flatObjectMetadata,
+          flatFieldMetadataMaps,
+          restrictedFields:
+            objectsPermissions[flatObjectMetadata.id]?.restrictedFields,
+        });
+
+      let aggregateFieldKey: string;
+
+      try {
+        aggregateFieldKey =
+          this.groupByArgProcessor.resolveToolAggregateFieldKeyOrThrow({
+            aggregateOperation,
+            aggregateFieldName,
+            availableAggregations,
+          });
+      } catch (error) {
+        if (error instanceof CommonQueryRunnerException) {
+          throw new RecordCrudException(
+            error.message,
+            RecordCrudExceptionCode.INVALID_REQUEST,
+          );
+        }
+
+        throw error;
+      }
+
+      const selectedFields = {
+        [aggregateFieldKey]: true,
+        groupByDimensionValues: true,
+      };
+
+      const mappedOrderBy: OrderByWithGroupBy = [
+        {
+          aggregate: {
+            [aggregateFieldKey]:
+              orderBy === 'ASC'
+                ? OrderByDirection.AscNullsLast
+                : OrderByDirection.DescNullsLast,
+          },
+        },
+      ];
+
+      const clampedLimit = limit
+        ? Math.min(limit, QUERY_MAX_RECORDS)
+        : QUERY_MAX_RECORDS;
+
+      const { results } = await this.commonGroupByRunner.execute(
+        {
+          filter: filter ?? {},
+          groupBy,
+          orderBy: mappedOrderBy,
+          selectedFields,
+          limit: clampedLimit,
+        },
+        queryRunnerContext,
+      );
+
+      const dimensionLabels = groupBy.map((entry) =>
+        this.getDimensionLabelFromGroupByEntry(entry),
+      );
+
+      this.logger.log(
+        `Grouped ${objectName} by ${dimensionLabels.join(', ')}: ${results.length} groups`,
+      );
+
+      return {
+        success: true,
+        message: `Grouped ${objectName} by ${dimensionLabels.join(', ')}: ${results.length} groups`,
+        result: {
+          groups: results.map((item) => ({
+            dimensions: item.groupByDimensionValues,
+            value: item[aggregateFieldKey],
+          })),
+          dimensionLabels,
+          aggregation: aggregateOperation,
+          groupCount: results.length,
+        },
+      };
+    } catch (error) {
+      if (error instanceof RecordCrudException) {
+        return {
+          success: false,
+          message: `Failed to group ${objectName} records`,
+          error: error.message,
+        };
+      }
+
+      this.logger.error(`Failed to group records: ${error}`);
+
+      return {
+        success: false,
+        message: `Failed to group ${objectName} records`,
+        error:
+          error instanceof Error ? error.message : 'Failed to group records',
+      };
+    }
+  }
+
+  private getDimensionLabelFromGroupByEntry(
+    entry: GroupByRecordsParams['groupBy'][number],
+  ): string {
+    const fieldEntries = Object.entries(entry);
+
+    if (fieldEntries.length === 0) {
+      return '';
+    }
+
+    const [fieldName, fieldDefinition] = fieldEntries[0];
+
+    if (fieldDefinition === true) {
+      return fieldName;
+    }
+
+    if (typeof fieldDefinition !== 'object' || fieldDefinition === null) {
+      return fieldName;
+    }
+
+    const nestedEntries = Object.entries(fieldDefinition);
+
+    if (nestedEntries.length !== 1) {
+      return fieldName;
+    }
+
+    const [nestedFieldName, nestedFieldDefinition] = nestedEntries[0];
+
+    if (nestedFieldDefinition !== true) {
+      return fieldName;
+    }
+
+    if (nestedFieldName === 'id' && fieldName.endsWith('Id')) {
+      return fieldName;
+    }
+
+    return `${fieldName}.${nestedFieldName}`;
+  }
+}
@@ -0,0 +1,15 @@
+import { type AggregateOperations } from 'twenty-shared/types';
+
+import { type ObjectRecordGroupBy } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';
+
+import { type RecordCrudExecutionContext } from './record-crud-execution-context.type';
+
+export type GroupByRecordsParams = RecordCrudExecutionContext & {
+  objectName: string;
+  groupBy: ObjectRecordGroupBy;
+  aggregateOperation?: keyof typeof AggregateOperations;
+  aggregateFieldName?: string;
+  limit?: number;
+  orderBy?: 'ASC' | 'DESC';
+  filter?: Record<string, unknown>;
+};
@@ -0,0 +1,8 @@
+import { type AggregateOperations } from 'twenty-shared/types';
+
+export type GroupByRecordsResult = {
+  groups: Array<{ dimensions: string[]; value: string | number | null }>;
+  dimensionLabels: string[];
+  aggregation: keyof typeof AggregateOperations;
+  groupCount: number;
+};
@@ -0,0 +1,111 @@
+import { GraphQLFloat, GraphQLInt } from 'graphql';
+import { AggregateOperations, FieldMetadataType } from 'twenty-shared/types';
+
+import { type AggregationField } from 'src/engine/api/graphql/workspace-schema-builder/utils/get-available-aggregations-from-object-fields.util';
+import { resolveAggregateFieldKey } from 'src/engine/core-modules/record-crud/utils/resolve-aggregate-field-key.util';
+
+const availableAggregations: Record<string, AggregationField> = {
+  totalCount: {
+    type: GraphQLInt,
+    description: 'Total count',
+    fromField: '*',
+    fromFieldType: FieldMetadataType.UUID,
+    aggregateOperation: AggregateOperations.COUNT,
+  },
+  sumEmployees: {
+    type: GraphQLFloat,
+    description: 'Sum of employees',
+    fromField: 'employees',
+    fromFieldType: FieldMetadataType.NUMBER,
+    aggregateOperation: AggregateOperations.SUM,
+  },
+  avgEmployees: {
+    type: GraphQLFloat,
+    description: 'Average of employees',
+    fromField: 'employees',
+    fromFieldType: FieldMetadataType.NUMBER,
+    aggregateOperation: AggregateOperations.AVG,
+  },
+  sumAmountAmountMicros: {
+    type: GraphQLFloat,
+    description: 'Sum of amount',
+    fromField: 'amount',
+    fromFieldType: FieldMetadataType.CURRENCY,
+    fromSubFields: ['amountMicros', 'currencyCode'],
+    subFieldForNumericOperation: 'amountMicros',
+    aggregateOperation: AggregateOperations.SUM,
+  },
+  avgAmountAmountMicros: {
+    type: GraphQLFloat,
+    description: 'Average of amount',
+    fromField: 'amount',
+    fromFieldType: FieldMetadataType.CURRENCY,
+    fromSubFields: ['amountMicros', 'currencyCode'],
+    subFieldForNumericOperation: 'amountMicros',
+    aggregateOperation: AggregateOperations.AVG,
+  },
+};
+
+describe('resolveAggregateFieldKey', () => {
+  it('resolves a simple NUMBER field', () => {
+    expect(
+      resolveAggregateFieldKey('SUM', 'employees', availableAggregations),
+    ).toBe('sumEmployees');
+  });
+
+  it('resolves a CURRENCY field with dot notation', () => {
+    expect(
+      resolveAggregateFieldKey(
+        'SUM',
+        'amount.amountMicros',
+        availableAggregations,
+      ),
+    ).toBe('sumAmountAmountMicros');
+  });
+
+  it('resolves a CURRENCY field with just the parent name', () => {
+    expect(
+      resolveAggregateFieldKey('SUM', 'amount', availableAggregations),
+    ).toBe('sumAmountAmountMicros');
+  });
+
+  it('rejects an invalid sub-field for a composite type', () => {
+    expect(
+      resolveAggregateFieldKey(
+        'SUM',
+        'amount.currencyCode',
+        availableAggregations,
+      ),
+    ).toBeNull();
+  });
+
+  it('rejects invalid multi-level dot notation', () => {
+    expect(
+      resolveAggregateFieldKey(
+        'SUM',
+        'amount.amountMicros.extra',
+        availableAggregations,
+      ),
+    ).toBeNull();
+  });
+
+  it('returns null for a non-existent field', () => {
+    expect(
+      resolveAggregateFieldKey('SUM', 'nonExistent', availableAggregations),
+    ).toBeNull();
+  });
+
+  it('matches the correct operation when multiple exist for the same field', () => {
+    expect(
+      resolveAggregateFieldKey('AVG', 'employees', availableAggregations),
+    ).toBe('avgEmployees');
+
+    expect(
+      resolveAggregateFieldKey(
+        'AVG',
+        'amount.amountMicros',
+        availableAggregations,
+      ),
+    ).toBe('avgAmountAmountMicros');
+  });
+});
@@ -0,0 +1,45 @@
+import { AggregateOperations } from 'twenty-shared/types';
+
+import { type AggregationField } from 'src/engine/api/graphql/workspace-schema-builder/utils/get-available-aggregations-from-object-fields.util';
+
+export const resolveAggregateFieldKey = (
+  aggregateOperation: keyof typeof AggregateOperations,
+  aggregateFieldName: string,
+  availableAggregations: Record<string, AggregationField>,
+): string | null => {
+  // Tool inputs use (aggregateOperation, aggregateFieldName), while GraphQL/REST
+  // already pass concrete aggregate keys (e.g. "sumEmployees"), so this helper
+  // intentionally adapts only the tool-surface contract.
+  const fieldPathParts = aggregateFieldName.split('.');
+
+  if (
+    fieldPathParts.length > 2 ||
+    fieldPathParts.some((fieldPathPart) => fieldPathPart.length === 0)
+  ) {
+    return null;
+  }
+
+  const [parentField, subField] = fieldPathParts;
+
+  const targetOperation = AggregateOperations[aggregateOperation];
+
+  const matchingEntry = Object.entries(availableAggregations).find(
+    ([, aggregation]) => {
+      if (aggregation.aggregateOperation !== targetOperation) {
+        return false;
+      }
+
+      if (aggregation.fromField !== parentField) {
+        return false;
+      }
+
+      if (subField) {
+        return aggregation.subFieldForNumericOperation === subField;
+      }
+
+      return true;
+    },
+  );
+
+  return matchingEntry?.[0] ?? null;
+};
@@ -0,0 +1,265 @@
+import {
+  AggregateOperations,
+  FirstDayOfTheWeek,
+  FieldMetadataType,
+  ObjectRecordGroupByDateGranularity,
+  RelationType,
+  type RestrictedFieldsPermissions,
+} from 'twenty-shared/types';
+import { isFieldMetadataDateKind } from 'twenty-shared/utils';
+import { z } from 'zod';
+
+import { getAvailableAggregationsFromObjectFields } from 'src/engine/api/graphql/workspace-schema-builder/utils/get-available-aggregations-from-object-fields.util';
+import { type ObjectMetadataForToolSchema } from 'src/engine/core-modules/record-crud/types/object-metadata-for-tool-schema.type';
+import { resolveAggregateFieldKey } from 'src/engine/core-modules/record-crud/utils/resolve-aggregate-field-key.util';
+import { generateRecordFilterSchema } from 'src/engine/core-modules/record-crud/zod-schemas/record-filter.zod-schema';
+import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
+import { getGroupableSubFieldsForCompositeType } from 'src/engine/metadata-modules/field-metadata/utils/get-groupable-sub-fields-for-composite-type.util';
+import { isFlatFieldMetadataSupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-supported-in-group-by.util';
+import { isFieldMetadataEntityOfType } from 'src/engine/utils/is-field-metadata-of-type.util';
+
+const dateGranularityValues = Object.values(
+  ObjectRecordGroupByDateGranularity,
+).filter((v) => v !== ObjectRecordGroupByDateGranularity.NONE) as [
+  string,
+  ...string[],
+];
+
+const dateGroupBySchema = z
+  .object({
+    granularity: z
+      .enum(dateGranularityValues)
+      .default(ObjectRecordGroupByDateGranularity.MONTH)
+      .describe('Date grouping granularity. Default: MONTH.'),
+    weekStartDay: z
+      .nativeEnum(FirstDayOfTheWeek)
+      .optional()
+      .describe(
+        'First day of week (MONDAY, SUNDAY, SATURDAY). Only used when granularity is WEEK.',
+      ),
+    timeZone: z
+      .string()
+      .default('UTC')
+      .describe(
+        'IANA timezone for date groupings (e.g. "America/New_York"). Default: UTC.',
+      ),
+  })
+  .strict()
+  .describe('Date field grouping configuration');
+
+const buildGroupByEntriesAndDescriptions = (
+  objectMetadata: ObjectMetadataForToolSchema,
+  restrictedFields?: RestrictedFieldsPermissions,
+): {
+  groupByEntries: z.ZodTypeAny[];
+  fieldNameDescriptions: string[];
+} => {
+  const groupByEntries: z.ZodTypeAny[] = [];
+  const fieldNameDescriptions: string[] = [];
+
+  for (const field of objectMetadata.fields) {
+    if (restrictedFields?.[field.id]?.canRead === false) {
+      continue;
+    }
+
+    if (!isFlatFieldMetadataSupportedInGroupBy(field)) {
+      continue;
+    }
+
+    if (isFieldMetadataEntityOfType(field, FieldMetadataType.RELATION)) {
+      if (field.settings?.relationType === RelationType.MANY_TO_ONE) {
+        const relationFieldName = `${field.name}Id`;
+
+        groupByEntries.push(
+          z.object({ [relationFieldName]: z.literal(true) }).strict(),
+        );
+        fieldNameDescriptions.push(relationFieldName);
+      }
+
+      continue;
+    }
+
+    if (isFieldMetadataEntityOfType(field, FieldMetadataType.MORPH_RELATION)) {
+      continue;
+    }
+
+    if (isFieldMetadataDateKind(field.type)) {
+      groupByEntries.push(
+        z.object({ [field.name]: dateGroupBySchema }).strict(),
+      );
+      fieldNameDescriptions.push(`${field.name} (date)`);
+      continue;
+    }
+
+    if (isCompositeFieldMetadataType(field.type)) {
+      const subFields = getGroupableSubFieldsForCompositeType(field.type);
+
+      if (subFields) {
+        for (const subField of subFields) {
+          groupByEntries.push(
+            z
+              .object({
+                [field.name]: z
+                  .object({ [subField]: z.literal(true) })
+                  .strict(),
+              })
+              .strict(),
+          );
+          fieldNameDescriptions.push(`${field.name}.${subField}`);
+        }
+      }
+
+      continue;
+    }
+
+    groupByEntries.push(z.object({ [field.name]: z.literal(true) }).strict());
+    fieldNameDescriptions.push(field.name);
+  }
+
+  return { groupByEntries, fieldNameDescriptions };
+};
+
+export const hasGroupByToolInputSchema = (
+  objectMetadata: ObjectMetadataForToolSchema,
+  restrictedFields?: RestrictedFieldsPermissions,
+): boolean => {
+  return (
+    buildGroupByEntriesAndDescriptions(objectMetadata, restrictedFields)
+      .groupByEntries.length > 0
+  );
+};
+
+export const generateGroupByToolInputSchema = (
+  objectMetadata: ObjectMetadataForToolSchema,
+  restrictedFields?: RestrictedFieldsPermissions,
+): z.ZodTypeAny | null => {
+  const { groupByEntries, fieldNameDescriptions } =
+    buildGroupByEntriesAndDescriptions(objectMetadata, restrictedFields);
+
+  if (groupByEntries.length === 0) {
+    return null;
+  }
+
+  const groupByEntrySchema =
+    groupByEntries.length === 1
+      ? groupByEntries[0]
+      : z.union(
+          groupByEntries as [z.ZodTypeAny, z.ZodTypeAny, ...z.ZodTypeAny[]],
+        );
+
+  const { filterShape, filterSchema } = generateRecordFilterSchema(
+    objectMetadata,
+    restrictedFields,
+  );
+
+  const availableAggregations = getAvailableAggregationsFromObjectFields(
+    objectMetadata.fields.filter(
+      (field) => restrictedFields?.[field.id]?.canRead !== false,
+    ),
+  );
+  const availableAggregateFieldNames = Array.from(
+    new Set(
+      Object.values(availableAggregations)
+        .filter(
+          (aggregation) =>
+            aggregation.aggregateOperation !== AggregateOperations.COUNT,
+        )
+        .map((aggregation) =>
+          aggregation.subFieldForNumericOperation
+            ? `${aggregation.fromField}.${aggregation.subFieldForNumericOperation}`
+            : aggregation.fromField,
+        ),
+    ),
+  );
+
+  return z
+    .object({
+      groupBy: z
+        .array(groupByEntrySchema)
+        .min(1)
+        .max(2)
+        .describe(
+          `Fields to group by (max 2). Each entry must be an object with exactly one field key. Examples: {"status": true}, {"companyId": true}, {"createdAt": {"granularity": "MONTH", "timeZone": "UTC"}}. Available: ${fieldNameDescriptions.join(', ')}.`,
+        ),
+      aggregateOperation: z
+        .enum(Object.keys(AggregateOperations) as [string, ...string[]])
+        .default(AggregateOperations.COUNT)
+        .describe(
+          'Aggregate operation to apply per group. Default: COUNT. Any operation other than COUNT requires aggregateFieldName.',
+        ),
+      aggregateFieldName: z
+        .string()
+        .optional()
+        .describe(
+          `Field to aggregate. Required for any operation other than COUNT. Available fields: ${availableAggregateFieldNames.join(', ')}.`,
+        ),
+      limit: z
+        .number()
+        .int()
+        .positive()
+        .max(100)
+        .default(50)
+        .describe(
+          'Maximum number of groups to return (default: 50, max: 100).',
+        ),
+      orderBy: z
+        .enum(['ASC', 'DESC'])
+        .default('DESC')
+        .describe(
+          'Order groups by aggregate value. DESC (default) gives "top N" behavior.',
+        ),
+      ...filterShape,
+      or: z
+        .array(filterSchema)
+        .optional()
+        .describe('OR condition - matches if ANY of the filters match'),
+      and: z
+        .array(filterSchema)
+        .optional()
+        .describe('AND condition - matches if ALL filters match'),
+      not: filterSchema
+        .optional()
+        .describe('NOT condition - matches if the filter does NOT match'),
+    })
+    .strict()
+    .superRefine((input, context) => {
+      const aggregateOperation =
+        input.aggregateOperation as keyof typeof AggregateOperations;
+
+      if (aggregateOperation === AggregateOperations.COUNT) {
+        if (input.aggregateFieldName) {
+          context.addIssue({
+            code: z.ZodIssueCode.custom,
+            message: 'aggregateFieldName is not supported for COUNT operation.',
+            path: ['aggregateFieldName'],
+          });
+        }
+
+        return;
+      }
+
+      if (!input.aggregateFieldName) {
+        context.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: `aggregateFieldName is required for ${aggregateOperation} operation.`,
+          path: ['aggregateFieldName'],
+        });
+
+        return;
+      }
+
+      const resolvedAggregateFieldKey = resolveAggregateFieldKey(
+        aggregateOperation,
+        input.aggregateFieldName,
+        availableAggregations,
+      );
+
+      if (!resolvedAggregateFieldKey) {
+        context.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: `No aggregation available for ${aggregateOperation} on field "${input.aggregateFieldName}".`,
+          path: ['aggregateFieldName'],
+        });
+      }
+    });
+};
@@ -19,6 +19,10 @@ import { generateUpdateRecordInputSchema } from 'src/engine/core-modules/record-
 import { DeleteToolInputSchema } from 'src/engine/core-modules/record-crud/zod-schemas/delete-tool.zod-schema';
 import { FindOneToolInputSchema } from 'src/engine/core-modules/record-crud/zod-schemas/find-one-tool.zod-schema';
 import { generateFindToolInputSchema } from 'src/engine/core-modules/record-crud/zod-schemas/find-tool.zod-schema';
+import {
+  generateGroupByToolInputSchema,
+  hasGroupByToolInputSchema,
+} from 'src/engine/core-modules/record-crud/zod-schemas/group-by-tool.zod-schema';
 import { ToolCategory } from 'twenty-shared/ai';
 import { type ToolDescriptor } from 'src/engine/core-modules/tool-provider/types/tool-descriptor.type';
 import { type ToolIndexEntry } from 'src/engine/core-modules/tool-provider/types/tool-index-entry.type';
@@ -138,6 +142,33 @@ export class DatabaseToolProvider implements ToolProvider {
          icon: flatObject.icon ?? undefined,
          operation: 'find_one',
        });
+
+        const groupBySchema = includeSchemas
+          ? generateGroupByToolInputSchema(objectMetadata, restrictedFields)
+          : null;
+        const hasGroupBySchema =
+          groupBySchema !== null ||
+          hasGroupByToolInputSchema(objectMetadata, restrictedFields);
+
+        if (hasGroupBySchema) {
+          descriptors.push({
+            name: `group_by_${snakePlural}`,
+            description: `Group ${objectMetadata.labelPlural} records by one or two fields and compute an aggregate (COUNT, SUM, AVG, MIN, MAX, etc.). Use for questions like "how many deals per stage?" or "total revenue by company". Returns groups with dimension values and aggregate results, ordered by the aggregate value.`,
+            category: ToolCategory.DATABASE_CRUD,
+            ...(includeSchemas &&
+              groupBySchema && {
+                inputSchema: z.toJSONSchema(groupBySchema),
+              }),
+            executionRef: {
+              kind: 'database_crud',
+              objectNameSingular: objectMetadata.nameSingular,
+              operation: 'group_by',
+            },
+            objectName: objectMetadata.nameSingular,
+            icon: flatObject.icon ?? undefined,
+            operation: 'group_by',
+          });
+        }
      }

      if (permission.canUpdateObjectRecords) {
@@ -1,9 +1,12 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';

+import { type AggregateOperations } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 import { Repository } from 'typeorm';

+import { type ObjectRecordGroupBy } from 'src/engine/api/graphql/workspace-query-builder/interfaces/object-record.interface';
+
 import { type FlatWorkspace } from 'src/engine/core-modules/workspace/types/flat-workspace.type';
 import { fromUserEntityToFlat } from 'src/engine/core-modules/user/utils/from-user-entity-to-flat.util';
 import { type ToolProviderContext } from 'src/engine/core-modules/tool-provider/interfaces/tool-provider-context.type';
@@ -19,6 +22,7 @@ import { CreateManyRecordsService } from 'src/engine/core-modules/record-crud/se
 import { CreateRecordService } from 'src/engine/core-modules/record-crud/services/create-record.service';
 import { DeleteRecordService } from 'src/engine/core-modules/record-crud/services/delete-record.service';
 import { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
+import { GroupByRecordsService } from 'src/engine/core-modules/record-crud/services/group-by-records.service';
 import { type FindRecordsParams } from 'src/engine/core-modules/record-crud/types/find-records-params.type';
 import { UpdateManyRecordsService } from 'src/engine/core-modules/record-crud/services/update-many-records.service';
 import { UpdateRecordService } from 'src/engine/core-modules/record-crud/services/update-record.service';
@@ -48,6 +52,7 @@ export class ToolExecutorService {

  constructor(
    private readonly findRecordsService: FindRecordsService,
+    private readonly groupByRecordsService: GroupByRecordsService,
    private readonly createRecordService: CreateRecordService,
    private readonly createManyRecordsService: CreateManyRecordsService,
    private readonly updateRecordService: UpdateRecordService,
@@ -181,6 +186,31 @@ export class ToolExecutorService {
          rolePermissionConfig: context.rolePermissionConfig,
          soft: true,
        });
+
+      case 'group_by': {
+        const {
+          groupBy,
+          aggregateOperation,
+          aggregateFieldName,
+          limit: groupByLimit,
+          orderBy: groupByOrderBy,
+          ...groupByFilter
+        } = args;
+
+        return this.groupByRecordsService.execute({
+          objectName: ref.objectNameSingular,
+          groupBy: groupBy as ObjectRecordGroupBy,
+          aggregateOperation: aggregateOperation as
+            | keyof typeof AggregateOperations
+            | undefined,
+          aggregateFieldName: aggregateFieldName as string | undefined,
+          limit: groupByLimit as number | undefined,
+          orderBy: groupByOrderBy as 'ASC' | 'DESC' | undefined,
+          filter: groupByFilter,
+          authContext,
+          rolePermissionConfig: context.rolePermissionConfig,
+        });
+      }
    }
  }

@@ -5,4 +5,5 @@ export type DatabaseCrudOperation =
  | 'create_many'
  | 'update'
  | 'update_many'
-  | 'delete';
+  | 'delete'
+  | 'group_by';
@@ -33,6 +33,7 @@ For simple CRUD operations (find/create/update/delete a record), you do NOT need
 - NEVER guess or construct API URLs — always use the appropriate database tool
 - The \`http_request\` tool is ONLY for external third-party APIs (not for Twenty's own data)
 - If you need to look up a record, learn and execute the corresponding find_one_* or find_many_* tool
+- For comparative/grouped analytics questions (by/per/top/most/least/average/total/ranking), use \`group_by_*\` instead of \`find_*\`; if multiple metrics are needed, run multiple \`group_by_*\` calls with the same dimensions and merge results.

 ## Data Efficiency

@@ -0,0 +1,140 @@
+import { type CompositeProperty, FieldMetadataType } from 'twenty-shared/types';
+
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { getGroupableSubFieldsForCompositeType } from 'src/engine/metadata-modules/field-metadata/utils/get-groupable-sub-fields-for-composite-type.util';
+import { isCompositePropertySupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-property-supported-in-group-by.util';
+import { isFlatFieldMetadataSupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-supported-in-group-by.util';
+
+const buildFlatFieldMetadata = (
+  type: FieldMetadataType,
+  name = 'field',
+  isSystem = false,
+): FlatFieldMetadata => ({ type, name, isSystem }) as FlatFieldMetadata;
+
+const buildCompositeProperty = (
+  type: FieldMetadataType,
+  hidden: CompositeProperty['hidden'] = false,
+): CompositeProperty => ({
+  name: 'subField',
+  type,
+  hidden,
+  isRequired: false,
+});
+
+describe('isFlatFieldMetadataSupportedInGroupBy', () => {
+  it('returns false for low-level field types', () => {
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.TS_VECTOR),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.RAW_JSON),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.FILES),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.POSITION),
+      ),
+    ).toBe(false);
+  });
+
+  it('returns true for regular field types', () => {
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.TEXT),
+      ),
+    ).toBe(true);
+  });
+
+  it('returns false for internal/system field names', () => {
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.TEXT, 'id'),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.DATE_TIME, 'deletedAt'),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.TS_VECTOR, 'searchVector'),
+      ),
+    ).toBe(false);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.ACTOR, 'createdBy'),
+      ),
+    ).toBe(false);
+  });
+
+  it('returns true for createdAt and updatedAt date fields even if system', () => {
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.DATE_TIME, 'createdAt', true),
+      ),
+    ).toBe(true);
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(FieldMetadataType.DATE_TIME, 'updatedAt', true),
+      ),
+    ).toBe(true);
+  });
+
+  it('returns false for other system fields', () => {
+    expect(
+      isFlatFieldMetadataSupportedInGroupBy(
+        buildFlatFieldMetadata(
+          FieldMetadataType.TEXT,
+          'customSystemField',
+          true,
+        ),
+      ),
+    ).toBe(false);
+  });
+});
+
+describe('isCompositePropertySupportedInGroupBy', () => {
+  it('returns false for hidden or raw_json composite properties', () => {
+    expect(
+      isCompositePropertySupportedInGroupBy(
+        buildCompositeProperty(FieldMetadataType.TEXT, true),
+      ),
+    ).toBe(false);
+    expect(
+      isCompositePropertySupportedInGroupBy(
+        buildCompositeProperty(FieldMetadataType.RAW_JSON),
+      ),
+    ).toBe(false);
+  });
+
+  it('returns true for visible non-raw_json composite properties', () => {
+    expect(
+      isCompositePropertySupportedInGroupBy(
+        buildCompositeProperty(FieldMetadataType.TEXT),
+      ),
+    ).toBe(true);
+  });
+});
+
+describe('getGroupableSubFieldsForCompositeType', () => {
+  it('returns null for non-composite field types', () => {
+    expect(getGroupableSubFieldsForCompositeType(FieldMetadataType.TEXT)).toBe(
+      null,
+    );
+  });
+
+  it('returns supported subfields for composite field types', () => {
+    expect(
+      getGroupableSubFieldsForCompositeType(FieldMetadataType.CURRENCY),
+    ).toEqual(expect.arrayContaining(['amountMicros', 'currencyCode']));
+  });
+});
@@ -0,0 +1,20 @@
+import {
+  compositeTypeDefinitions,
+  FieldMetadataType,
+} from 'twenty-shared/types';
+
+import { isCompositePropertySupportedInGroupBy } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-property-supported-in-group-by.util';
+
+export const getGroupableSubFieldsForCompositeType = (
+  type: FieldMetadataType,
+): string[] | null => {
+  const compositeTypeDefinition = compositeTypeDefinitions.get(type);
+
+  if (!compositeTypeDefinition) {
+    return null;
+  }
+
+  return compositeTypeDefinition.properties
+    .filter(isCompositePropertySupportedInGroupBy)
+    .map((property) => property.name);
+};
@@ -0,0 +1,9 @@
+import { type CompositeProperty, FieldMetadataType } from 'twenty-shared/types';
+
+export const isCompositePropertySupportedInGroupBy = (
+  property: CompositeProperty,
+): boolean => {
+  return (
+    property.hidden !== true && property.type !== FieldMetadataType.RAW_JSON
+  );
+};
@@ -0,0 +1,34 @@
+import { FieldMetadataType } from 'twenty-shared/types';
+import { isFieldMetadataDateKind } from 'twenty-shared/utils';
+
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { shouldExcludeFieldFromAgentToolSchema } from 'src/engine/metadata-modules/field-metadata/utils/should-exclude-field-from-agent-tool-schema.util';
+
+const NON_GROUPABLE_FIELD_TYPES = new Set<FieldMetadataType>([
+  FieldMetadataType.TS_VECTOR,
+  FieldMetadataType.RAW_JSON,
+  FieldMetadataType.FILES,
+  FieldMetadataType.POSITION,
+]);
+
+const ALWAYS_GROUPABLE_SYSTEM_DATE_FIELD_NAMES = new Set([
+  'createdAt',
+  'updatedAt',
+]);
+
+export const isFlatFieldMetadataSupportedInGroupBy = (
+  fieldMetadata: FlatFieldMetadata,
+): boolean => {
+  const isAlwaysGroupableSystemDateField =
+    ALWAYS_GROUPABLE_SYSTEM_DATE_FIELD_NAMES.has(fieldMetadata.name) &&
+    isFieldMetadataDateKind(fieldMetadata.type);
+
+  if (
+    !isAlwaysGroupableSystemDateField &&
+    shouldExcludeFieldFromAgentToolSchema(fieldMetadata)
+  ) {
+    return false;
+  }
+
+  return !NON_GROUPABLE_FIELD_TYPES.has(fieldMetadata.type);
+};