From fa6d1394af95e7f541aa0f071212deec4921a03b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?F=C3=A9lix=20Malfait?= <felix.malfait@gmail.com>
Date: Sun, 21 Jun 2026 22:16:26 +0200
Subject: [PATCH] feat(workflow): Pick Record round robin strategy (2/3)
 (#21900)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## Overview

Second PR in the Pick Record stack. Adds a **Round Robin** selection
strategy alongside Random, so an assignment workflow can distribute
records *evenly* across a candidate pool (e.g. rotate company ownership
across a set of workspace members) rather than just randomly.

**Stacked on #21899** — review/merge that one first. This PR's diff
against `main` includes PR 1's commits until #21899 merges.

## What changed

- Widened the `strategy` enum (`RANDOM` → `RANDOM | ROUND_ROBIN`) in the
shared schema and the server input type.
- Editor now shows a **Strategy** selector (Random / Round robin). The
candidate-pool label changed from "Pick at random from" to the neutral
"Pick from" since random is no longer the only mode.
- Executor implements round robin.

## Design decisions & tradeoffs

1. **State store: Redis `incrBy` (atomic), keyed
`pick-record:round-robin:{workspaceId}:{stepId}`.** Round robin needs a
persistent cursor, and workflow runs are **not** serialized — two runs
can execute the same step concurrently — so the increment must be
atomic. `CacheStorageService.incrBy` (workflow cache namespace) is a
single atomic Redis op, needs no schema change, and is already
injectable. Index = `(cursor - 1) % poolSize`.

**Tradeoff — durability:** a Redis flush/eviction resets the cursor,
which restarts the cycle from an offset. That causes a one-time
*fairness drift*, never a *correctness* bug (no double-assignment, since
each increment is atomic). If strict durability is ever required, the
cursor can move to a Postgres counter table with `INSERT … ON CONFLICT …
DO UPDATE SET cursor = cursor + 1 RETURNING cursor` (atomic + durable) —
deliberately **not** done here to avoid a migration for what is, in
practice, an acceptable reset.

2. **Deterministic pool ordering.** The resolved pool is sorted by `id`
before the cursor is applied, so position→record mapping is stable
run-to-run regardless of fetch order. Without this, round robin wouldn't
reliably cycle.

3. **Cursor key uses `stepId`.** Stable across runs of a published
version. Republishing a version may mint new step ids, which resets the
cursor — acceptable and documented here.

4. **Slot-on-increment.** The cursor increments when the step runs
(reserving a position); if a later step in the run fails, that position
is effectively skipped. Minor, acceptable unfairness — flagged rather
than adding cross-step compensation.

## Testing

Added `pick-record-round-robin-workflow.integration-spec.ts`: builds a
workflow with a 3-record pool and `ROUND_ROBIN`, runs it 4 times
sequentially, and asserts the picks are exactly `[p0, p1, p2, p0]` (full
cycle + wraparound) against the deterministically-ordered pool. Passes
locally alongside PR 1's random test (2 suites / 3 tests). `typecheck` +
`lint:diff-with-main` green for shared/server/front.

## Follow-up

- PR 3: `LOAD_BALANCED` (fewest related records wins).

https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8

---
_Generated by [Claude
Code](https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8)_

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/21900?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->
---
 .../WorkflowEditActionPickRecord.tsx          |  41 ++-
 .../pick-record.workflow-action.ts            |  54 +++-
 .../workflow-pick-record-action-input.type.ts |   2 +-
 ...d-round-robin-workflow.integration-spec.ts | 243 ++++++++++++++++++
 .../pick-record-action-settings-schema.ts     |   5 +-
 5 files changed, 336 insertions(+), 9 deletions(-)
 create mode 100644 packages/twenty-server/test/integration/graphql/suites/workflow/pick-record-round-robin-workflow.integration-spec.ts

diff --git a/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/pick-record-action/components/WorkflowEditActionPickRecord.tsx b/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/pick-record-action/components/WorkflowEditActionPickRecord.tsx
index c60ab8c040..8036810155 100644
--- a/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/pick-record-action/components/WorkflowEditActionPickRecord.tsx
+++ b/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/pick-record-action/components/WorkflowEditActionPickRecord.tsx
@@ -4,12 +4,14 @@ import { msg } from '@lingui/core/macro';
 import { useLingui } from '@lingui/react/macro';
 import { useEffect, useState } from 'react';
 import { isDefined } from 'twenty-shared/utils';
+import { type SelectOption } from 'twenty-ui/input';
 import { HorizontalSeparator } from 'twenty-ui/layout';
 import { useDebouncedCallback } from 'use-debounce';
 
 import { useFilteredObjectMetadataItems } from '@/object-metadata/hooks/useFilteredObjectMetadataItems';
 import { useObjectMetadataSelectHelpers } from '@/object-metadata/hooks/useObjectMetadataSelectHelpers';
 import { FormMultiRecordPicker } from '@/object-record/record-field/ui/form-types/components/FormMultiRecordPicker';
+import { Select } from '@/ui/input/components/Select';
 import { SelectControl } from '@/ui/input/components/SelectControl';
 import { Dropdown } from '@/ui/layout/dropdown/components/Dropdown';
 import { useCloseDropdown } from '@/ui/layout/dropdown/hooks/useCloseDropdown';
@@ -45,8 +47,12 @@ type WorkflowEditActionPickRecordProps = {
       };
 };
 
+type PickRecordStrategy =
+  WorkflowPickRecordAction['settings']['input']['strategy'];
+
 type PickRecordFormData = {
   objectNameSingular: string;
+  strategy: PickRecordStrategy;
   recordIds: string[];
 };
 
@@ -66,11 +72,17 @@ export const WorkflowEditActionPickRecord = ({
 
   const [formData, setFormData] = useState<PickRecordFormData>(() => ({
     objectNameSingular: action.settings.input.objectName,
+    strategy: action.settings.input.strategy,
     recordIds: action.settings.input.recordIds,
   }));
 
   const isFormDisabled = actionOptions.readonly ?? false;
 
+  const strategyOptions: SelectOption<PickRecordStrategy>[] = [
+    { label: t`Random`, value: 'RANDOM' },
+    { label: t`Round robin`, value: 'ROUND_ROBIN' },
+  ];
+
   const selectedObjectMetadataItem = objectMetadataItems.find(
     (item) => item.nameSingular === formData.objectNameSingular,
   );
@@ -94,7 +106,7 @@ export const WorkflowEditActionPickRecord = ({
           ...action.settings,
           input: {
             objectName: updatedFormData.objectNameSingular,
-            strategy: action.settings.input.strategy,
+            strategy: updatedFormData.strategy,
             recordIds: updatedFormData.recordIds,
           },
         },
@@ -115,6 +127,7 @@ export const WorkflowEditActionPickRecord = ({
     }
 
     const newFormData: PickRecordFormData = {
+      ...formData,
       objectNameSingular: value,
       recordIds: [],
     };
@@ -124,6 +137,20 @@ export const WorkflowEditActionPickRecord = ({
     closeDropdown(dropdownId);
   };
 
+  const handleStrategyChange = (strategy: PickRecordStrategy) => {
+    if (isFormDisabled === true) {
+      return;
+    }
+
+    const newFormData: PickRecordFormData = {
+      ...formData,
+      strategy,
+    };
+
+    setFormData(newFormData);
+    saveAction(newFormData);
+  };
+
   const handleRecordIdsChange = (recordIds: string[]) => {
     if (isFormDisabled === true) {
       return;
@@ -163,12 +190,22 @@ export const WorkflowEditActionPickRecord = ({
           />
         </StyledObjectSelectContainer>
 
+        <Select
+          dropdownId={`workflow-edit-action-pick-record-strategy-${action.id}`}
+          label={t`Strategy`}
+          fullWidth
+          disabled={isFormDisabled}
+          value={formData.strategy}
+          options={strategyOptions}
+          onChange={handleStrategyChange}
+        />
+
         <HorizontalSeparator noMargin />
 
         {isDefined(selectedObjectMetadataItem) && (
           <FormMultiRecordPicker
             key={selectedObjectMetadataItem.nameSingular}
-            label={t`Pick at random from`}
+            label={t`Pick from`}
             objectNameSingular={selectedObjectMetadataItem.nameSingular}
             defaultValue={formData.recordIds}
             onChange={(value) =>
diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/pick-record.workflow-action.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/pick-record.workflow-action.ts
index 04566175ca..a9385e19d1 100644
--- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/pick-record.workflow-action.ts
+++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/pick-record.workflow-action.ts
@@ -6,6 +6,9 @@ import { resolveInput } from 'twenty-shared/utils';
 
 import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';
 
+import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
 import {
   WorkflowStepExecutorException,
@@ -16,13 +19,20 @@ import { type WorkflowActionInput } from 'src/modules/workflow/workflow-executor
 import { type WorkflowActionOutput } from 'src/modules/workflow/workflow-executor/types/workflow-action-output.type';
 import { findStepOrThrow } from 'src/modules/workflow/workflow-executor/utils/find-step-or-throw.util';
 import { isWorkflowPickRecordAction } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/guards/is-workflow-pick-record-action.guard';
-import { type WorkflowPickRecordActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type';
+import {
+  type WorkflowPickRecordActionInput,
+  type WorkflowPickRecordStrategy,
+} from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type';
+
+const ROUND_ROBIN_CURSOR_TTL_MS = 1000 * 60 * 60 * 24 * 90;
 
 @Injectable()
 export class PickRecordWorkflowAction implements WorkflowAction {
   constructor(
     private readonly findRecordsService: FindRecordsService,
     private readonly workflowExecutionContextService: WorkflowExecutionContextService,
+    @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
+    private readonly cacheStorageService: CacheStorageService,
   ) {}
 
   async execute({
@@ -43,7 +53,7 @@ export class PickRecordWorkflowAction implements WorkflowAction {
       );
     }
 
-    const { objectName, recordIds } = resolveInput(
+    const { objectName, strategy, recordIds } = resolveInput(
       step.settings.input,
       context,
     ) as WorkflowPickRecordActionInput;
@@ -80,9 +90,43 @@ export class PickRecordWorkflowAction implements WorkflowAction {
       };
     }
 
-    const pickedRecord =
-      candidateRecords[Math.floor(Math.random() * candidateRecords.length)];
+    const orderedRecords = [...candidateRecords].sort((recordA, recordB) =>
+      String(recordA.id).localeCompare(String(recordB.id)),
+    );
 
-    return { result: pickedRecord };
+    const pickedIndex = await this.getPickedIndex({
+      strategy,
+      candidateCount: orderedRecords.length,
+      workspaceId: runInfo.workspaceId,
+      stepId: currentStepId,
+    });
+
+    return { result: orderedRecords[pickedIndex] };
+  }
+
+  private async getPickedIndex({
+    strategy,
+    candidateCount,
+    workspaceId,
+    stepId,
+  }: {
+    strategy: WorkflowPickRecordStrategy;
+    candidateCount: number;
+    workspaceId: string;
+    stepId: string;
+  }): Promise<number> {
+    if (strategy === 'ROUND_ROBIN') {
+      const cursorKey = `pick-record:round-robin:${workspaceId}:${stepId}`;
+      const nextCursor = await this.cacheStorageService.incrBy(cursorKey, 1);
+
+      await this.cacheStorageService.expire(
+        cursorKey,
+        ROUND_ROBIN_CURSOR_TTL_MS,
+      );
+
+      return (nextCursor - 1) % candidateCount;
+    }
+
+    return Math.floor(Math.random() * candidateCount);
   }
 }
diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type.ts
index 615144b8da..ebd6f26170 100644
--- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type.ts
+++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type.ts
@@ -1,4 +1,4 @@
-export type WorkflowPickRecordStrategy = 'RANDOM';
+export type WorkflowPickRecordStrategy = 'RANDOM' | 'ROUND_ROBIN';
 
 export type WorkflowPickRecordActionInput = {
   objectName: string;
diff --git a/packages/twenty-server/test/integration/graphql/suites/workflow/pick-record-round-robin-workflow.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/workflow/pick-record-round-robin-workflow.integration-spec.ts
new file mode 100644
index 0000000000..8ba2e924a3
--- /dev/null
+++ b/packages/twenty-server/test/integration/graphql/suites/workflow/pick-record-round-robin-workflow.integration-spec.ts
@@ -0,0 +1,243 @@
+import request from 'supertest';
+import {
+  destroyWorkflowRun,
+  runWorkflowVersion,
+  waitForWorkflowCompletion,
+} from 'test/integration/graphql/suites/workflow/utils/workflow-run-test.util';
+
+const client = request(`http://localhost:${APP_PORT}`);
+
+describe('Pick Record Workflow - round robin (e2e)', () => {
+  let createdWorkflowId: string | null = null;
+  let createdWorkflowVersionId: string | null = null;
+  let pickRecordStepId: string | null = null;
+  let orderedCandidateRecordIds: string[] = [];
+
+  beforeAll(async () => {
+    const createWorkflowResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation CreateWorkflow {
+            createWorkflow(data: { name: "Pick Record Round Robin Test" }) {
+              id
+            }
+          }
+        `,
+      });
+
+    createdWorkflowId = createWorkflowResponse.body.data.createWorkflow.id;
+
+    const getWorkflowResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query GetWorkflow($id: UUID!) {
+            workflow(filter: { id: { eq: $id } }) {
+              versions {
+                edges {
+                  node {
+                    id
+                  }
+                }
+              }
+            }
+          }
+        `,
+        variables: { id: createdWorkflowId },
+      });
+
+    createdWorkflowVersionId =
+      getWorkflowResponse.body.data.workflow.versions.edges[0].node.id;
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation UpdateWorkflowVersion($id: UUID!, $data: WorkflowVersionUpdateInput!) {
+            updateWorkflowVersion(id: $id, data: $data) {
+              id
+            }
+          }
+        `,
+        variables: {
+          id: createdWorkflowVersionId,
+          data: {
+            trigger: {
+              name: 'Manual Trigger',
+              type: 'MANUAL',
+              settings: { outputSchema: {} },
+              nextStepIds: [],
+              position: { x: 0, y: 0 },
+            },
+          },
+        },
+      });
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation CreateWorkflowVersionStep($input: CreateWorkflowVersionStepInput!) {
+            createWorkflowVersionStep(input: $input) {
+              stepsDiff
+            }
+          }
+        `,
+        variables: {
+          input: {
+            workflowVersionId: createdWorkflowVersionId,
+            stepType: 'PICK_RECORD',
+            parentStepId: 'trigger',
+            position: { x: 200, y: 0 },
+          },
+        },
+      });
+
+    const getStepsResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query GetWorkflowVersion($id: UUID!) {
+            workflowVersion(filter: { id: { eq: $id } }) {
+              steps
+            }
+          }
+        `,
+        variables: { id: createdWorkflowVersionId },
+      });
+
+    const pickRecordStep =
+      getStepsResponse.body.data.workflowVersion.steps.find(
+        (step: { type: string }) => step.type === 'PICK_RECORD',
+      );
+
+    pickRecordStepId = pickRecordStep.id;
+
+    const companiesResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query Companies {
+            companies(first: 3) {
+              edges {
+                node {
+                  id
+                }
+              }
+            }
+          }
+        `,
+      });
+
+    const candidateRecordIds = companiesResponse.body.data.companies.edges.map(
+      (edge: { node: { id: string } }) => edge.node.id,
+    );
+
+    expect(candidateRecordIds.length).toBe(3);
+
+    // The executor sorts the resolved pool deterministically by id before
+    // applying the round-robin cursor, so the expected cycle order is the
+    // pool sorted the same way.
+    orderedCandidateRecordIds = [...candidateRecordIds].sort(
+      (idA: string, idB: string) => idA.localeCompare(idB),
+    );
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation UpdateWorkflowVersionStep($input: UpdateWorkflowVersionStepInput!) {
+            updateWorkflowVersionStep(input: $input) {
+              id
+            }
+          }
+        `,
+        variables: {
+          input: {
+            workflowVersionId: createdWorkflowVersionId,
+            step: {
+              ...pickRecordStep,
+              settings: {
+                ...pickRecordStep.settings,
+                input: {
+                  objectName: 'company',
+                  strategy: 'ROUND_ROBIN',
+                  recordIds: candidateRecordIds,
+                },
+              },
+            },
+          },
+        },
+      });
+
+    const activateResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation ActivateWorkflowVersion($workflowVersionId: UUID!) {
+            activateWorkflowVersion(workflowVersionId: $workflowVersionId)
+          }
+        `,
+        variables: { workflowVersionId: createdWorkflowVersionId },
+      });
+
+    expect(activateResponse.body.data.activateWorkflowVersion).toBe(true);
+  });
+
+  afterAll(async () => {
+    if (createdWorkflowId) {
+      await client
+        .post('/graphql')
+        .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+        .send({
+          query: `
+            mutation DestroyWorkflow($id: ID!) {
+              destroyWorkflow(id: $id) {
+                id
+              }
+            }
+          `,
+          variables: { id: createdWorkflowId },
+        });
+    }
+  });
+
+  it('cycles through the pool in order on consecutive runs', async () => {
+    const pickedRecordIds: string[] = [];
+
+    for (let runIndex = 0; runIndex < 4; runIndex++) {
+      const workflowRunId = await runWorkflowVersion({
+        workflowVersionId: createdWorkflowVersionId!,
+        payload: {},
+      });
+
+      const workflowRun = await waitForWorkflowCompletion(workflowRunId);
+
+      expect(workflowRun?.status).toBe('COMPLETED');
+
+      const result = workflowRun?.state?.stepInfos?.[pickRecordStepId!]
+        ?.result as { id: string } | undefined;
+
+      expect(result?.id).toBeDefined();
+      pickedRecordIds.push(result!.id);
+
+      await destroyWorkflowRun(workflowRunId);
+    }
+
+    expect(pickedRecordIds).toEqual([
+      orderedCandidateRecordIds[0],
+      orderedCandidateRecordIds[1],
+      orderedCandidateRecordIds[2],
+      orderedCandidateRecordIds[0],
+    ]);
+  });
+});
diff --git a/packages/twenty-shared/src/workflow/schemas/pick-record-action-settings-schema.ts b/packages/twenty-shared/src/workflow/schemas/pick-record-action-settings-schema.ts
index 7a55874c1e..3e848d00f5 100644
--- a/packages/twenty-shared/src/workflow/schemas/pick-record-action-settings-schema.ts
+++ b/packages/twenty-shared/src/workflow/schemas/pick-record-action-settings-schema.ts
@@ -1,7 +1,10 @@
 import { z } from 'zod';
 import { baseWorkflowActionSettingsSchema } from './base-workflow-action-settings-schema';
 
-export const workflowPickRecordStrategySchema = z.enum(['RANDOM']);
+export const workflowPickRecordStrategySchema = z.enum([
+  'RANDOM',
+  'ROUND_ROBIN',
+]);
 
 export const workflowPickRecordActionSettingsSchema =
   baseWorkflowActionSettingsSchema.extend({