feat(workflow): add Pick Record action (1/3 — random selection) (#21899)

## Overview Adds a new workflow action, **Pick Record**, that selects **one** record from a configured candidate pool and exposes the chosen record as the step's output. Downstream steps can then reference it through the normal variable picker — e.g. assign an owner in an _Update Record_ step by setting **Account Owner = `{{step.<pickRecordId>.id}}`**. This is the foundation for building **assignment workflows** (round-robin / load-balanced owner assignment, reviewer rotation, etc.) in Twenty. ## This is PR 1 of a 3-PR stack | PR | Strategy | Adds | |----|----------|------| | **1 (this one)** | `RANDOM` | The whole `PICK_RECORD` action, end-to-end, stateless | | 2 | `ROUND_ROBIN` | A persistent, atomically-incremented per-step cursor + the strategy selector UI | | 3 | `LOAD_BALANCED` | "fewest related records wins" via an aggregate count | Each PR widens the `strategy` enum (a backward-compatible change), so no data migration is needed between them. ## How it works - **Editor**: pick an Object, then pick the candidate records (a multi-record selector). A random record is selected from that pool at run time. - **Output**: a single record of the chosen object — the same output shape as `CREATE_RECORD`/`UPDATE_RECORD` — so it drills into `{{step.x.id}}`, `{{step.x.name}}`, … in the variable picker. - **Execution**: reuses `FindRecordsService` to fetch the pool (`id IN (recordIds)`, which also transparently drops any deleted candidates), then returns one at random. ## Design decisions & tradeoffs 1. **Standalone step that outputs a variable, not an inline "random" mode on the relation field.** This mirrors Attio's round-robin block. The decisive reason is composition: the chosen record is almost always reused (assign owner **and** create a follow-up task for them **and** email them). A variable is chosen once and reused everywhere; an inline per-field value would re-roll independently in each place. It also keeps the (stateful) round-robin/load-balanced logic out of the field inputs. Tradeoff: one extra step to wire up vs. an inline control — accepted for the composability win. An inline "Assign automatically" entry point can still be layered on later as sugar that inserts this step. 2. **Co-located in the `record-crud` action module and reuses `FindRecordsService`.** Avoids duplicating module wiring (auth context, permissions, object-metadata resolution) and the data-access path. Tradeoff: "Pick" is a selection rather than a CRUD op, so the folder name is slightly broad; chose reuse + low risk over a separate module. Can be extracted if the family grows. 3. **`strategy` exists in the schema (defaulted `RANDOM`) but the selector is hidden in this PR.** A dropdown with a single option would be UX slop, and adding the field only in PR 2 would force a data backfill for any `PICK_RECORD` steps created in between. Keeping the field now (hidden) avoids both. PR 2 introduces the selector once there's a real choice. 4. **Pool is an explicit static list (`recordIds`) for v1.** Matches the most common assignment case ("rotate among these N people") and reuses the existing `FormMultiRecordPicker`. A filter-based pool (reusing the Find Records filter UI) and a list-from-a-previous-step pool are natural follow-ups, intentionally out of scope here to keep the stack focused on the three strategies. 5. **Output schema is computed on the frontend** (like `CREATE_RECORD`), derived from `input.objectName` — so it is **not** added to `PERSISTED_OUTPUT_SCHEMA_TYPES` and needs no server-side schema computation. 6. **Validation**: `PICK_RECORD` is added to object-name metadata validation (so a deleted/invalid target object is flagged) via a dedicated `OBJECT_TARGETING_ACTION_TYPES` set — deliberately **not** to `VARIABLE_CONSUMING_ACTION_TYPES`, because a static pool legitimately references no upstream variable and would otherwise raise a spurious "no variable reference" warning. 7. **Empty pool → step error** at run time (respecting the step's error-handling options) rather than a silent no-op, since an empty pool is a misconfiguration or fully-deleted set. 8. **`Math.random`** is used for selection — no cryptographic guarantee is needed for assignment fairness. ## Testing Per our testing convention (integration test over service/`.spec` tests): added `pick-record-workflow.integration-spec.ts`, which builds a workflow with a manual trigger + a `PICK_RECORD` step, configures a known two-record pool, runs it, and asserts the run completes and the picked record is **always** within the configured pool (verifying the pool filter) across repeated runs. Local verification (typecheck + lint for shared/server/front) is green; running the integration suite and attaching editor screenshots in a follow-up comment. ## Follow-ups - PR 2: `ROUND_ROBIN` + persistent atomic cursor (Redis `incrBy` vs. a Postgres counter table — tradeoff to be documented on that PR) + strategy selector. - PR 3: `LOAD_BALANCED`. - Later (not in this stack): filter-based / variable-list pools, an inline "Assign automatically" entry point on relation fields, OOO-skip / weighting. https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8 --- _Generated by [Claude Code](https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/21899?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>
2026-06-21 21:51:16 +02:00
parent f4219449db
commit 573fd00ea7
27 changed files with 672 additions and 2 deletions
@@ -0,0 +1,255 @@
+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 (e2e)', () => {
+  let createdWorkflowId: string | null = null;
+  let createdWorkflowVersionId: string | null = null;
+  let pickRecordStepId: string | null = null;
+  let candidateRecordIds: 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 Test Workflow" }) {
+              id
+            }
+          }
+        `,
+      });
+
+    expect(createWorkflowResponse.body.errors).toBeUndefined();
+    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 } }) {
+              id
+              versions {
+                edges {
+                  node {
+                    id
+                  }
+                }
+              }
+            }
+          }
+        `,
+        variables: { id: createdWorkflowId },
+      });
+
+    createdWorkflowVersionId =
+      getWorkflowResponse.body.data.workflow.versions.edges[0].node.id;
+
+    const updateTriggerResponse = 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 },
+            },
+          },
+        },
+      });
+
+    expect(updateTriggerResponse.body.errors).toBeUndefined();
+
+    const createStepResponse = 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 },
+          },
+        },
+      });
+
+    expect(createStepResponse.body.errors).toBeUndefined();
+
+    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 } }) {
+              id
+              steps
+            }
+          }
+        `,
+        variables: { id: createdWorkflowVersionId },
+      });
+
+    const steps = getStepsResponse.body.data.workflowVersion.steps;
+    const pickRecordStep = steps.find(
+      (step: { type: string }) => step.type === 'PICK_RECORD',
+    );
+
+    expect(pickRecordStep).toBeDefined();
+    pickRecordStepId = pickRecordStep.id;
+
+    const companiesResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query Companies {
+            companies(first: 2) {
+              edges {
+                node {
+                  id
+                }
+              }
+            }
+          }
+        `,
+      });
+
+    expect(companiesResponse.body.errors).toBeUndefined();
+    candidateRecordIds = companiesResponse.body.data.companies.edges.map(
+      (edge: { node: { id: string } }) => edge.node.id,
+    );
+
+    expect(candidateRecordIds.length).toBe(2);
+
+    const updateStepResponse = 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: 'RANDOM',
+                  recordIds: candidateRecordIds,
+                },
+              },
+            },
+          },
+        },
+      });
+
+    expect(updateStepResponse.body.errors).toBeUndefined();
+
+    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.errors).toBeUndefined();
+    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 },
+        });
+    }
+  });
+
+  const pickRecordOnce = async (): Promise<{ id: string }> => {
+    const workflowRunId = await runWorkflowVersion({
+      workflowVersionId: createdWorkflowVersionId!,
+      payload: {},
+    });
+
+    const workflowRun = await waitForWorkflowCompletion(workflowRunId);
+
+    expect(workflowRun?.status).toBe('COMPLETED');
+    expect(workflowRun?.state?.stepInfos?.[pickRecordStepId!]?.status).toBe(
+      'SUCCESS',
+    );
+
+    const result = workflowRun?.state?.stepInfos?.[pickRecordStepId!]
+      ?.result as { id: string } | undefined;
+
+    await destroyWorkflowRun(workflowRunId);
+
+    expect(result?.id).toBeDefined();
+
+    return result!;
+  };
+
+  it('picks a record and exposes it as the step output', async () => {
+    const pickedRecord = await pickRecordOnce();
+
+    expect(candidateRecordIds).toContain(pickedRecord.id);
+  });
+
+  it('only ever picks records from the configured pool', async () => {
+    for (let runIndex = 0; runIndex < 5; runIndex++) {
+      const pickedRecord = await pickRecordOnce();
+
+      expect(candidateRecordIds).toContain(pickedRecord.id);
+    }
+  });
+});