fix: resolve workflow form step auto-open race condition (#21053)

## Summary
- Fix intermittent failure where the Quick Lead workflow form step did
not auto-open
- Root cause: race conditions between SSE events, Apollo cache writes,
and the `runWorkflowVersion` mutation timing
- Add generic monotonicity guard in the SSE handler that drops stale
updates for all records (not just WorkflowRun)

## Changes
- **`useTriggerOptimisticEffectFromSseUpdateEvents.ts`**: Compare
incoming `updatedAt` with cached record before writing — skip if stale.
Moved `upsertRecordsInStore` after the guard so neither Apollo cache nor
Jotai store receive stale data.
- **`useRunWorkflowVersion.tsx`**: Await mutation before opening side
panel; register SSE listener eagerly before mutation
- **`useWorkflowRun.ts`**: Simplified back to plain `useFindOneRecord` +
schema parse (no extra state needed)
- **`generateWorkflowRunDiagram.ts`**: `shouldOpenStep` matches both
PENDING and RUNNING for form steps (backend RUNNING means "waiting for
user input")
- **`WorkflowRunVisualizerEffect.tsx`**: Pass `runStatus` directly
without status mapping
- **`WorkflowRunStepNodeDetail.tsx`**: Form is interactive when step is
PENDING or RUNNING
- **Deleted `latestWorkflowRunFamilyState.ts`**: No longer needed — the
generic SSE guard replaces it

## Test plan
- [x] Hard refresh, run Quick Lead workflow 10+ times — form should
always auto-open
- [x] Complete the form and verify all subsequent steps execute without
getting stuck
- [x] Verify the workflow diagram is always visible (never disappears)
- [x] Verify other record types still update correctly via SSE (e.g.
edit a person in another tab)

packages/twenty-front/src/modules/sse-db-event/hooks/useChangeQueryListenState.ts

@@ -0,0 +1,53 @@
+import { requiredQueryListenersState } from '@/sse-db-event/states/requiredQueryListenersState';
+import { useCallback } from 'react';
+import {
+  type MetadataGqlOperationSignature,
+  type RecordGqlOperationSignature,
+} from 'twenty-shared/types';
+import { useStore } from 'jotai';
+
+export const useChangeQueryListenState = () => {
+  const store = useStore();
+
+  const changeQueryIdListenState = useCallback(
+    (
+      shouldListen: boolean,
+      targetQueryId: string,
+      targetOperationSignature:
+        | RecordGqlOperationSignature
+        | MetadataGqlOperationSignature,
+    ) => {
+      const currentRequiredQueryListeners = store.get(
+        requiredQueryListenersState.atom,
+      );
+
+      const listeningForThisQueryIsActive = currentRequiredQueryListeners.some(
+        (listener) => listener.queryId === targetQueryId,
+      );
+
+      if (shouldListen === listeningForThisQueryIsActive) {
+        return;
+      }
+
+      if (shouldListen) {
+        store.set(requiredQueryListenersState.atom, [
+          ...currentRequiredQueryListeners,
+          {
+            queryId: targetQueryId,
+            operationSignature: targetOperationSignature,
+          },
+        ]);
+      } else {
+        store.set(
+          requiredQueryListenersState.atom,
+          currentRequiredQueryListeners.filter(
+            (listener) => listener.queryId !== targetQueryId,
+          ),
+        );
+      }
+    },
+    [store],
+  );
+
+  return { changeQueryIdListenState };
+};

packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts

@@ -1,10 +1,9 @@
-import { requiredQueryListenersState } from '@/sse-db-event/states/requiredQueryListenersState';
-import { useCallback, useEffect } from 'react';
+import { useChangeQueryListenState } from '@/sse-db-event/hooks/useChangeQueryListenState';
+import { useEffect } from 'react';
 import {
   type MetadataGqlOperationSignature,
   type RecordGqlOperationSignature,
 } from 'twenty-shared/types';
-import { useStore } from 'jotai';
 
 export const useListenToEventsForQuery = ({
   queryId,
@@ -15,46 +14,7 @@ export const useListenToEventsForQuery = ({
     | RecordGqlOperationSignature
     | MetadataGqlOperationSignature;
 }) => {
-  const store = useStore();
-  const changeQueryIdListenState = useCallback(
-    (
-      shouldListen: boolean,
-      targetQueryId: string,
-      targetOperationSignature:
-        | RecordGqlOperationSignature
-        | MetadataGqlOperationSignature,
-    ) => {
-      const currentRequiredQueryListeners = store.get(
-        requiredQueryListenersState.atom,
-      );
-
-      const listeningForThisQueryIsActive = currentRequiredQueryListeners.some(
-        (listener) => listener.queryId === targetQueryId,
-      );
-
-      if (shouldListen === listeningForThisQueryIsActive) {
-        return;
-      }
-
-      if (shouldListen) {
-        store.set(requiredQueryListenersState.atom, [
-          ...currentRequiredQueryListeners,
-          {
-            queryId: targetQueryId,
-            operationSignature: targetOperationSignature,
-          },
-        ]);
-      } else {
-        store.set(
-          requiredQueryListenersState.atom,
-          currentRequiredQueryListeners.filter(
-            (listener) => listener.queryId !== targetQueryId,
-          ),
-        );
-      }
-    },
-    [store],
-  );
+  const { changeQueryIdListenState } = useChangeQueryListenState();
 
   useEffect(() => {
     changeQueryIdListenState(true, queryId, operationSignature);

packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerOptimisticEffectFromSseUpdateEvents.ts

@@ -45,8 +45,6 @@ export const useTriggerOptimisticEffectFromSseUpdateEvents = () => {
           continue;
         }
 
-        upsertRecordsInStore({ partialRecords: [updatedRecord] });
-
         const computedOptimisticRecord = {
           ...computeOptimisticRecordFromInput({
             cache: apolloCoreClient.cache,
@@ -76,6 +74,15 @@ export const useTriggerOptimisticEffectFromSseUpdateEvents = () => {
           objectPermissionsByObjectMetadataId,
         });
 
+        if (
+          isDefined(cachedRecord?.updatedAt) &&
+          isDefined(updatedRecord.updatedAt) &&
+          new Date(updatedRecord.updatedAt as string).getTime() <
+            new Date(cachedRecord!.updatedAt as string).getTime()
+        ) {
+          continue;
+        }
+
         const cachedRecordWithConnection = getRecordNodeFromRecord({
           record: cachedRecord,
           objectMetadataItem,
@@ -91,6 +98,8 @@ export const useTriggerOptimisticEffectFromSseUpdateEvents = () => {
           continue;
         }
 
+        upsertRecordsInStore({ partialRecords: [updatedRecord] });
+
         updateRecordFromCache({
           objectMetadataItems,
           objectMetadataItem,

packages/twenty-front/src/modules/workflow/hooks/useRunWorkflowVersion.tsx

@@ -15,7 +15,9 @@ import { useUpsertRecordsInStore } from '@/object-record/record-store/hooks/useU
 import { recordStoreFamilyState } from '@/object-record/record-store/states/recordStoreFamilyState';
 import { computeOptimisticCreateRecordBaseRecordInput } from '@/object-record/utils/computeOptimisticCreateRecordBaseRecordInput';
 import { computeOptimisticRecordFromInput } from '@/object-record/utils/computeOptimisticRecordFromInput';
+import { useChangeQueryListenState } from '@/sse-db-event/hooks/useChangeQueryListenState';
 import { RUN_WORKFLOW_VERSION } from '@/workflow/graphql/mutations/runWorkflowVersion';
+import { getWorkflowRunSseQueryId } from '@/workflow/utils/getWorkflowRunSseQueryId';
 import { type WorkflowRun } from '@/workflow/types/Workflow';
 import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue';
 import { useCallback } from 'react';
@@ -63,6 +65,7 @@ export const useRunWorkflowVersion = () => {
     });
 
   const { openRecordInSidePanel } = useOpenRecordInSidePanel();
+  const { changeQueryIdListenState } = useChangeQueryListenState();
 
   const setRecordInStore = useCallback(
     (workflowRun: WorkflowRun) => {
@@ -137,9 +140,24 @@ export const useRunWorkflowVersion = () => {
 
     setRecordInStore(recordCreatedInCache);
 
-    await mutate({
-      variables: { input: { workflowVersionId, workflowRunId, payload } },
-    });
+    const sseQueryId = getWorkflowRunSseQueryId(workflowRunId);
+    const sseOperationSignature = {
+      objectNameSingular: CoreObjectNameSingular.WorkflowRun,
+      variables: {
+        filter: { id: { eq: workflowRunId } },
+      },
+    };
+
+    changeQueryIdListenState(true, sseQueryId, sseOperationSignature);
+
+    try {
+      await mutate({
+        variables: { input: { workflowVersionId, workflowRunId, payload } },
+      });
+    } catch (error) {
+      changeQueryIdListenState(false, sseQueryId, sseOperationSignature);
+      throw error;
+    }
 
     openRecordInSidePanel({
       objectNameSingular: CoreObjectNameSingular.WorkflowRun,

packages/twenty-front/src/modules/workflow/utils/getWorkflowRunSseQueryId.ts

@@ -0,0 +1,2 @@
+export const getWorkflowRunSseQueryId = (workflowRunId: string) =>
+  `workflow-run-${workflowRunId}`;

packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunSSESubscribeEffect.tsx

@@ -1,12 +1,13 @@
 import { CoreObjectNameSingular } from 'twenty-shared/types';
 import { useListenToEventsForQuery } from '@/sse-db-event/hooks/useListenToEventsForQuery';
+import { getWorkflowRunSseQueryId } from '@/workflow/utils/getWorkflowRunSseQueryId';
 
 export const WorkflowRunSSESubscribeEffect = ({
   workflowRunId,
 }: {
   workflowRunId: string;
 }) => {
-  const queryId = `workflow-run-${workflowRunId}`;
+  const queryId = getWorkflowRunSseQueryId(workflowRunId);
 
   useListenToEventsForQuery({
     queryId,

packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunVisualizerEffect.tsx

@@ -20,6 +20,7 @@ import { useStepsOutputSchema } from '@/workflow/workflow-variables/hooks/useSte
 import { useStore } from 'jotai';
 import { useCallback, useContext, useEffect } from 'react';
 import { isDefined } from 'twenty-shared/utils';
+
 import { useIcons } from 'twenty-ui/display';
 
 export const WorkflowRunVisualizerEffect = ({

packages/twenty-front/src/modules/workflow/workflow-diagram/utils/generateWorkflowRunDiagram.ts

@@ -25,10 +25,16 @@ const shouldOpenStep = ({
 }) => {
   const step = steps.find((step) => step.id === nodeId);
   const stepInfo = stepInfos?.[nodeId];
-  const isStepPending = isDefined(stepInfo) && stepInfo.status === 'PENDING';
   const isStepOpenable = isDefined(step) && ['FORM'].includes(step.type);
 
-  return isStepPending && isStepOpenable;
+  if (!isStepOpenable || !isDefined(stepInfo)) {
+    return false;
+  }
+
+  return (
+    stepInfo.status === StepStatus.PENDING ||
+    stepInfo.status === StepStatus.RUNNING
+  );
 };
 
 export const generateWorkflowRunDiagram = ({

packages/twenty-front/src/modules/workflow/workflow-steps/components/WorkflowRunStepNodeDetail.tsx

@@ -205,7 +205,9 @@ export const WorkflowRunStepNodeDetail = ({
               key={stepId}
               action={stepDefinition.definition}
               actionOptions={{
-                readonly: stepExecutionStatus !== 'PENDING',
+                readonly:
+                  stepExecutionStatus !== 'PENDING' &&
+                  stepExecutionStatus !== 'RUNNING',
               }}
             />
           );