fix(settings): harden async lifecycle consistency

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deploy-readiness.test.ts

@@ -10,7 +10,7 @@ vi.mock('@/stores/operation-queue/store', () => ({
       getState: () => ({
         hasOperationError: false,
         hasPendingOperations: () => false,
-        waitForWorkflowOperations: () => Promise.resolve(true),
+        waitForWorkflowOperations: () => Promise.resolve('drained'),
       }),
     }
   ),

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deploy-readiness.ts

@@ -131,8 +131,8 @@ export function useDeployReadiness(workflowId: string | null): DeployReadiness {
     const queue = useOperationQueueStore.getState()
     if (queue.hasOperationError) return false
 
-    const drained = await queue.waitForWorkflowOperations(workflowId)
-    if (!drained) return false
+    const drainResult = await queue.waitForWorkflowOperations(workflowId)
+    if (drainResult !== 'drained') return false
 
     const latestQueue = useOperationQueueStore.getState()
     const diff = useWorkflowDiffStore.getState()

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx

@@ -9,13 +9,16 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 const {
   executionStoreState,
   mockCancel,
+  mockAdoptScopedExecution,
+  mockBeginScopedExecution,
+  mockEndScopedExecution,
   mockExecute,
   mockExecuteFromBlock,
   mockFetch,
   mockHandleExecutionCancelledConsole,
   mockHandleExecutionErrorConsole,
-  mockPersistenceExecutionEnded,
-  mockPersistenceExecutionStarted,
+  mockLoadExecutionPointer,
+  mockReconnect,
   mockRequestJson,
   mockResolveStartCandidates,
   mockSelectBestTrigger,
@@ -86,13 +89,16 @@ const {
   return {
     executionStoreState,
     mockCancel: vi.fn(),
+    mockAdoptScopedExecution: vi.fn(),
+    mockBeginScopedExecution: vi.fn(() => ({})),
+    mockEndScopedExecution: vi.fn(() => true),
     mockExecute: vi.fn(),
     mockExecuteFromBlock: vi.fn(),
     mockFetch: vi.fn(),
     mockHandleExecutionCancelledConsole: vi.fn(),
     mockHandleExecutionErrorConsole: vi.fn(),
-    mockPersistenceExecutionEnded: vi.fn(),
-    mockPersistenceExecutionStarted: vi.fn(() => ({})),
+    mockLoadExecutionPointer: vi.fn(),
+    mockReconnect: vi.fn(),
     mockRequestJson: vi.fn(),
     mockResolveStartCandidates: vi.fn(),
     mockSelectBestTrigger: vi.fn(),
@@ -210,7 +216,7 @@ vi.mock('@/hooks/use-execution-stream', () => {
     useExecutionStream: () => ({
       execute: mockExecute,
       executeFromBlock: mockExecuteFromBlock,
-      reconnect: vi.fn(),
+      reconnect: mockReconnect,
       cancel: mockCancel,
       cancelExecute: vi.fn(),
       cancelReconnect: vi.fn(),
@@ -241,11 +247,12 @@ vi.mock('@/stores/execution', () => ({
 vi.mock('@/stores/terminal', () => ({
   clearExecutionPointer: vi.fn(),
   consolePersistence: {
-    executionStarted: mockPersistenceExecutionStarted,
-    executionEnded: mockPersistenceExecutionEnded,
+    adoptScopedExecution: mockAdoptScopedExecution,
+    beginScopedExecution: mockBeginScopedExecution,
+    endScopedExecution: mockEndScopedExecution,
     persist: vi.fn(),
   },
-  loadExecutionPointer: vi.fn(),
+  loadExecutionPointer: mockLoadExecutionPointer,
   saveExecutionPointer: vi.fn(),
   useTerminalConsoleStore: Object.assign(
     (selector: (state: typeof terminalStoreState) => unknown) => selector(terminalStoreState),
@@ -406,10 +413,14 @@ describe('useWorkflowExecution cancellation', () => {
 describe('useWorkflowExecution attachment uploads', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    terminalStoreState._hasHydrated = false
     executionStoreState.getWorkflowExecution.mockReturnValue(
       executionStoreState.workflowExecutions.get('workflow-1')!
     )
     executionStoreState.getCurrentExecutionId.mockReturnValue(null)
+    mockAdoptScopedExecution.mockReturnValue(undefined)
+    mockLoadExecutionPointer.mockResolvedValue(null)
+    mockReconnect.mockResolvedValue(undefined)
     mockResolveStartCandidates.mockReturnValue([])
     mockSelectBestTrigger.mockReturnValue([])
     vi.stubGlobal('fetch', mockFetch)
@@ -548,7 +559,7 @@ describe('useWorkflowExecution attachment uploads', () => {
   it('does not let an overlapping run without lifecycle ownership end the active run', async () => {
     const persistenceExecution = {}
     let resolveActiveRun: (() => void) | undefined
-    mockPersistenceExecutionStarted.mockReturnValueOnce(persistenceExecution)
+    mockBeginScopedExecution.mockReturnValueOnce(persistenceExecution)
     mockExecute.mockImplementationOnce(
       () =>
         new Promise<void>((resolve) => {
@@ -571,16 +582,49 @@ describe('useWorkflowExecution attachment uploads', () => {
       await result().handleRunWorkflow()
     })
 
-    expect(mockPersistenceExecutionStarted).toHaveBeenCalledTimes(1)
-    expect(mockPersistenceExecutionEnded).not.toHaveBeenCalled()
+    expect(mockBeginScopedExecution).toHaveBeenCalledTimes(1)
+    expect(mockEndScopedExecution).not.toHaveBeenCalled()
 
     await act(async () => {
       resolveActiveRun?.()
       await drainStream(activeRun)
     })
 
-    expect(mockPersistenceExecutionEnded).toHaveBeenCalledOnce()
-    expect(mockPersistenceExecutionEnded).toHaveBeenCalledWith(persistenceExecution)
+    expect(mockEndScopedExecution).toHaveBeenCalledOnce()
+    expect(mockEndScopedExecution).toHaveBeenCalledWith('workflow-1', persistenceExecution)
+
+    unmount()
+  })
+
+  it('adopts and finishes persistence ownership created before the hook mounted', async () => {
+    const persistenceExecution = {}
+    terminalStoreState._hasHydrated = true
+    executionStoreState.getWorkflowExecution.mockReturnValue({
+      ...executionStoreState.getWorkflowExecution(),
+      status: 'running',
+      isExecuting: true,
+      currentExecutionId: 'execution-1',
+    })
+    executionStoreState.getCurrentExecutionId.mockReturnValue('execution-1')
+    mockLoadExecutionPointer.mockResolvedValue({
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+      lastEventId: 0,
+    })
+    mockAdoptScopedExecution.mockReturnValue(persistenceExecution)
+    mockReconnect.mockImplementationOnce(async ({ callbacks }) => {
+      callbacks.onExecutionCompleted({ finalBlockLogs: [] })
+    })
+
+    const { unmount } = renderWorkflowExecutionHook()
+    await act(async () => {
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+
+    expect(mockBeginScopedExecution).not.toHaveBeenCalled()
+    expect(mockAdoptScopedExecution).toHaveBeenCalledWith('workflow-1')
+    expect(mockEndScopedExecution).toHaveBeenCalledWith('workflow-1', persistenceExecution)
 
     unmount()
   })

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts

@@ -432,22 +432,19 @@ export function useWorkflowExecution() {
   const setCurrentExecutionId = useExecutionStore((s) => s.setCurrentExecutionId)
   const getCurrentExecutionId = useExecutionStore((s) => s.getCurrentExecutionId)
   const rawSetIsExecuting = useExecutionStore((s) => s.setIsExecuting)
-  const persistenceExecutionsRef = useRef(new Map<string, ConsolePersistenceExecution>())
 
   const endPersistenceExecution = useCallback((workflowId: string) => {
-    const persistenceExecution = persistenceExecutionsRef.current.get(workflowId)
+    const persistenceExecution = consolePersistence.adoptScopedExecution(workflowId)
     if (!persistenceExecution) return
-    persistenceExecutionsRef.current.delete(workflowId)
-    consolePersistence.executionEnded(persistenceExecution)
+    consolePersistence.endScopedExecution(workflowId, persistenceExecution)
   }, [])
 
   const setIsExecuting = useCallback(
     (workflowId: string, executing: boolean): ConsolePersistenceExecution | undefined => {
       const wasExecuting = useExecutionStore.getState().getWorkflowExecution(workflowId).isExecuting
       if (executing) {
         if (!wasExecuting) {
-          const startedExecution = consolePersistence.executionStarted()
-          persistenceExecutionsRef.current.set(workflowId, startedExecution)
+          const startedExecution = consolePersistence.beginScopedExecution(workflowId)
           rawSetIsExecuting(workflowId, true)
           return startedExecution
         }
@@ -465,12 +462,11 @@ export function useWorkflowExecution() {
   const finishOwnedExecution = useCallback(
     (workflowId: string, persistenceExecution: ConsolePersistenceExecution | undefined) => {
       if (!persistenceExecution) return
-      if (persistenceExecutionsRef.current.get(workflowId) !== persistenceExecution) return
-      endPersistenceExecution(workflowId)
+      if (!consolePersistence.endScopedExecution(workflowId, persistenceExecution)) return
       clearExecutionPointer(workflowId)
       rawSetIsExecuting(workflowId, false)
     },
-    [endPersistenceExecution, rawSetIsExecuting]
+    [rawSetIsExecuting]
   )
   const setIsDebugging = useExecutionStore((s) => s.setIsDebugging)
   const setPendingBlocks = useExecutionStore((s) => s.setPendingBlocks)
@@ -2490,7 +2486,7 @@ export function useWorkflowExecution() {
           setCurrentExecutionId(reconnectWorkflowId, capturedExecutionId)
           reconnectPersistenceExecution =
             setIsExecuting(reconnectWorkflowId, true) ??
-            persistenceExecutionsRef.current.get(reconnectWorkflowId)
+            consolePersistence.adoptScopedExecution(reconnectWorkflowId)
           activationOwnsPersistence = Boolean(reconnectPersistenceExecution)
           if (fromEventId === 0) {
             clearExecutionEntries(capturedExecutionId)

apps/sim/hooks/use-collaborative-workflow.ts

@@ -865,8 +865,13 @@ export function useCollaborativeWorkflow() {
           workflowId,
         })
         diffStore.markExternalUpdatePending(workflowId)
-        void operationQueue.waitForWorkflowOperations(workflowId).then((ready) => {
-          if (!ready) {
+        void operationQueue.waitForWorkflowOperations(workflowId).then((result) => {
+          if (result === 'cancelled') {
+            useWorkflowDiffStore.getState().clearExternalUpdatePending(workflowId)
+            return
+          }
+
+          if (result === 'failed') {
             const latestQueue = useOperationQueueStore.getState()
             if (latestQueue.hasPendingOperations(workflowId) && !latestQueue.hasOperationError) {
               return
@@ -879,6 +884,7 @@ export function useCollaborativeWorkflow() {
             )
             return
           }
+
           void replayPendingExternalUpdate(workflowId, 'deferred external update after local save')
         })
         return