From 9c087cd466ac67bcc6713c613d80c0878b2a33bb Mon Sep 17 00:00:00 2001 From: waleed Date: Sun, 22 Feb 2026 15:01:00 -0800 Subject: [PATCH] fix(parallel): error-sticky block run status to prevent branch success masking failure --- .../hooks/use-workflow-execution.ts | 28 ++++++++++++++++-- .../utils/workflow-execution-utils.ts | 29 +++++++++++++++++-- 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts index 10186d6876..9754879ac9 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts @@ -20,7 +20,10 @@ import { TriggerUtils, } from '@/lib/workflows/triggers/triggers' import { useCurrentWorkflow } from '@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-current-workflow' -import { updateActiveBlockRefCount } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils' +import { + resolveBlockRunStatus, + updateActiveBlockRefCount, +} from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils' import { getBlock } from '@/blocks' import type { SerializableExecutionState } from '@/executor/execution/types' import type { @@ -65,6 +68,8 @@ interface BlockEventHandlerConfig { workflowEdges: Array<{ id: string; target: string; sourceHandle?: string | null }> activeBlocksSet: Set activeBlockRefCounts: Map + /** Tracks blocks that have errored in any parallel branch — prevents later success from masking errors. */ + blockRunErrors: Set accumulatedBlockLogs: BlockLog[] accumulatedBlockStates: Map executedBlockIds: Set @@ -312,6 +317,7 @@ export function useWorkflowExecution() { workflowEdges, activeBlocksSet, activeBlockRefCounts, + blockRunErrors, accumulatedBlockLogs, accumulatedBlockStates, executedBlockIds, @@ -484,7 +490,12 @@ export function useWorkflowExecution() { const onBlockCompleted = (data: BlockCompletedData) => { if (isStaleExecution()) return updateActiveBlocks(data.blockId, false) - if (workflowId) setBlockRunStatus(workflowId, data.blockId, 'success') + if (workflowId) + setBlockRunStatus( + workflowId, + data.blockId, + resolveBlockRunStatus(blockRunErrors, data.blockId, 'success') + ) executedBlockIds.add(data.blockId) accumulatedBlockStates.set(data.blockId, { @@ -515,7 +526,12 @@ export function useWorkflowExecution() { const onBlockError = (data: BlockErrorData) => { if (isStaleExecution()) return updateActiveBlocks(data.blockId, false) - if (workflowId) setBlockRunStatus(workflowId, data.blockId, 'error') + if (workflowId) + setBlockRunStatus( + workflowId, + data.blockId, + resolveBlockRunStatus(blockRunErrors, data.blockId, 'error') + ) executedBlockIds.add(data.blockId) accumulatedBlockStates.set(data.blockId, { @@ -1280,6 +1296,7 @@ export function useWorkflowExecution() { const activeBlocksSet = new Set() const activeBlockRefCounts = new Map() + const blockRunErrors = new Set() const streamedContent = new Map() const accumulatedBlockLogs: BlockLog[] = [] const accumulatedBlockStates = new Map() @@ -1293,6 +1310,7 @@ export function useWorkflowExecution() { workflowEdges, activeBlocksSet, activeBlockRefCounts, + blockRunErrors, accumulatedBlockLogs, accumulatedBlockStates, executedBlockIds, @@ -1904,6 +1922,7 @@ export function useWorkflowExecution() { const executedBlockIds = new Set() const activeBlocksSet = new Set() const activeBlockRefCounts = new Map() + const blockRunErrors = new Set() try { const blockHandlers = buildBlockEventHandlers({ @@ -1912,6 +1931,7 @@ export function useWorkflowExecution() { workflowEdges, activeBlocksSet, activeBlockRefCounts, + blockRunErrors, accumulatedBlockLogs, accumulatedBlockStates, executedBlockIds, @@ -2108,6 +2128,7 @@ export function useWorkflowExecution() { const workflowEdges = useWorkflowStore.getState().edges const activeBlocksSet = new Set() const activeBlockRefCounts = new Map() + const blockRunErrors = new Set() const accumulatedBlockLogs: BlockLog[] = [] const accumulatedBlockStates = new Map() const executedBlockIds = new Set() @@ -2120,6 +2141,7 @@ export function useWorkflowExecution() { workflowEdges, activeBlocksSet, activeBlockRefCounts, + blockRunErrors, accumulatedBlockLogs, accumulatedBlockStates, executedBlockIds, diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts index ff1baf222a..988726070d 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts @@ -5,6 +5,22 @@ import { useTerminalConsoleStore } from '@/stores/terminal' import { useWorkflowRegistry } from '@/stores/workflows/registry/store' import { useWorkflowStore } from '@/stores/workflows/workflow/store' +/** + * Returns the run status to persist for a block, using an error-sticky policy. + * Once any parallel branch for a block has errored, the block status stays 'error' + * even if a later branch completes successfully — preventing failures from being masked. + */ +export function resolveBlockRunStatus( + erroredBlocks: Set, + blockId: string, + status: 'success' | 'error' +): 'success' | 'error' { + if (status === 'error') { + erroredBlocks.add(blockId) + } + return erroredBlocks.has(blockId) ? 'error' : 'success' +} + /** * Updates the active blocks set and ref counts for a single block. * Ref counting ensures a block stays active until all parallel branches for it complete. @@ -64,6 +80,7 @@ export async function executeWorkflowWithFullLogging( const activeBlocksSet = new Set() const activeBlockRefCounts = new Map() + const blockRunErrors = new Set() const payload: any = { input: options.workflowInput, @@ -154,7 +171,11 @@ export async function executeWorkflowWithFullLogging( ) setActiveBlocks(wfId, new Set(activeBlocksSet)) - setBlockRunStatus(wfId, event.data.blockId, 'success') + setBlockRunStatus( + wfId, + event.data.blockId, + resolveBlockRunStatus(blockRunErrors, event.data.blockId, 'success') + ) addConsole({ input: event.data.input || {}, @@ -190,7 +211,11 @@ export async function executeWorkflowWithFullLogging( ) setActiveBlocks(wfId, new Set(activeBlocksSet)) - setBlockRunStatus(wfId, event.data.blockId, 'error') + setBlockRunStatus( + wfId, + event.data.blockId, + resolveBlockRunStatus(blockRunErrors, event.data.blockId, 'error') + ) addConsole({ input: event.data.input || {},