fix(parallel): error-sticky block run status to prevent branch success masking failure

This commit is contained in:
waleed
2026-02-22 15:01:00 -08:00
parent 6b407eeb60
commit 9c087cd466
2 changed files with 52 additions and 5 deletions
@@ -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<string>
activeBlockRefCounts: Map<string, number>
/** Tracks blocks that have errored in any parallel branch — prevents later success from masking errors. */
blockRunErrors: Set<string>
accumulatedBlockLogs: BlockLog[]
accumulatedBlockStates: Map<string, BlockState>
executedBlockIds: Set<string>
@@ -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<string>()
const activeBlockRefCounts = new Map<string, number>()
const blockRunErrors = new Set<string>()
const streamedContent = new Map<string, string>()
const accumulatedBlockLogs: BlockLog[] = []
const accumulatedBlockStates = new Map<string, BlockState>()
@@ -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<string>()
const activeBlocksSet = new Set<string>()
const activeBlockRefCounts = new Map<string, number>()
const blockRunErrors = new Set<string>()
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<string>()
const activeBlockRefCounts = new Map<string, number>()
const blockRunErrors = new Set<string>()
const accumulatedBlockLogs: BlockLog[] = []
const accumulatedBlockStates = new Map<string, BlockState>()
const executedBlockIds = new Set<string>()
@@ -2120,6 +2141,7 @@ export function useWorkflowExecution() {
workflowEdges,
activeBlocksSet,
activeBlockRefCounts,
blockRunErrors,
accumulatedBlockLogs,
accumulatedBlockStates,
executedBlockIds,
@@ -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<string>,
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<string>()
const activeBlockRefCounts = new Map<string, number>()
const blockRunErrors = new Set<string>()
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 || {},