improvement(executor): correctness-by-construction for workflow logs

Replace the post-hoc reconciliation layer with a deterministic emission
protocol: drain pending callback promises at terminal boundaries, mint a
per-invocation blockExecutionId, and key console entries by that ID.
Eliminates races between block:* and execution:* events without changing
per-block latency.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -919,7 +919,8 @@ async function handleExecutePost(
             blockType: string,
             executionOrder: number,
             iterationContext?: IterationContext,
-            childWorkflowContext?: ChildWorkflowContext
+            childWorkflowContext?: ChildWorkflowContext,
+            blockExecutionId?: string
           ) => {
             reqLogger.info('onBlockStart called', { blockId, blockName, blockType })
             sendEvent({
@@ -945,6 +946,7 @@ async function handleExecutePost(
                   childWorkflowBlockId: childWorkflowContext.parentBlockId,
                   childWorkflowName: childWorkflowContext.workflowName,
                 }),
+                ...(blockExecutionId && { blockExecutionId }),
               },
             })
           }
@@ -955,7 +957,8 @@ async function handleExecutePost(
             blockType: string,
             callbackData: any,
             iterationContext?: IterationContext,
-            childWorkflowContext?: ChildWorkflowContext
+            childWorkflowContext?: ChildWorkflowContext,
+            blockExecutionId?: string
           ) => {
             const hasError = callbackData.output?.error
             const childWorkflowData = childWorkflowContext
@@ -969,6 +972,11 @@ async function handleExecutePost(
               ? { childWorkflowInstanceId: callbackData.childWorkflowInstanceId }
               : {}
 
+            const resolvedBlockExecutionId = blockExecutionId ?? callbackData.blockExecutionId
+            const blockExecData = resolvedBlockExecutionId
+              ? { blockExecutionId: resolvedBlockExecutionId }
+              : {}
+
             if (hasError) {
               reqLogger.info('onBlockComplete (error) called', {
                 blockId,
@@ -1002,6 +1010,7 @@ async function handleExecutePost(
                   }),
                   ...childWorkflowData,
                   ...instanceData,
+                  ...blockExecData,
                 },
               })
             } else {
@@ -1036,6 +1045,7 @@ async function handleExecutePost(
                   }),
                   ...childWorkflowData,
                   ...instanceData,
+                  ...blockExecData,
                 },
               })
             }
@@ -1165,7 +1175,6 @@ async function handleExecutePost(
                 data: {
                   error: timeoutErrorMessage,
                   duration: result.metadata?.duration || 0,
-                  finalBlockLogs: result.logs,
                 },
               })
               finalMetaStatus = 'error'
@@ -1179,7 +1188,6 @@ async function handleExecutePost(
                 workflowId,
                 data: {
                   duration: result.metadata?.duration || 0,
-                  finalBlockLogs: result.logs,
                 },
               })
               finalMetaStatus = 'cancelled'
@@ -1244,7 +1252,6 @@ async function handleExecutePost(
             data: {
               error: executionResult?.error || errorMessage,
               duration: executionResult?.metadata?.duration || 0,
-              finalBlockLogs: executionResult?.logs,
             },
           })
           finalMetaStatus = 'error'

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

@@ -218,31 +218,25 @@ export function useWorkflowExecution() {
       durationMs?: number
       blockLogs: BlockLog[]
       isPreExecutionError?: boolean
-      finalBlockLogs?: BlockLog[]
     }) => {
       if (!params.workflowId) return
       sharedHandleExecutionErrorConsole(
-        { addConsole, updateConsole, cancelRunningEntries },
+        { addConsole, updateConsole },
         { ...params, workflowId: params.workflowId }
       )
     },
-    [addConsole, cancelRunningEntries, updateConsole]
+    [addConsole, updateConsole]
   )
 
   const handleExecutionCancelledConsole = useCallback(
-    (params: {
-      workflowId?: string
-      executionId?: string
-      durationMs?: number
-      finalBlockLogs?: BlockLog[]
-    }) => {
+    (params: { workflowId?: string; executionId?: string; durationMs?: number }) => {
       if (!params.workflowId) return
       sharedHandleExecutionCancelledConsole(
-        { addConsole, updateConsole, cancelRunningEntries },
+        { addConsole, updateConsole },
         { ...params, workflowId: params.workflowId }
       )
     },
-    [addConsole, cancelRunningEntries, updateConsole]
+    [addConsole, updateConsole]
   )
 
   const buildBlockEventHandlers = useCallback(
@@ -1024,8 +1018,6 @@ export function useWorkflowExecution() {
           accumulatedBlockLogs,
           accumulatedBlockStates,
           executedBlockIds,
-          consoleMode: 'update',
-          includeStartConsoleEntry: true,
           onBlockCompleteCallback: onBlockComplete,
         })
 
@@ -1228,7 +1220,6 @@ export function useWorkflowExecution() {
                 durationMs: data.duration,
                 blockLogs: accumulatedBlockLogs,
                 isPreExecutionError,
-                finalBlockLogs: data.finalBlockLogs,
               })
 
               if (activeWorkflowId && !isExecutingFromChat) {
@@ -1255,7 +1246,6 @@ export function useWorkflowExecution() {
                 workflowId: activeWorkflowId,
                 executionId: executionIdRef.current,
                 durationMs: data?.duration,
-                finalBlockLogs: data?.finalBlockLogs,
               })
 
               if (activeWorkflowId && !isExecutingFromChat) {
@@ -1672,8 +1662,6 @@ export function useWorkflowExecution() {
           accumulatedBlockLogs,
           accumulatedBlockStates,
           executedBlockIds,
-          consoleMode: 'update',
-          includeStartConsoleEntry: true,
         })
 
         await executionStream.executeFromBlock({
@@ -1743,7 +1731,6 @@ export function useWorkflowExecution() {
                 error: data.error,
                 durationMs: data.duration,
                 blockLogs: accumulatedBlockLogs,
-                finalBlockLogs: data.finalBlockLogs,
               })
 
               setCurrentExecutionId(workflowId, null)
@@ -1756,7 +1743,6 @@ export function useWorkflowExecution() {
                 workflowId,
                 executionId: executionIdRef.current,
                 durationMs: data?.duration,
-                finalBlockLogs: data?.finalBlockLogs,
               })
 
               setCurrentExecutionId(workflowId, null)
@@ -1903,8 +1889,6 @@ export function useWorkflowExecution() {
         accumulatedBlockLogs,
         accumulatedBlockStates,
         executedBlockIds,
-        consoleMode: 'update',
-        includeStartConsoleEntry: true,
       })
 
       const capturedExecutionId = executionId
@@ -2005,7 +1989,6 @@ export function useWorkflowExecution() {
                   executionId: capturedExecutionId,
                   error: data.error,
                   blockLogs: accumulatedBlockLogs,
-                  finalBlockLogs: data.finalBlockLogs,
                 })
               },
               onExecutionCancelled: (data) => {
@@ -2026,7 +2009,6 @@ export function useWorkflowExecution() {
                   workflowId: reconnectWorkflowId,
                   executionId: capturedExecutionId,
                   durationMs: data?.duration,
-                  finalBlockLogs: data?.finalBlockLogs,
                 })
               },
             },