diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index 1370aea8684..c4fa02057ea 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/apps/sim/app/api/workflows/[id]/execute/route.ts @@ -1866,7 +1866,8 @@ async function handleExecutePost( blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { reqLogger.info('onBlockStart called', { blockId, blockName, blockType }) await sendEvent({ @@ -1892,6 +1893,7 @@ async function handleExecutePost( childWorkflowBlockId: childWorkflowContext.parentBlockId, childWorkflowName: childWorkflowContext.workflowName, }), + ...(blockExecutionId && { blockExecutionId }), }, }) } @@ -1902,7 +1904,8 @@ async function handleExecutePost( blockType: string, callbackData: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { const compactCallbackData = { ...callbackData, @@ -1982,6 +1985,9 @@ async function handleExecutePost( }), ...childWorkflowData, ...instanceData, + ...((blockExecutionId || callbackData.blockExecutionId) && { + blockExecutionId: blockExecutionId ?? callbackData.blockExecutionId, + }), }, }) } else { @@ -2021,6 +2027,9 @@ async function handleExecutePost( }), ...childWorkflowData, ...instanceData, + ...((blockExecutionId || callbackData.blockExecutionId) && { + blockExecutionId: blockExecutionId ?? callbackData.blockExecutionId, + }), }, }) } @@ -2028,6 +2037,7 @@ async function handleExecutePost( const onStream = async (streamingExec: StreamingExecution) => { const blockId = (streamingExec.execution as any).blockId + const { blockExecutionId } = streamingExec // Live answer text rides the sink when available (pending deltas // stream as the model generates; chunk_reset clears intermediate @@ -2038,6 +2048,7 @@ async function handleExecutePost( // Sync window: attach sink before first await so pump delivers thinking/tools. const unsubscribe = forwardAgentStreamToExecutionEvents(streamingExec, { blockId, + blockExecutionId, executionId, workflowId, sendEvent, @@ -2079,7 +2090,7 @@ async function handleExecutePost( timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId, chunk, display }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }), chunk, display }, }) } @@ -2089,7 +2100,7 @@ async function handleExecutePost( timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }) }, }) } } catch (error) { @@ -2150,7 +2161,8 @@ async function handleExecutePost( childWorkflowInstanceId: string, iterationContext?: IterationContext, executionOrder?: number, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { await sendEvent({ type: 'block:childWorkflowStarted', @@ -2174,6 +2186,7 @@ async function handleExecutePost( childWorkflowName: childWorkflowContext.workflowName, }), ...(executionOrder !== undefined && { executionOrder }), + ...(blockExecutionId && { blockExecutionId }), }, }) } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx index 15c7171064c..2ad1db5d4ca 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx @@ -2,11 +2,13 @@ * @vitest-environment jsdom */ import { act, type ReactNode } from 'react' +import { sleep } from '@sim/utils/helpers' import { QueryClient, QueryClientProvider } from '@tanstack/react-query' import { createRoot, type Root } from 'react-dom/client' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const { + chatStoreState, executionStoreState, mockCancel, mockExecute, @@ -22,6 +24,9 @@ const { workflowBlocks, workflowStoreState, } = vi.hoisted(() => { + const chatStoreState = { + selectedWorkflowOutputs: [] as string[], + } const workflowBlocks = { start: { id: 'start', @@ -82,6 +87,7 @@ const { } return { + chatStoreState, executionStoreState, mockCancel: vi.fn(), mockExecute: vi.fn(), @@ -167,9 +173,18 @@ vi.mock('@/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-current-workflow vi.mock('@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils', () => ({ addHttpErrorConsoleEntry: vi.fn(), - createBlockEventHandlers: () => ({ + createBlockEventHandlers: (config: { + onBlockCompleteCallback?: ( + blockId: string, + output: unknown, + blockExecutionId?: string + ) => Promise + }) => ({ onBlockStarted: vi.fn(), - onBlockCompleted: vi.fn(), + onBlockCompleted: vi.fn( + (data: { blockId: string; output: unknown; blockExecutionId?: string }) => + config.onBlockCompleteCallback?.(data.blockId, data.output, data.blockExecutionId) + ), onBlockError: vi.fn(), onBlockChildWorkflowStarted: vi.fn(), }), @@ -221,7 +236,7 @@ vi.mock('@/serializer', () => ({ vi.mock('@/stores/chat/store', () => ({ useChatStore: { getState: () => ({ - getSelectedWorkflowOutput: () => [], + getSelectedWorkflowOutput: () => chatStoreState.selectedWorkflowOutputs, }), }, })) @@ -343,6 +358,7 @@ async function drainStream(value: unknown): Promise { describe('useWorkflowExecution cancellation', () => { beforeEach(() => { vi.clearAllMocks() + chatStoreState.selectedWorkflowOutputs = [] executionStoreState.getCurrentExecutionId.mockReturnValue('execution-1') mockRequestJson.mockResolvedValue({ success: true }) }) @@ -402,6 +418,7 @@ describe('useWorkflowExecution cancellation', () => { describe('useWorkflowExecution attachment uploads', () => { beforeEach(() => { vi.clearAllMocks() + chatStoreState.selectedWorkflowOutputs = [] executionStoreState.getCurrentExecutionId.mockReturnValue(null) mockResolveStartCandidates.mockReturnValue([]) mockSelectBestTrigger.mockReturnValue([]) @@ -584,6 +601,52 @@ describe('useWorkflowExecution attachment uploads', () => { unmount() }) + it('does not append a later sibling output after the block has streamed', async () => { + chatStoreState.selectedWorkflowOutputs = ['agent-1_content'] + mockExecute.mockImplementationOnce(async (options) => { + options.onExecutionId?.('execution-1') + await options.callbacks?.onStreamChunk?.({ + blockId: 'agent-1', + blockExecutionId: 'invoke-streamed', + chunk: 'streamed answer', + }) + await sleep(0) + await options.callbacks?.onBlockCompleted?.({ + blockId: 'agent-1', + blockExecutionId: 'invoke-streamed', + output: { content: 'streamed answer' }, + }) + await options.callbacks?.onBlockCompleted?.({ + blockId: 'agent-1', + blockExecutionId: 'invoke-later', + output: { content: 'later answer' }, + }) + }) + + const { result, unmount } = renderWorkflowExecutionHook() + const decoder = new TextDecoder() + let streamedText = '' + + await act(async () => { + const runResult = await result().handleRunWorkflow({ input: 'chat input' }) + if (!isChatWorkflowRunResult(runResult)) { + throw new Error('Expected a chat workflow run result') + } + const reader = runResult.stream.getReader() + while (true) { + const { done, value } = await reader.read() + if (done) break + streamedText += decoder.decode(value, { stream: true }) + } + streamedText += decoder.decode() + }) + + expect(streamedText).toContain('streamed answer') + expect(streamedText).not.toContain('later answer') + + unmount() + }) + it('preserves legacy live thinking when no display projection field is sent', async () => { mockExecute.mockImplementationOnce(async (options) => { options.onExecutionId?.('execution-1') 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 f999a2294ee..76fdc99a08d 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 @@ -262,55 +262,86 @@ function createAgentStreamChrome({ executionIdRef, updateConsole }: AgentStreamC const toolCallsByBlock = new Map>() const toolOrderByBlock = new Map() const thinkingFlushTimers = new Map>() + const blockIdByInvocation = new Map() + const blockExecutionIdByInvocation = new Map() + + const invocationKey = (blockId: string, blockExecutionId?: string) => { + const key = blockExecutionId ?? blockId + blockIdByInvocation.set(key, blockId) + if (blockExecutionId) blockExecutionIdByInvocation.set(key, blockExecutionId) + return key + } - const flushThinking = (blockId: string) => { - const timer = thinkingFlushTimers.get(blockId) + const flushThinking = (blockId: string, blockExecutionId?: string) => { + const key = invocationKey(blockId, blockExecutionId) + const timer = thinkingFlushTimers.get(key) if (timer !== undefined) { clearTimeout(timer) - thinkingFlushTimers.delete(blockId) + thinkingFlushTimers.delete(key) } - const thinking = thinkingByBlock.get(blockId) + const thinking = thinkingByBlock.get(key) if (thinking === undefined) return updateConsole( blockId, - { agentStreamThinking: thinking, agentStreamActive: true }, + { + ...(blockExecutionId && { blockExecutionId }), + agentStreamThinking: thinking, + agentStreamActive: true, + }, executionIdRef.current ) } - const clearThinking = (blockId: string) => { - const timer = thinkingFlushTimers.get(blockId) + const clearThinking = (blockId: string, blockExecutionId?: string) => { + const key = invocationKey(blockId, blockExecutionId) + const timer = thinkingFlushTimers.get(key) if (timer !== undefined) { clearTimeout(timer) - thinkingFlushTimers.delete(blockId) + thinkingFlushTimers.delete(key) } - thinkingByBlock.delete(blockId) - updateConsole(blockId, { clearAgentStreamThinking: true }, executionIdRef.current) + thinkingByBlock.delete(key) + updateConsole( + blockId, + { ...(blockExecutionId && { blockExecutionId }), clearAgentStreamThinking: true }, + executionIdRef.current + ) } - const settleBlock = (blockId: string, status: 'success' | 'error' | 'cancelled') => { - flushThinking(blockId) - const map = toolCallsByBlock.get(blockId) - const order = toolOrderByBlock.get(blockId) + const settleBlock = ( + blockId: string, + status: 'success' | 'error' | 'cancelled', + blockExecutionId?: string + ) => { + const key = invocationKey(blockId, blockExecutionId) + flushThinking(blockId, blockExecutionId) + const map = toolCallsByBlock.get(key) + const order = toolOrderByBlock.get(key) if (map && order) { settleRunningToolCalls(map, status) updateConsole( blockId, { + ...(blockExecutionId && { blockExecutionId }), agentStreamActive: false, agentStreamToolCalls: snapshotToolCalls(order, map), }, executionIdRef.current ) } else { - updateConsole(blockId, { agentStreamActive: false }, executionIdRef.current) + updateConsole( + blockId, + { ...(blockExecutionId && { blockExecutionId }), agentStreamActive: false }, + executionIdRef.current + ) } } const settleAll = (status: 'success' | 'error' | 'cancelled') => { - const blockIds = new Set([...thinkingByBlock.keys(), ...toolCallsByBlock.keys()]) - for (const blockId of blockIds) { - settleBlock(blockId, status) + const invocationKeys = new Set([...thinkingByBlock.keys(), ...toolCallsByBlock.keys()]) + for (const key of invocationKeys) { + const blockId = blockIdByInvocation.get(key) + if (!blockId) continue + settleBlock(blockId, status, blockExecutionIdByInvocation.get(key)) } } @@ -323,34 +354,39 @@ function createAgentStreamChrome({ executionIdRef, updateConsole }: AgentStreamC const hasDisplayProjection = Object.hasOwn(data, 'display') const text = hasDisplayProjection ? display?.text : data.text if (display?.clearLiveDisplay || (hasDisplayProjection && typeof text !== 'string')) { - clearThinking(data.blockId) + clearThinking(data.blockId, data.blockExecutionId) return } if (!text) return - const prev = thinkingByBlock.get(data.blockId) ?? '' - thinkingByBlock.set(data.blockId, prev + text) - if (!thinkingFlushTimers.has(data.blockId)) { + const key = invocationKey(data.blockId, data.blockExecutionId) + const prev = thinkingByBlock.get(key) ?? '' + thinkingByBlock.set(key, prev + text) + if (!thinkingFlushTimers.has(key)) { thinkingFlushTimers.set( - data.blockId, - setTimeout(() => flushThinking(data.blockId), AGENT_STREAM_THINKING_FLUSH_MS) + key, + setTimeout( + () => flushThinking(data.blockId, data.blockExecutionId), + AGENT_STREAM_THINKING_FLUSH_MS + ) ) } } const onStreamTool = (data: StreamToolData) => { - if (!toolCallsByBlock.has(data.blockId)) { - toolCallsByBlock.set(data.blockId, new Map()) - toolOrderByBlock.set(data.blockId, []) + const key = invocationKey(data.blockId, data.blockExecutionId) + if (!toolCallsByBlock.has(key)) { + toolCallsByBlock.set(key, new Map()) + toolOrderByBlock.set(key, []) } - const map = toolCallsByBlock.get(data.blockId)! - const order = toolOrderByBlock.get(data.blockId)! + const map = toolCallsByBlock.get(key)! + const order = toolOrderByBlock.get(key)! applyToolCallPhase( map, order, { - key: toolCallKey(data.blockId, data.id), + key: toolCallKey(key, data.id), id: data.id, name: data.name, phase: data.phase, @@ -362,6 +398,7 @@ function createAgentStreamChrome({ executionIdRef, updateConsole }: AgentStreamC updateConsole( data.blockId, { + ...(data.blockExecutionId && { blockExecutionId: data.blockExecutionId }), agentStreamToolCalls: snapshotToolCalls(order, map), agentStreamActive: true, }, @@ -371,7 +408,7 @@ function createAgentStreamChrome({ executionIdRef, updateConsole }: AgentStreamC const onStreamDone = (data: StreamDoneData) => { logger.info('Stream done for block:', data.blockId) - settleBlock(data.blockId, 'success') + settleBlock(data.blockId, 'success', data.blockExecutionId) } return { @@ -747,6 +784,7 @@ export function useWorkflowExecution() { async start(controller) { const { encodeSSE } = await import('@/lib/core/utils/sse') const streamedChunks = new Map() + const streamedBlockIds = new Set() const streamReadingPromises: Promise[] = [] const safeEnqueue = (data: Uint8Array) => { @@ -767,28 +805,30 @@ export function useWorkflowExecution() { if (!streamingExecution.stream) return const reader = streamingExecution.stream.getReader() const blockId = (streamingExecution.execution as any)?.blockId + const streamKey = streamingExecution.blockExecutionId ?? blockId - if (blockId && !streamedChunks.has(blockId)) { - streamedChunks.set(blockId, []) + if (streamKey && !streamedChunks.has(streamKey)) { + streamedChunks.set(streamKey, []) } try { while (true) { const { done, value } = await reader.read() if (done) { - if (blockId) { - streamCompletionTimes.set(blockId, Date.now()) + if (streamKey) { + streamCompletionTimes.set(streamKey, Date.now()) } break } const chunk = new TextDecoder().decode(value) - if (blockId) { - streamedChunks.get(blockId)!.push(chunk) + if (streamKey) { + if (blockId) streamedBlockIds.add(blockId) + streamedChunks.get(streamKey)!.push(chunk) } let chunkToSend = chunk - if (blockId && !processedFirstChunk.has(blockId)) { - processedFirstChunk.add(blockId) + if (streamKey && !processedFirstChunk.has(streamKey)) { + processedFirstChunk.add(streamKey) if (streamedChunks.size > 1) { chunkToSend = `\n\n${chunk}` } @@ -806,21 +846,27 @@ export function useWorkflowExecution() { /** * Intermediate-turn reconciliation: drop the block's streamed text - * (chunk_reset frame) and remove its bookkeeping entirely so - * separator counting ignores it and the final turn (or, if none - * re-streams, onBlockComplete's output fallback) starts clean. + * (chunk_reset frame) and remove its per-invocation chunks so + * separator counting ignores them. The block-level streamed marker + * remains to prevent a sibling invocation from appending stale output. */ - const onStreamReset = (blockId: string) => { - if (!streamedChunks.has(blockId)) return - streamedChunks.delete(blockId) - processedFirstChunk.delete(blockId) + const onStreamReset = (blockId: string, blockExecutionId?: string) => { + const streamKey = blockExecutionId ?? blockId + if (!streamedChunks.has(streamKey)) return + streamedChunks.delete(streamKey) + processedFirstChunk.delete(streamKey) safeEnqueue(encodeSSE({ blockId, event: 'chunk_reset' })) } // Handle non-streaming blocks (like Function blocks) - const onBlockComplete = async (blockId: string, output: any) => { + const onBlockComplete = async ( + blockId: string, + output: any, + blockExecutionId?: string + ) => { + const streamKey = blockExecutionId ?? blockId // Skip if this block already had streaming content (avoid duplicates) - if (streamedChunks.has(blockId)) { + if (streamedChunks.has(streamKey) || streamedBlockIds.has(blockId)) { logger.debug('[handleRunWorkflow] Skipping onBlockComplete for streaming block', { blockId, }) @@ -863,7 +909,7 @@ export function useWorkflowExecution() { safeEnqueue(encodeSSE({ blockId, chunk: separator + formattedOutput })) // Track that we've sent output for this block - streamedChunks.set(blockId, [formattedOutput]) + streamedChunks.set(streamKey, [formattedOutput]) } } } @@ -896,8 +942,9 @@ export function useWorkflowExecution() { // Update block logs with actual stream completion times if (result.logs && streamCompletionTimes.size > 0) { result.logs.forEach((log: BlockLog) => { - if (streamCompletionTimes.has(log.blockId)) { - const completionTime = streamCompletionTimes.get(log.blockId)! + const streamKey = log.blockExecutionId ?? log.blockId + if (streamCompletionTimes.has(streamKey)) { + const completionTime = streamCompletionTimes.get(streamKey)! const startTime = new Date(log.startedAt).getTime() // Update the log with actual stream completion time @@ -1016,10 +1063,10 @@ export function useWorkflowExecution() { workflowInput?: any, onStream?: (se: StreamingExecution) => Promise, executionId?: string, - onBlockComplete?: (blockId: string, output: any) => Promise, + onBlockComplete?: (blockId: string, output: any, blockExecutionId?: string) => Promise, overrideTriggerType?: 'chat' | 'manual' | 'api', stopAfterBlockId?: string, - onStreamReset?: (blockId: string) => void + onStreamReset?: (blockId: string, blockExecutionId?: string) => void ): Promise => { // Use diff workflow for execution when available, regardless of canvas view state const executionWorkflowState = null as { @@ -1310,16 +1357,17 @@ export function useWorkflowExecution() { onBlockStarted: blockHandlers.onBlockStarted, onBlockCompleted: blockHandlers.onBlockCompleted, onBlockError: (data) => { - agentStreamChrome.settleBlock(data.blockId, 'error') + agentStreamChrome.settleBlock(data.blockId, 'error', data.blockExecutionId) blockHandlers.onBlockError(data) }, onBlockChildWorkflowStarted: blockHandlers.onBlockChildWorkflowStarted, onStreamChunk: (data) => { - if (!streamedChunks.has(data.blockId)) { - streamedChunks.set(data.blockId, []) + const streamKey = data.blockExecutionId ?? data.blockId + if (!streamedChunks.has(streamKey)) { + streamedChunks.set(streamKey, []) } - streamedChunks.get(data.blockId)!.push(data.chunk) + streamedChunks.get(streamKey)!.push(data.chunk) // Call onStream callback if provided (create a fake StreamingExecution) if (onStream && isExecutingFromChat) { @@ -1332,6 +1380,7 @@ export function useWorkflowExecution() { const streamingExec: StreamingExecution = { stream, + ...(data.blockExecutionId && { blockExecutionId: data.blockExecutionId }), execution: { success: true, output: { content: '' }, @@ -1348,9 +1397,9 @@ export function useWorkflowExecution() { onStreamChunkReset: (data) => { // Live-streamed text belonged to an intermediate turn (tools // follow); the final turn re-streams as regular chunks. - streamedChunks.delete(data.blockId) + streamedChunks.delete(data.blockExecutionId ?? data.blockId) if (onStreamReset && isExecutingFromChat) { - onStreamReset(data.blockId) + onStreamReset(data.blockId, data.blockExecutionId) } }, @@ -2088,7 +2137,7 @@ export function useWorkflowExecution() { onBlockStarted: blockHandlers.onBlockStarted, onBlockCompleted: blockHandlers.onBlockCompleted, onBlockError: (data) => { - agentStreamChrome.settleBlock(data.blockId, 'error') + agentStreamChrome.settleBlock(data.blockId, 'error', data.blockExecutionId) blockHandlers.onBlockError(data) }, onBlockChildWorkflowStarted: blockHandlers.onBlockChildWorkflowStarted, diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.integration.test.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.integration.test.ts index 642b6327c58..4a14f087639 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.integration.test.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.integration.test.ts @@ -21,6 +21,7 @@ describe('reconcileFinalBlockLogs (real store)', () => { useTerminalConsoleStore.setState({ workflowEntries: {}, entryIdsByBlockExecution: {}, + entryIdByBlockExecutionId: {}, entryLocationById: {}, isOpen: false, _hasHydrated: true, diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts index 10db82c7614..af7772943c6 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts @@ -612,6 +612,7 @@ describe('workflow-execution-utils', () => { blockName: 'Function', blockType: 'function', executionId: 'exec-1', + blockExecutionId: 'fn-invoke-1', executionOrder: 3, isRunning: true, }) @@ -620,6 +621,7 @@ describe('workflow-execution-utils', () => { reconcileFinalBlockLogs(updateConsole, 'wf-1', 'exec-1', [ makeLog({ blockId: 'fn-1', + blockExecutionId: 'fn-invoke-1', executionOrder: 3, success: false, error: 'JSON parse failed', @@ -628,6 +630,7 @@ describe('workflow-execution-utils', () => { expect(updateConsole).toHaveBeenCalledTimes(1) expect(updateConsole.mock.calls[0][1]).toMatchObject({ + blockExecutionId: 'fn-invoke-1', success: false, error: 'JSON parse failed', isRunning: false, @@ -1315,6 +1318,7 @@ describe('workflow-execution-utils', () => { terminalConsoleMockFns.mockAddConsole({ workflowId: 'wf-1', blockId: 'fn-leaf', + blockExecutionId: 'leaf-invoke-0', blockType: 'function', blockName: 'Leaf', executionId: 'exec-1', @@ -1329,6 +1333,7 @@ describe('workflow-execution-utils', () => { terminalConsoleMockFns.mockAddConsole({ workflowId: 'wf-1', blockId: 'fn-leaf', + blockExecutionId: 'leaf-invoke-1', blockType: 'function', blockName: 'Leaf', executionId: 'exec-1', @@ -1354,6 +1359,7 @@ describe('workflow-execution-utils', () => { name: 'Leaf', type: 'function', blockId: 'fn-leaf', + blockExecutionId: 'leaf-invoke-0', executionOrder: 2, loopId: 'loop-1', iterationIndex: 0, @@ -1368,6 +1374,7 @@ describe('workflow-execution-utils', () => { name: 'Leaf', type: 'function', blockId: 'fn-leaf', + blockExecutionId: 'leaf-invoke-1', executionOrder: 3, loopId: 'loop-1', iterationIndex: 1, @@ -1386,10 +1393,12 @@ describe('workflow-execution-utils', () => { // still-running iteration is actually mutated. We assert the args carry // distinct iteration identities so the store can target the right row. expect(updateConsole.mock.calls[0][1]).toMatchObject({ + blockExecutionId: 'leaf-invoke-0', executionOrder: 2, iterationCurrent: 0, }) expect(updateConsole.mock.calls[1][1]).toMatchObject({ + blockExecutionId: 'leaf-invoke-1', executionOrder: 3, iterationCurrent: 1, replaceOutput: { i: 1 }, 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 4dc59961db6..eb6ffd23390 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 @@ -127,7 +127,11 @@ export interface BlockEventHandlerConfig { accumulatedBlockStates: Map executedBlockIds: Set includeStartConsoleEntry: boolean - onBlockCompleteCallback?: (blockId: string, output: unknown) => Promise + onBlockCompleteCallback?: ( + blockId: string, + output: unknown, + blockExecutionId?: string + ) => Promise } interface BlockEventHandlerDeps { @@ -239,6 +243,7 @@ export function createBlockEventHandlers( ...('childWorkflowInstanceId' in data && { childWorkflowInstanceId: data.childWorkflowInstanceId, }), + ...(data.blockExecutionId && { blockExecutionId: data.blockExecutionId }), }) const parentIterationsMatch = ( @@ -260,6 +265,7 @@ export function createBlockEventHandlers( type StartedIdentity = { blockId: string + blockExecutionId?: string executionOrder?: number iterationCurrent?: BlockStartedData['iterationCurrent'] iterationTotal?: BlockStartedData['iterationTotal'] @@ -273,6 +279,7 @@ export function createBlockEventHandlers( const startedEntryKey = (data: StartedIdentity) => JSON.stringify({ blockId: data.blockId, + blockExecutionId: data.blockExecutionId, executionOrder: data.executionOrder, iterationCurrent: data.iterationCurrent, iterationTotal: data.iterationTotal, @@ -286,6 +293,7 @@ export function createBlockEventHandlers( const matchesStartedIdentity = (entry: ConsoleEntry, data: StartedIdentity) => entry.executionId === executionIdRef.current && entry.blockId === data.blockId && + (data.blockExecutionId ? entry.blockExecutionId === data.blockExecutionId : true) && (data.executionOrder === undefined || entry.executionOrder === data.executionOrder) && entry.iterationCurrent === data.iterationCurrent && entry.iterationTotal === data.iterationTotal && @@ -324,6 +332,7 @@ export function createBlockEventHandlers( childWorkflowName: data.childWorkflowName, }), ...(data.executionOrder !== undefined && { executionOrder: data.executionOrder }), + ...(data.blockExecutionId && { blockExecutionId: data.blockExecutionId }), }, executionIdRef.current ) @@ -334,6 +343,7 @@ export function createBlockEventHandlers( options: { success: boolean; output?: unknown; error?: string } ): BlockLog => ({ blockId: data.blockId, + ...(data.blockExecutionId && { blockExecutionId: data.blockExecutionId }), blockName: data.blockName || 'Unknown Block', blockType: data.blockType || 'unknown', input: data.input || {}, @@ -470,7 +480,10 @@ export function createBlockEventHandlers( updateConsoleEntry(data) if (onBlockCompleteCallback) { - onBlockCompleteCallback(data.blockId, data.output).catch((error) => { + const completionPromise = data.blockExecutionId + ? onBlockCompleteCallback(data.blockId, data.output, data.blockExecutionId) + : onBlockCompleteCallback(data.blockId, data.output) + completionPromise.catch((error) => { logger.error('Error in onBlockComplete callback:', { blockId: data.blockId, error }) }) } @@ -554,11 +567,31 @@ export function reconcileFinalBlockLogs( for (const log of finalBlockLogs) { const errorMessage = normalizeDisplayError(log.error) const entries = useTerminalConsoleStore.getState().getWorkflowEntries(workflowId) - const matchesFinalLog = (entry: ConsoleEntry) => - entry.blockId === log.blockId && - entry.executionId === executionId && - entry.executionOrder === log.executionOrder - const matchingEntry = entries.find(matchesFinalLog) + const invocationEntry = log.blockExecutionId + ? entries.find( + (entry) => + entry.blockId === log.blockId && + entry.executionId === executionId && + entry.blockExecutionId === log.blockExecutionId + ) + : undefined + const legacyCandidates = invocationEntry + ? [] + : entries.filter( + (entry) => + entry.blockExecutionId === undefined && + matchesFinalBlockLogIdentity(entry, log, executionId) + ) + if (!invocationEntry && legacyCandidates.length > 1) { + logger.warn('Ignoring ambiguous legacy final block log', { + blockId: log.blockId, + executionId, + executionOrder: log.executionOrder, + candidateCount: legacyCandidates.length, + }) + } + const matchingEntry = + invocationEntry ?? (legacyCandidates.length === 1 ? legacyCandidates[0] : undefined) const hasExistingContent = matchingEntry?.input !== undefined || matchingEntry?.output !== undefined || @@ -579,6 +612,7 @@ export function reconcileFinalBlockLogs( log.blockId, { executionOrder: log.executionOrder, + ...(log.blockExecutionId && { blockExecutionId: log.blockExecutionId }), blockName: log.blockName, blockType: log.blockType, replaceOutput: (log.output ?? {}) as Record, @@ -684,6 +718,7 @@ function spanConsoleIdentity(span: TraceSpan, childWorkflowInstanceId: string): const iterationContainerId = span.loopId ?? span.parallelId const iterationType = span.loopId ? 'loop' : span.parallelId ? 'parallel' : undefined return { + ...(span.blockExecutionId && { blockExecutionId: span.blockExecutionId }), blockName: span.name, blockType: span.type, ...(span.executionOrder !== undefined && { executionOrder: span.executionOrder }), @@ -703,18 +738,31 @@ function findConsoleEntryForSpan( ): ConsoleEntry | undefined { if (!span.blockId) return undefined const identity = spanConsoleIdentity(span, childWorkflowInstanceId) - return useTerminalConsoleStore - .getState() - .getWorkflowEntries(workflowId) - .find( - (entry) => - entry.blockId === span.blockId && - entry.executionId === executionId && - matchesConsoleIdentity(entry, identity) - ) + const entries = useTerminalConsoleStore.getState().getWorkflowEntries(workflowId) + const invocationEntry = span.blockExecutionId + ? entries.find( + (entry) => + entry.blockId === span.blockId && + entry.executionId === executionId && + entry.blockExecutionId === span.blockExecutionId + ) + : undefined + if (invocationEntry) return invocationEntry + + const legacyCandidates = entries.filter( + (entry) => + entry.blockExecutionId === undefined && + entry.blockId === span.blockId && + entry.executionId === executionId && + matchesConsoleIdentity(entry, { ...identity, blockExecutionId: undefined }) + ) + return legacyCandidates.length === 1 ? legacyCandidates[0] : undefined } function matchesConsoleIdentity(entry: ConsoleEntry, identity: ConsoleUpdate): boolean { + if (identity.blockExecutionId !== undefined) { + return entry.blockExecutionId === identity.blockExecutionId + } if (identity.executionOrder !== undefined && entry.executionOrder !== identity.executionOrder) { return false } @@ -730,6 +778,15 @@ function matchesConsoleIdentity(entry: ConsoleEntry, identity: ConsoleUpdate): b ) { return false } + if (identity.iterationType !== undefined && entry.iterationType !== identity.iterationType) { + return false + } + if ( + identity.parentIterations !== undefined && + JSON.stringify(entry.parentIterations ?? []) !== JSON.stringify(identity.parentIterations) + ) { + return false + } if ( identity.childWorkflowBlockId !== undefined && entry.childWorkflowBlockId !== identity.childWorkflowBlockId @@ -746,6 +803,30 @@ function matchesConsoleIdentity(entry: ConsoleEntry, identity: ConsoleUpdate): b return true } +function matchesFinalBlockLogIdentity( + entry: ConsoleEntry, + log: SecretSafeBlockLog, + executionId: string +): boolean { + const iterationContainerId = log.loopId ?? log.parallelId + const iterationType = log.loopId ? 'loop' : log.parallelId ? 'parallel' : undefined + const childWorkflowInstanceId = + typeof log.output?._childWorkflowInstanceId === 'string' + ? log.output._childWorkflowInstanceId + : undefined + return ( + entry.blockId === log.blockId && + entry.executionId === executionId && + entry.executionOrder === log.executionOrder && + entry.iterationCurrent === log.iterationIndex && + entry.iterationType === iterationType && + entry.iterationContainerId === iterationContainerId && + JSON.stringify(entry.parentIterations ?? []) === JSON.stringify(log.parentIterations ?? []) && + (childWorkflowInstanceId === undefined || + entry.childWorkflowInstanceId === childWorkflowInstanceId) + ) +} + function normalizeDisplayError(error: unknown): string | undefined { if (error === undefined || error === null) return undefined const message = typeof error === 'string' ? error : toError(error).message @@ -951,7 +1032,7 @@ interface WorkflowExecutionOptions { workflowInput?: any onStream?: (se: StreamingExecution) => Promise executionId?: string - onBlockComplete?: (blockId: string, output: any) => Promise + onBlockComplete?: (blockId: string, output: any, blockExecutionId?: string) => Promise overrideTriggerType?: 'chat' | 'manual' | 'api' | 'copilot' | 'webhook' | 'schedule' triggerBlockId?: string useDraftState?: boolean diff --git a/apps/sim/executor/execution/block-executor.retry.test.ts b/apps/sim/executor/execution/block-executor.retry.test.ts index 54811c71f8f..cd6b6c169b6 100644 --- a/apps/sim/executor/execution/block-executor.retry.test.ts +++ b/apps/sim/executor/execution/block-executor.retry.test.ts @@ -9,6 +9,7 @@ import { BlockType, EDGE } from '@/executor/constants' import type { DAGNode } from '@/executor/dag/builder' import { BlockExecutor } from '@/executor/execution/block-executor' import { ExecutionState } from '@/executor/execution/state' +import type { ContextExtensions } from '@/executor/execution/types' import type { BlockHandler, ExecutionContext } from '@/executor/types' import { VariableResolver } from '@/executor/variables/resolver' import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types' @@ -62,7 +63,12 @@ function createNode(block: SerializedBlock, withErrorPort = false): DAGNode { } as unknown as DAGNode } -function buildExecutor(block: SerializedBlock, handler: BlockHandler, state: ExecutionState) { +function buildExecutor( + block: SerializedBlock, + handler: BlockHandler, + state: ExecutionState, + callbacks: Pick = {} +) { const workflow: SerializedWorkflow = { version: '1', blocks: [block], @@ -87,6 +93,7 @@ function buildExecutor(block: SerializedBlock, handler: BlockHandler, state: Exe useDraftState: false, startTime: new Date().toISOString(), }, + ...callbacks, }, state ) @@ -137,6 +144,50 @@ describe('BlockExecutor retry', () => { expect(ctx.blockLogs[0]?.tries).toBe(2) }) + it('keeps one invocation ID across retries and mints a new ID for the next invocation', async () => { + const block = createBlock(enabled) + const handlerInvocationIds: Array = [] + const execute: BlockHandler['execute'] = vi.fn(async (_ctx, _block, _inputs, nodeMetadata) => { + handlerInvocationIds.push(nodeMetadata?.blockExecutionId) + if (handlerInvocationIds.length === 1) throw new Error('retry me') + return { ok: true } + }) + const onBlockStart = vi.fn(async () => {}) + const onBlockComplete = vi.fn(async () => {}) + const state = new ExecutionState() + const ctx = createContext(state) + const executor = buildExecutor(block, { canHandle: () => true, execute }, state, { + onBlockStart, + onBlockComplete, + }) + + await executor.execute(ctx, createNode(block), block) + await executor.execute(ctx, createNode(block), block) + await vi.waitFor(() => expect(onBlockComplete).toHaveBeenCalledTimes(2)) + + const [firstLog, secondLog] = ctx.blockLogs + expect(firstLog.blockExecutionId).toBeTypeOf('string') + expect(secondLog.blockExecutionId).toBeTypeOf('string') + expect(secondLog.blockExecutionId).not.toBe(firstLog.blockExecutionId) + expect(handlerInvocationIds).toEqual([ + firstLog.blockExecutionId, + firstLog.blockExecutionId, + secondLog.blockExecutionId, + ]) + expect(onBlockStart.mock.calls.map((call) => call[6])).toEqual([ + firstLog.blockExecutionId, + secondLog.blockExecutionId, + ]) + expect(onBlockComplete.mock.calls.map((call) => call[6])).toEqual([ + firstLog.blockExecutionId, + secondLog.blockExecutionId, + ]) + expect(onBlockComplete.mock.calls.map((call) => call[3]?.blockExecutionId)).toEqual([ + firstLog.blockExecutionId, + secondLog.blockExecutionId, + ]) + }) + it('stops at maxTries and rethrows the final error unchanged', async () => { const block = createBlock(enabled) const failure = new Error('still failing') diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index c0758935e81..9a1efcc0941 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/apps/sim/executor/execution/block-executor.test.ts @@ -718,20 +718,25 @@ describe('BlockExecutor', () => { state ) - await executor.execute(createContext(state), createNode(block), block) + const ctx = createContext(state) + await executor.execute(ctx, createNode(block), block) + const blockExecutionId = ctx.blockLogs[0]?.blockExecutionId - expect(onBlockStart).toHaveBeenCalled() + expect(blockExecutionId).toBeTypeOf('string') + expect(onBlockStart.mock.calls[0]?.[6]).toBe(blockExecutionId) expect(onBlockComplete).toHaveBeenCalledWith( block.id, 'Human in the Loop', BlockType.HUMAN_IN_THE_LOOP, expect.objectContaining({ + blockExecutionId, output: expect.objectContaining({ response: { status: 'paused' }, }), }), undefined, - undefined + undefined, + blockExecutionId ) expect(state.getBlockOutput(block.id)).toEqual(output) }) diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index bc4241208fd..14e37c87c40 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -1,5 +1,6 @@ import { createLogger, type Logger } from '@sim/logger' import { sleep } from '@sim/utils/helpers' +import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types' import { redactApiKeys } from '@/lib/core/security/redaction' @@ -146,7 +147,8 @@ export class BlockExecutor { blockCtx, node, block, - blockLog.executionOrder + blockLog.executionOrder, + blockLog.blockExecutionId ) await blockStartPromise } @@ -157,6 +159,7 @@ export class BlockExecutor { const nodeMetadata = { ...this.buildNodeMetadata(node), executionOrder: blockLog?.executionOrder, + blockExecutionId: blockLog?.blockExecutionId, } let cleanupSelfReference: (() => void) | undefined @@ -258,6 +261,7 @@ export class BlockExecutor { node, block, streamingExec, + blockLog?.blockExecutionId, resolvedInputs, normalizeStringArray(blockCtx.selectedOutputs) ) @@ -396,7 +400,8 @@ export class BlockExecutor { blockLog.endedAt, childWorkflowInstanceId, stateProvenance, - displayProvenance + displayProvenance, + blockLog.blockExecutionId ) } @@ -613,7 +618,8 @@ export class BlockExecutor { blockLog.endedAt, undefined, softOutputProvenance, - displayProvenance + displayProvenance, + blockLog.blockExecutionId ) } @@ -727,7 +733,8 @@ export class BlockExecutor { blockLog.endedAt, childWorkflowInstanceId, errorOutputProvenance, - displayProvenance + displayProvenance, + blockLog.blockExecutionId ) } @@ -809,6 +816,7 @@ export class BlockExecutor { return { blockId, + blockExecutionId: generateId(), blockName, blockType: block.metadata?.id ?? DEFAULTS.BLOCK_TYPE, startedAt, @@ -926,7 +934,8 @@ export class BlockExecutor { ctx: ExecutionContext, node: DAGNode, block: SerializedBlock, - executionOrder: number + executionOrder: number, + blockExecutionId?: string ): Promise | undefined { if (!this.contextExtensions.onBlockStart) return undefined @@ -942,7 +951,8 @@ export class BlockExecutor { blockType, executionOrder, iterationContext, - ctx.childWorkflowContext + ctx.childWorkflowContext, + blockExecutionId ) .catch((error) => { this.execLogger.warn('Block start callback failed', { @@ -971,7 +981,8 @@ export class BlockExecutor { endedAt: string, childWorkflowInstanceId?: string, resolvedSecretTraceProvenance?: ResolvedSecretTraceProvenanceV1, - displayResolvedSecretTraceProvenance?: ResolvedSecretTraceProvenanceV1 + displayResolvedSecretTraceProvenance?: ResolvedSecretTraceProvenanceV1, + blockExecutionId?: string ): void { if (!this.contextExtensions.onBlockComplete) return @@ -996,9 +1007,11 @@ export class BlockExecutor { executionOrder, endedAt, childWorkflowInstanceId, + blockExecutionId, }, iterationContext, - ctx.childWorkflowContext + ctx.childWorkflowContext, + blockExecutionId ) })().catch((error) => { this.execLogger.warn('Block completion callback failed', { @@ -1084,6 +1097,7 @@ export class BlockExecutor { node: DAGNode, block: SerializedBlock, streamingExec: StreamingExecution, + blockExecutionId: string | undefined, resolvedInputs: Record, selectedOutputs: string[] ): Promise { @@ -1137,6 +1151,7 @@ export class BlockExecutor { onStreamPromise = ctx .onStream({ ...streamingExecutionForConsumer, + blockExecutionId, stream: processedClientStream, streamFormat: 'text', subscribe: pump.subscribe, diff --git a/apps/sim/executor/execution/types.ts b/apps/sim/executor/execution/types.ts index 13aeba2664b..f57aa4d576d 100644 --- a/apps/sim/executor/execution/types.ts +++ b/apps/sim/executor/execution/types.ts @@ -161,6 +161,8 @@ export interface WorkflowNodeMetadata 'subflowType' | 'subflowId' | 'branchIndex' | 'branchTotal' | 'originalBlockId' | 'isLoopNode' > { nodeId: string + /** Unique identity for this individual block invocation. */ + blockExecutionId?: string loopId?: string parallelId?: string executionOrder?: number @@ -193,6 +195,8 @@ export interface BlockCompletionCallbackData { endedAt: string /** Per-invocation unique ID linking this workflow block execution to its child block events. */ childWorkflowInstanceId?: string + /** Unique identity for this individual block invocation. */ + blockExecutionId?: string } export interface ExecutionCallbacks { @@ -203,7 +207,8 @@ export interface ExecutionCallbacks { blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise onBlockComplete?: ( blockId: string, @@ -211,7 +216,8 @@ export interface ExecutionCallbacks { blockType: string, output: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise /** Fires immediately after instanceId is generated, before child execution begins. */ onChildWorkflowInstanceReady?: ( @@ -219,7 +225,8 @@ export interface ExecutionCallbacks { childWorkflowInstanceId: string, iterationContext?: IterationContext, executionOrder?: number, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise } @@ -296,7 +303,8 @@ export interface ContextExtensions { blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise onBlockComplete?: ( blockId: string, @@ -304,7 +312,8 @@ export interface ContextExtensions { blockType: string, output: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise /** Context identifying this execution as a child of a workflow block */ @@ -316,7 +325,8 @@ export interface ContextExtensions { childWorkflowInstanceId: string, iterationContext?: IterationContext, executionOrder?: number, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise /** diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.ts b/apps/sim/executor/handlers/workflow/workflow-handler.ts index f211567974c..46718623180 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.ts @@ -466,7 +466,8 @@ export class WorkflowBlockHandler implements BlockHandler { instanceId, iterationContext, nodeMetadata?.executionOrder, - ctx.childWorkflowContext + ctx.childWorkflowContext, + nodeMetadata?.blockExecutionId ) } @@ -666,7 +667,8 @@ export class WorkflowBlockHandler implements BlockHandler { blockType, executionOrder, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) => { if (activeSession && emitsSessionMarkers) { try { @@ -687,7 +689,8 @@ export class WorkflowBlockHandler implements BlockHandler { blockType, executionOrder, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) } } @@ -697,7 +700,8 @@ export class WorkflowBlockHandler implements BlockHandler { blockType, output, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) => { if (activeSession && emitsSessionMarkers) { try { @@ -713,7 +717,8 @@ export class WorkflowBlockHandler implements BlockHandler { blockType, output, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) } } diff --git a/apps/sim/executor/types.ts b/apps/sim/executor/types.ts index 47a0613fe6a..a5500d83757 100644 --- a/apps/sim/executor/types.ts +++ b/apps/sim/executor/types.ts @@ -261,6 +261,8 @@ export interface StartBlockRunMetadata { export interface BlockLog { blockId: string + /** Unique identity for this individual block invocation. */ + blockExecutionId?: string blockName?: string blockType?: string startedAt: string @@ -484,7 +486,8 @@ export interface ExecutionContext { blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise onBlockComplete?: ( blockId: string, @@ -492,7 +495,8 @@ export interface ExecutionContext { blockType: string, output: any, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise /** Context identifying this execution as a child of a workflow block */ @@ -504,7 +508,8 @@ export interface ExecutionContext { childWorkflowInstanceId: string, iterationContext?: IterationContext, executionOrder?: number, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => Promise /** @@ -607,6 +612,8 @@ export interface ExecutionResult { } export interface StreamingExecution { + /** Unique identity for the block invocation that owns this stream. */ + blockExecutionId?: string /** * Provider stream payload. Format is declared by {@link streamFormat}: * - `'text'` (default): UTF-8 answer bytes (`ReadableStream`) @@ -667,6 +674,8 @@ interface BlockExecutor { */ export interface BlockNodeMetadata { nodeId: string + /** Unique identity for this individual block invocation. */ + blockExecutionId?: string loopId?: string parallelId?: string branchIndex?: number diff --git a/apps/sim/executor/utils/subflow-utils.test.ts b/apps/sim/executor/utils/subflow-utils.test.ts index 60f657d9bc6..b106dfe042a 100644 --- a/apps/sim/executor/utils/subflow-utils.test.ts +++ b/apps/sim/executor/utils/subflow-utils.test.ts @@ -8,8 +8,13 @@ import { LARGE_ARRAY_MANIFEST_VERSION, } from '@/lib/execution/payloads/large-array-manifest-metadata' import { LARGE_VALUE_REF_MARKER } from '@/lib/execution/payloads/large-value-ref' +import type { ContextExtensions } from '@/executor/execution/types' import type { ExecutionContext } from '@/executor/types' -import { findEffectiveContainerId } from '@/executor/utils/subflow-utils' +import { + addSubflowErrorLog, + emitSubflowSuccessEvents, + findEffectiveContainerId, +} from '@/executor/utils/subflow-utils' import { resolveArrayInputAsync } from '@/executor/utils/subflow-utils.server' import type { VariableResolver } from '@/executor/variables/resolver' @@ -241,3 +246,36 @@ describe('findEffectiveContainerId', () => { ).toBe('inner-parallel__clone3__obranch-2') }) }) + +describe('subflow invocation identity', () => { + it('mints one stable, distinct ID for each synthetic subflow log and callback', async () => { + const ctx = { + blockLogs: [], + blockStates: new Map(), + workflow: { + blocks: [ + { id: 'loop-1', metadata: { name: 'Loop' } }, + { id: 'parallel-1', metadata: { name: 'Parallel' } }, + ], + }, + } as unknown as ExecutionContext + const onBlockStart = vi.fn( + async (..._args: Parameters>) => {} + ) + const onBlockComplete = vi.fn( + async (..._args: Parameters>) => {} + ) + const callbacks = { onBlockStart, onBlockComplete } as ContextExtensions + + await addSubflowErrorLog(ctx, 'loop-1', 'loop', 'failed', {}, callbacks) + await addSubflowErrorLog(ctx, 'loop-1', 'loop', 'failed again', {}, callbacks) + await emitSubflowSuccessEvents(ctx, 'parallel-1', 'parallel', { results: [] }, callbacks) + + const ids = ctx.blockLogs.map((log) => log.blockExecutionId) + expect(ids.every((id) => typeof id === 'string')).toBe(true) + expect(new Set(ids).size).toBe(3) + expect(onBlockStart.mock.calls.map((call) => call[6])).toEqual(ids.slice(0, 2)) + expect(onBlockComplete.mock.calls.map((call) => call[3].blockExecutionId)).toEqual(ids) + expect(onBlockComplete.mock.calls.map((call) => call[6])).toEqual(ids) + }) +}) diff --git a/apps/sim/executor/utils/subflow-utils.ts b/apps/sim/executor/utils/subflow-utils.ts index 0dc00d93466..a4f9b47f676 100644 --- a/apps/sim/executor/utils/subflow-utils.ts +++ b/apps/sim/executor/utils/subflow-utils.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' +import { generateId } from '@sim/utils/id' import { DEFAULTS } from '@/executor/constants' import type { ContextExtensions } from '@/executor/execution/types' import { type BlockLog, type ExecutionContext, getNextExecutionOrder } from '@/executor/types' @@ -222,12 +223,14 @@ export async function addSubflowErrorLog( ): Promise { const now = new Date().toISOString() const execOrder = getNextExecutionOrder(ctx) + const blockExecutionId = generateId() const block = ctx.workflow?.blocks?.find((b) => b.id === blockId) const blockName = block?.metadata?.name || (blockType === 'loop' ? 'Loop' : 'Parallel') const blockLog: BlockLog = { blockId, + blockExecutionId, blockName, blockType, startedAt: now, @@ -244,7 +247,15 @@ export async function addSubflowErrorLog( if (contextExtensions?.onBlockStart) { try { - await contextExtensions.onBlockStart(blockId, blockName, blockType, execOrder) + await contextExtensions.onBlockStart( + blockId, + blockName, + blockType, + execOrder, + undefined, + undefined, + blockExecutionId + ) } catch (error) { logger.warn('Subflow error start callback failed', { blockId, @@ -256,14 +267,23 @@ export async function addSubflowErrorLog( if (contextExtensions?.onBlockComplete) { try { - await contextExtensions.onBlockComplete(blockId, blockName, blockType, { - input: inputData, - output: { error: errorMessage }, - executionTime: 0, - startedAt: now, - executionOrder: execOrder, - endedAt: now, - }) + await contextExtensions.onBlockComplete( + blockId, + blockName, + blockType, + { + input: inputData, + output: { error: errorMessage }, + executionTime: 0, + startedAt: now, + executionOrder: execOrder, + endedAt: now, + blockExecutionId, + }, + undefined, + undefined, + blockExecutionId + ) } catch (error) { logger.warn('Subflow error completion callback failed', { blockId, @@ -290,6 +310,7 @@ export async function emitSubflowSuccessEvents( ): Promise { const now = new Date().toISOString() const executionOrder = getNextExecutionOrder(ctx) + const blockExecutionId = generateId() const block = ctx.workflow?.blocks.find((b) => b.id === blockId) const blockName = block?.metadata?.name ?? blockType const iterationContext = buildContainerIterationContext(ctx, blockId) @@ -297,6 +318,7 @@ export async function emitSubflowSuccessEvents( ctx.blockLogs.push({ blockId, + blockExecutionId, blockName, blockType, startedAt: now, @@ -322,8 +344,11 @@ export async function emitSubflowSuccessEvents( startedAt: now, executionOrder, endedAt: now, + blockExecutionId, }, - iterationContext + iterationContext, + undefined, + blockExecutionId ) } catch (error) { logger.warn('Subflow success completion callback failed', { diff --git a/apps/sim/lib/logs/execution/trace-spans/span-factory.ts b/apps/sim/lib/logs/execution/trace-spans/span-factory.ts index e1b4a837c8d..f8e014ea596 100644 --- a/apps/sim/lib/logs/execution/trace-spans/span-factory.ts +++ b/apps/sim/lib/logs/execution/trace-spans/span-factory.ts @@ -107,6 +107,7 @@ function createBaseSpan(log: ValidBlockLog): TraceSpan { status: log.error ? 'error' : 'success', children: [], blockId: log.blockId, + ...(log.blockExecutionId && { blockExecutionId: log.blockExecutionId }), executionOrder: log.executionOrder, input: log.input, output, diff --git a/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts b/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts index 09a0c25858c..d13002d58e8 100644 --- a/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts +++ b/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts @@ -9,6 +9,32 @@ import { stripCustomToolPrefix } from '@/executor/constants' import type { ExecutionResult } from '@/executor/types' describe('buildTraceSpans', () => { + it('preserves block invocation identity on trace spans', () => { + const { traceSpans } = buildTraceSpans({ + success: true, + output: {}, + logs: [ + { + blockId: 'function-1', + blockExecutionId: 'invoke-1', + blockName: 'Function', + blockType: 'function', + startedAt: '2026-08-22T10:00:00.000Z', + endedAt: '2026-08-22T10:00:00.010Z', + durationMs: 10, + success: true, + executionOrder: 1, + output: { ok: true }, + }, + ], + }) + + expect(traceSpans[0]).toMatchObject({ + blockId: 'function-1', + blockExecutionId: 'invoke-1', + }) + }) + it.concurrent('extracts sequential segments from timeSegments data', () => { const mockExecutionResult: ExecutionResult = { success: true, diff --git a/apps/sim/lib/logs/types.ts b/apps/sim/lib/logs/types.ts index eb25b8823ec..f2d28b6b249 100644 --- a/apps/sim/lib/logs/types.ts +++ b/apps/sim/lib/logs/types.ts @@ -282,6 +282,8 @@ export interface TraceSpan { tokens?: TokenInfo relativeStartMs?: number blockId?: string + /** Unique identity for the block invocation represented by this span. */ + blockExecutionId?: string executionOrder?: number input?: Record output?: Record diff --git a/apps/sim/lib/tokenization/streaming.ts b/apps/sim/lib/tokenization/streaming.ts index ca552fa8292..d3e08ac5811 100644 --- a/apps/sim/lib/tokenization/streaming.ts +++ b/apps/sim/lib/tokenization/streaming.ts @@ -133,7 +133,7 @@ export function processStreamingBlockLogs( let processedCount = 0 for (const log of logs) { - const content = streamedContentMap.get(log.blockId) + const content = streamedContentMap.get(log.blockExecutionId ?? log.blockId) if (content && processStreamingBlockLog(log, content)) { processedCount++ } diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index ad20d6d3807..42cb1566476 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -33,9 +33,10 @@ export interface ExecuteWorkflowOptions { blockId: string, blockName: string, blockType: string, - executionOrder: number + executionOrder: number, + blockExecutionId?: string ) => Promise - onBlockComplete?: (blockId: string, output: unknown) => Promise + onBlockComplete?: (blockId: string, output: unknown, blockExecutionId?: string) => Promise /** Transfers post-execution logging ownership to the streaming caller after execution succeeds. */ skipLoggingComplete?: boolean includeFileBase64?: boolean @@ -170,14 +171,31 @@ export async function executeWorkflow( blockId: string, blockName: string, blockType: string, - executionOrder: number + executionOrder: number, + _iterationContext, + _childWorkflowContext, + blockExecutionId ) => { - await streamConfig.onBlockStart!(blockId, blockName, blockType, executionOrder) + await streamConfig.onBlockStart!( + blockId, + blockName, + blockType, + executionOrder, + blockExecutionId + ) } : undefined, onBlockComplete: streamConfig?.onBlockComplete - ? async (blockId: string, _blockName: string, _blockType: string, output: unknown) => { - await streamConfig.onBlockComplete!(blockId, output) + ? async ( + blockId: string, + _blockName: string, + _blockType: string, + output, + _iterationContext, + _childWorkflowContext, + blockExecutionId + ) => { + await streamConfig.onBlockComplete!(blockId, output, blockExecutionId) } : undefined, }, diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts index 81295f679d5..e29da73b997 100644 --- a/apps/sim/lib/workflows/executor/execution-core.ts +++ b/apps/sim/lib/workflows/executor/execution-core.ts @@ -679,7 +679,8 @@ async function executeWorkflowCoreImpl( blockType: string, output: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { let persistenceSucceeded = false const persistencePromise = (async () => { @@ -707,7 +708,8 @@ async function executeWorkflowCoreImpl( blockType, output, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) } catch (error) { logger.warn( @@ -731,7 +733,8 @@ async function executeWorkflowCoreImpl( blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { let persistenceSucceeded = false const persistencePromise = (async () => { @@ -759,7 +762,8 @@ async function executeWorkflowCoreImpl( blockType, executionOrder, iterationContext, - childWorkflowContext + childWorkflowContext, + blockExecutionId ) } catch (error) { logger.warn( diff --git a/apps/sim/lib/workflows/executor/execution-events.ts b/apps/sim/lib/workflows/executor/execution-events.ts index e13003fe1f8..503ca6962ed 100644 --- a/apps/sim/lib/workflows/executor/execution-events.ts +++ b/apps/sim/lib/workflows/executor/execution-events.ts @@ -145,6 +145,7 @@ interface BlockStartedEvent extends BaseExecutionEvent { parentIterations?: ParentIteration[] childWorkflowBlockId?: string childWorkflowName?: string + blockExecutionId?: string } } @@ -174,6 +175,7 @@ interface BlockCompletedEvent extends BaseExecutionEvent { childWorkflowName?: string /** Per-invocation unique ID for correlating child block events with this workflow block. */ childWorkflowInstanceId?: string + blockExecutionId?: string } } @@ -203,6 +205,7 @@ interface BlockErrorEvent extends BaseExecutionEvent { childWorkflowName?: string /** Per-invocation unique ID for correlating child block events with this workflow block. */ childWorkflowInstanceId?: string + blockExecutionId?: string } } @@ -225,6 +228,7 @@ interface BlockChildWorkflowStartedEvent extends BaseExecutionEvent { childWorkflowBlockId?: string childWorkflowName?: string executionOrder?: number + blockExecutionId?: string } } @@ -236,6 +240,7 @@ interface StreamChunkEvent extends BaseExecutionEvent { workflowId: string data: { blockId: string + blockExecutionId?: string chunk: string display?: ExecutionEventDisplayData } @@ -252,6 +257,7 @@ interface StreamChunkResetEvent extends BaseExecutionEvent { workflowId: string data: { blockId: string + blockExecutionId?: string } } @@ -263,6 +269,7 @@ interface StreamDoneEvent extends BaseExecutionEvent { workflowId: string data: { blockId: string + blockExecutionId?: string } } @@ -276,6 +283,7 @@ interface StreamThinkingEvent extends BaseExecutionEvent { workflowId: string data: { blockId: string + blockExecutionId?: string text: string display?: ExecutionEventDisplayData } @@ -290,6 +298,7 @@ interface StreamToolEvent extends BaseExecutionEvent { workflowId: string data: { blockId: string + blockExecutionId?: string phase: 'start' | 'end' id: string name: string @@ -365,7 +374,8 @@ export function createExecutionCallbacks(options: { blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { await sendBufferedEvent({ type: 'block:started', @@ -390,6 +400,7 @@ export function createExecutionCallbacks(options: { childWorkflowBlockId: childWorkflowContext.parentBlockId, childWorkflowName: childWorkflowContext.workflowName, }), + ...(blockExecutionId && { blockExecutionId }), }, }) } @@ -400,7 +411,8 @@ export function createExecutionCallbacks(options: { blockType: string, callbackData: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { const callbackError = callbackData.output?.error const iterationData = iterationContext @@ -424,6 +436,10 @@ export function createExecutionCallbacks(options: { const instanceData = callbackData.childWorkflowInstanceId ? { childWorkflowInstanceId: callbackData.childWorkflowInstanceId } : {} + const invocationData = + blockExecutionId || callbackData.blockExecutionId + ? { blockExecutionId: blockExecutionId ?? callbackData.blockExecutionId } + : {} if (callbackError) { await sendBufferedEvent({ type: 'block:error', @@ -443,6 +459,7 @@ export function createExecutionCallbacks(options: { ...iterationData, ...childWorkflowData, ...instanceData, + ...invocationData, }, }) } else { @@ -464,14 +481,20 @@ export function createExecutionCallbacks(options: { ...iterationData, ...childWorkflowData, ...instanceData, + ...invocationData, }, }) } } const onStream = async (streamingExecution: unknown) => { - const streamingExec = streamingExecution as { stream: ReadableStream; execution: any } + const streamingExec = streamingExecution as { + stream: ReadableStream + execution: any + blockExecutionId?: string + } const blockId = streamingExec.execution?.blockId + const { blockExecutionId } = streamingExec const reader = streamingExec.stream.getReader() const decoder = new TextDecoder() @@ -485,7 +508,7 @@ export function createExecutionCallbacks(options: { timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId, chunk }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }), chunk }, }) } await sendBufferedEvent({ @@ -493,7 +516,7 @@ export function createExecutionCallbacks(options: { timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }) }, }) } finally { try { @@ -507,7 +530,8 @@ export function createExecutionCallbacks(options: { childWorkflowInstanceId: string, iterationContext?: IterationContext, executionOrder?: number, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { await sendBufferedEvent({ type: 'block:childWorkflowStarted', @@ -531,6 +555,7 @@ export function createExecutionCallbacks(options: { childWorkflowName: childWorkflowContext.workflowName, }), ...(executionOrder !== undefined && { executionOrder }), + ...(blockExecutionId && { blockExecutionId }), }, }) } diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 2dfef26f5d4..2b9e901a931 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -390,7 +390,7 @@ interface StartResumeExecutionArgs { userId: string sendEvent?: (event: ExecutionEvent) => void onStream?: (streamingExec: StreamingExecution) => Promise - onBlockComplete?: (blockId: string, output: unknown) => Promise + onBlockComplete?: (blockId: string, output: unknown, blockExecutionId?: string) => Promise abortSignal?: AbortSignal } @@ -1039,7 +1039,7 @@ export class PauseResumeManager { userId: string sendEvent?: (event: ExecutionEvent) => void onStream?: (streamingExec: StreamingExecution) => Promise - onBlockComplete?: (blockId: string, output: unknown) => Promise + onBlockComplete?: (blockId: string, output: unknown, blockExecutionId?: string) => Promise abortSignal?: AbortSignal }): Promise { const { @@ -1588,7 +1588,8 @@ export class PauseResumeManager { blockType: string, executionOrder: number, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { await writeBufferedEvent({ type: 'block:started', @@ -1613,6 +1614,7 @@ export class PauseResumeManager { childWorkflowBlockId: childWorkflowContext.parentBlockId, childWorkflowName: childWorkflowContext.workflowName, }), + ...(blockExecutionId && { blockExecutionId }), }, } as ExecutionEvent) }, @@ -1622,7 +1624,8 @@ export class PauseResumeManager { blockType: string, callbackData: BlockCompletionCallbackData, iterationContext?: IterationContext, - childWorkflowContext?: ChildWorkflowContext + childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { const output = callbackData.output as Record | undefined const hasError = output?.error @@ -1659,6 +1662,9 @@ export class PauseResumeManager { ...(callbackData.childWorkflowInstanceId ? { childWorkflowInstanceId: callbackData.childWorkflowInstanceId } : {}), + ...((blockExecutionId || callbackData.blockExecutionId) && { + blockExecutionId: blockExecutionId ?? callbackData.blockExecutionId, + }), } await writeBufferedEvent({ @@ -1688,14 +1694,20 @@ export class PauseResumeManager { } as ExecutionEvent) if (externalOnBlockComplete) { - await externalOnBlockComplete(blockId, callbackData.output) + await externalOnBlockComplete( + blockId, + callbackData.output, + blockExecutionId ?? callbackData.blockExecutionId + ) } }, onChildWorkflowInstanceReady: async ( blockId: string, childWorkflowInstanceId: string, iterationContext?: IterationContext, - executionOrder?: number + executionOrder?: number, + _childWorkflowContext?: ChildWorkflowContext, + blockExecutionId?: string ) => { await writeBufferedEvent({ type: 'block:childWorkflowStarted', @@ -1710,6 +1722,7 @@ export class PauseResumeManager { iterationContainerId: iterationContext.iterationContainerId, }), ...(executionOrder !== undefined && { executionOrder }), + ...(blockExecutionId && { blockExecutionId }), }, } as ExecutionEvent) }, @@ -1723,6 +1736,7 @@ export class PauseResumeManager { ? streamingExec.execution.blockId : undefined const blockId = typeof blockIdValue === 'string' ? blockIdValue : '' + const { blockExecutionId } = streamingExec // Live answer text rides the sink when available; the byte stream is // then drained without re-emitting chunks (same final-turn content). @@ -1730,6 +1744,7 @@ export class PauseResumeManager { const unsubscribe = forwardAgentStreamToExecutionEvents(streamingExec, { blockId, + blockExecutionId, executionId: resumeExecutionId, workflowId, sendEvent: writeBufferedEvent, @@ -1760,7 +1775,7 @@ export class PauseResumeManager { timestamp: new Date().toISOString(), executionId: resumeExecutionId, workflowId, - data: { blockId, chunk, display }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }), chunk, display }, } as ExecutionEvent) } await writeBufferedEvent({ @@ -1768,7 +1783,7 @@ export class PauseResumeManager { timestamp: new Date().toISOString(), executionId: resumeExecutionId, workflowId, - data: { blockId }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }) }, } as ExecutionEvent) } catch (streamError) { logger.error( diff --git a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts index efe186f3b05..dc79c591071 100644 --- a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts +++ b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.test.ts @@ -33,7 +33,13 @@ describe('forwardAgentStreamToExecutionEvents', () => { makeStreamingExec((handler) => { sinkHandler = handler }, unsubscribe), - { blockId: 'agent-1', executionId: 'exec-1', workflowId: 'wf-1', sendEvent } + { + blockId: 'agent-1', + blockExecutionId: 'invoke-1', + executionId: 'exec-1', + workflowId: 'wf-1', + sendEvent, + } ) expect(sinkHandler).toBeTypeOf('function') @@ -53,16 +59,29 @@ describe('forwardAgentStreamToExecutionEvents', () => { type: 'stream:thinking', executionId: 'exec-1', workflowId: 'wf-1', - data: { blockId: 'agent-1', text: 'plan ' }, + data: { blockId: 'agent-1', blockExecutionId: 'invoke-1', text: 'plan ' }, }) expect(sendEvent.mock.calls[0][0].data).not.toHaveProperty('display') expect(sendEvent.mock.calls[1][0]).toMatchObject({ type: 'stream:tool', - data: { blockId: 'agent-1', phase: 'start', id: 't1', name: 'http_request' }, + data: { + blockId: 'agent-1', + blockExecutionId: 'invoke-1', + phase: 'start', + id: 't1', + name: 'http_request', + }, }) expect(sendEvent.mock.calls[2][0]).toMatchObject({ type: 'stream:tool', - data: { blockId: 'agent-1', phase: 'end', id: 't1', name: 'http_request', status: 'success' }, + data: { + blockId: 'agent-1', + blockExecutionId: 'invoke-1', + phase: 'end', + id: 't1', + name: 'http_request', + status: 'success', + }, }) unsub() @@ -96,6 +115,7 @@ describe('forwardAgentStreamToExecutionEvents', () => { }), { blockId: 'agent-1', + blockExecutionId: 'invoke-1', executionId: 'exec-1', workflowId: 'wf-1', sendEvent, @@ -116,10 +136,20 @@ describe('forwardAgentStreamToExecutionEvents', () => { expect(calls).toEqual([ { type: 'stream:chunk', - data: { blockId: 'agent-1', chunk: 'Let me check…' }, + data: { + blockId: 'agent-1', + blockExecutionId: 'invoke-1', + chunk: 'Let me check…', + }, + }, + { + type: 'stream:chunk_reset', + data: { blockId: 'agent-1', blockExecutionId: 'invoke-1' }, + }, + { + type: 'stream:chunk', + data: { blockId: 'agent-1', blockExecutionId: 'invoke-1', chunk: 'Answer' }, }, - { type: 'stream:chunk_reset', data: { blockId: 'agent-1' } }, - { type: 'stream:chunk', data: { blockId: 'agent-1', chunk: 'Answer' } }, ]) }) diff --git a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts index 74ace5d93ae..0912010ea23 100644 --- a/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts +++ b/apps/sim/lib/workflows/streaming/forward-agent-stream-events.ts @@ -17,6 +17,7 @@ import type { StreamingExecution } from '@/executor/types' export interface ForwardAgentStreamEventsOptions { blockId: string + blockExecutionId?: string executionId: string workflowId: string sendEvent: (event: ExecutionEvent) => void | Promise @@ -72,6 +73,7 @@ export function forwardAgentStreamToExecutionEvents( const { blockId, + blockExecutionId, executionId, workflowId, sendEvent, @@ -91,6 +93,7 @@ export function forwardAgentStreamToExecutionEvents( workflowId, data: { blockId, + ...(blockExecutionId && { blockExecutionId }), text: event.text, ...(display !== undefined ? { display } : {}), }, @@ -103,7 +106,13 @@ export function forwardAgentStreamToExecutionEvents( timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId, phase: 'start', id: event.id, name: event.name }, + data: { + blockId, + ...(blockExecutionId && { blockExecutionId }), + phase: 'start', + id: event.id, + name: event.name, + }, }) return } @@ -113,7 +122,14 @@ export function forwardAgentStreamToExecutionEvents( timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId, phase: 'end', id: event.id, name: event.name, status: event.status }, + data: { + blockId, + ...(blockExecutionId && { blockExecutionId }), + phase: 'end', + id: event.id, + name: event.name, + status: event.status, + }, }) return } @@ -131,6 +147,7 @@ export function forwardAgentStreamToExecutionEvents( workflowId, data: { blockId, + ...(blockExecutionId && { blockExecutionId }), chunk: event.text, ...(display !== undefined ? { display } : {}), }, @@ -144,7 +161,7 @@ export function forwardAgentStreamToExecutionEvents( timestamp: new Date().toISOString(), executionId, workflowId, - data: { blockId }, + data: { blockId, ...(blockExecutionId && { blockExecutionId }) }, }) } }, diff --git a/apps/sim/lib/workflows/streaming/streaming.test.ts b/apps/sim/lib/workflows/streaming/streaming.test.ts index 5e5391358cc..d677f0e0d3e 100644 --- a/apps/sim/lib/workflows/streaming/streaming.test.ts +++ b/apps/sim/lib/workflows/streaming/streaming.test.ts @@ -1043,6 +1043,70 @@ describe('createStreamingResponse agent-events-v1', () => { expect(events.some((event) => event.event === 'final')).toBe(true) }) + it('suppresses repeated selected output when one invocation streams', async () => { + const stream = await createStreamingResponse({ + requestId: 'request-invocation-selected-output', + streamConfig: { + selectedOutputs: ['agent-1_content'], + }, + executeFn: async ({ onStream, onBlockComplete }) => { + await onStream({ + stream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('current answer')) + controller.close() + }, + }), + streamFormat: 'text', + blockExecutionId: 'invoke-current', + execution: { + blockId: 'agent-1', + success: true, + output: { content: 'current answer' }, + logs: [], + metadata: {}, + }, + } as any) + await onBlockComplete('agent-1', { content: 'current answer' }, 'invoke-current') + await onBlockComplete('agent-1', { content: 'later answer' }, 'invoke-later') + + return { + success: true, + output: { content: 'current answer' }, + logs: [ + { + blockId: 'agent-1', + blockExecutionId: 'invoke-later', + output: { content: 'later answer' }, + startedAt: new Date().toISOString(), + endedAt: new Date().toISOString(), + durationMs: 1, + success: true, + }, + { + blockId: 'agent-1', + blockExecutionId: 'invoke-current', + output: { content: '' }, + startedAt: new Date().toISOString(), + endedAt: new Date().toISOString(), + durationMs: 1, + success: true, + }, + ], + } as any + }, + }) + + const events = await collectSSEEvents(stream) + expect(events.filter((event) => event.chunk !== undefined)).toEqual([ + { blockId: 'agent-1', chunk: 'current answer' }, + ]) + expect(events.find((event) => event.event === 'final')).toEqual({ + event: 'final', + data: { success: true, output: {} }, + }) + }) + it('stays fully text-only when both policies are off', async () => { const stream = await createStreamingResponse({ requestId: 'request-1', diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts index c330388dd0b..3fce00db6a5 100644 --- a/apps/sim/lib/workflows/streaming/streaming.ts +++ b/apps/sim/lib/workflows/streaming/streaming.ts @@ -88,7 +88,7 @@ interface StreamingConfig { export type StreamingExecutorFn = (callbacks: { onStream: (streamingExec: StreamingExecution) => Promise - onBlockComplete: (blockId: string, output: unknown) => Promise + onBlockComplete: (blockId: string, output: unknown, blockExecutionId?: string) => Promise abortSignal: AbortSignal }) => Promise @@ -129,6 +129,7 @@ export function agentStreamProtocolResponseHeaders(options: { interface StreamingState { streamedChunks: Map + streamedBlockIds: Set processedOutputs: Set streamCompletionTimes: Map completedBlockIds: Set @@ -319,6 +320,7 @@ async function buildMinimalResult( result: ExecutionResult, selectedOutputs: string[] | undefined, streamedContent: Map, + streamedBlockIds: Set, completedBlockIds: Set, streamedSelectedOutputKeys: Set, requestId: string, @@ -388,7 +390,7 @@ async function buildMinimalResult( for (const descriptor of getSelectedOutputDescriptors(selectedOutputs)) { const { blockId, path } = descriptor - if (streamedContent.has(blockId)) { + if (streamedContent.has(blockId) || streamedBlockIds.has(blockId)) { continue } @@ -454,15 +456,16 @@ function updateLogsWithStreamedContent( streamCompletionTimes: Map ): BlockLog[] { return logs.map((log: BlockLog) => { - if (!streamedContent.has(log.blockId)) { + const streamKey = log.blockExecutionId ?? log.blockId + if (!streamedContent.has(streamKey)) { return log } - const content = streamedContent.get(log.blockId) + const content = streamedContent.get(streamKey) const updatedLog = { ...log } - if (streamCompletionTimes.has(log.blockId)) { - const completionTime = streamCompletionTimes.get(log.blockId)! + if (streamCompletionTimes.has(streamKey)) { + const completionTime = streamCompletionTimes.get(streamKey)! const startTime = new Date(log.startedAt).getTime() updatedLog.endedAt = new Date(completionTime).toISOString() updatedLog.durationMs = completionTime - startTime @@ -537,6 +540,7 @@ export async function createStreamingResponse( async start(controller) { const state: StreamingState = { streamedChunks: new Map(), + streamedBlockIds: new Set(), processedOutputs: new Set(), streamCompletionTimes: new Map(), completedBlockIds: new Set(), @@ -608,6 +612,7 @@ export async function createStreamingResponse( logger.warn(`[${requestId}] Streaming execution missing blockId`) return } + const streamKey = streamingExec.blockExecutionId ?? blockId /** * Negotiated clients get answer text live from the sink (pending deltas @@ -675,15 +680,16 @@ export async function createStreamingResponse( while (true) { const { done, value } = await reader.read() if (done) { - state.streamCompletionTimes.set(blockId, Date.now()) + state.streamCompletionTimes.set(streamKey, Date.now()) break } const textChunk = decoder.decode(value, { stream: true }) - if (!state.streamedChunks.has(blockId)) { - state.streamedChunks.set(blockId, []) + state.streamedBlockIds.add(blockId) + if (!state.streamedChunks.has(streamKey)) { + state.streamedChunks.set(streamKey, []) } - state.streamedChunks.get(blockId)!.push(textChunk) + state.streamedChunks.get(streamKey)!.push(textChunk) if (!sinkAnswerText) { emitAnswerChunk(textChunk) @@ -708,14 +714,18 @@ export async function createStreamingResponse( const includeFileBase64 = streamConfig.includeFileBase64 ?? true const base64MaxBytes = streamConfig.base64MaxBytes - const onBlockCompleteCallback = async (blockId: string, output: unknown) => { + const onBlockCompleteCallback = async ( + blockId: string, + output: unknown, + blockExecutionId?: string + ) => { state.completedBlockIds.add(blockId) if (!streamConfig.selectedOutputs?.length) { return } - if (state.streamedChunks.has(blockId)) { + if (state.streamedBlockIds.has(blockId)) { return } @@ -867,6 +877,7 @@ export async function createStreamingResponse( result, streamConfig.selectedOutputs, streamedContent, + state.streamedBlockIds, state.completedBlockIds, state.streamedSelectedOutputKeys, requestId, diff --git a/apps/sim/stores/index.ts b/apps/sim/stores/index.ts index 529cb891a55..a70a8e893f5 100644 --- a/apps/sim/stores/index.ts +++ b/apps/sim/stores/index.ts @@ -37,6 +37,7 @@ export const resetAllStores = () => { useTerminalConsoleStore.setState({ workflowEntries: {}, entryIdsByBlockExecution: {}, + entryIdByBlockExecutionId: {}, entryLocationById: {}, isOpen: false, }) diff --git a/apps/sim/stores/terminal/console/store.test.ts b/apps/sim/stores/terminal/console/store.test.ts index 4655bd1943b..2ffcad1c672 100644 --- a/apps/sim/stores/terminal/console/store.test.ts +++ b/apps/sim/stores/terminal/console/store.test.ts @@ -1,6 +1,7 @@ /** * @vitest-environment node */ +import { createLogger } from '@sim/logger' import { beforeEach, describe, expect, it, vi } from 'vitest' const { mockSaveBlob } = vi.hoisted(() => ({ @@ -16,12 +17,19 @@ vi.unmock('@/stores/terminal/console/store') import { useTerminalConsoleStore } from '@/stores/terminal/console/store' +const storeLoggerCallIndex = vi + .mocked(createLogger) + .mock.calls.findIndex(([name]) => name === 'TerminalConsoleStore') +const storeLogger = vi.mocked(createLogger).mock.results[storeLoggerCallIndex]?.value + describe('terminal console store', () => { beforeEach(() => { mockSaveBlob.mockClear() + storeLogger?.warn.mockClear() useTerminalConsoleStore.setState({ workflowEntries: {}, entryIdsByBlockExecution: {}, + entryIdByBlockExecutionId: {}, entryLocationById: {}, isOpen: false, _hasHydrated: true, @@ -145,6 +153,172 @@ describe('terminal console store', () => { expect(after.getWorkflowEntries('wf-1')[0].output).toMatchObject({ status: 'updated' }) }) + describe('per-invocation attribution', () => { + it('isolates colliding iteration updates by blockExecutionId', () => { + const addConsole = useTerminalConsoleStore.getState().addConsole + for (const blockExecutionId of ['invoke-0', 'invoke-1']) { + addConsole({ + workflowId: 'wf-1', + blockId: 'function-1', + blockExecutionId, + blockName: 'Function', + blockType: 'function', + executionId: 'exec-1', + executionOrder: 7, + iterationCurrent: 0, + iterationType: 'loop', + iterationContainerId: 'loop-1', + isRunning: true, + }) + } + + useTerminalConsoleStore.getState().updateConsole( + 'function-1', + { + blockExecutionId: 'invoke-0', + blockName: 'Function success', + replaceOutput: { iteration: 0 }, + success: true, + startedAt: '2026-08-22T10:00:00.000Z', + endedAt: '2026-08-22T10:00:00.010Z', + durationMs: 10, + isRunning: false, + }, + 'exec-1' + ) + useTerminalConsoleStore.getState().updateConsole( + 'function-1', + { + blockExecutionId: 'invoke-1', + blockName: 'Function failure', + replaceOutput: {}, + error: 'iteration-1-failure', + success: false, + startedAt: '2026-08-22T10:00:01.000Z', + endedAt: '2026-08-22T10:00:01.020Z', + durationMs: 20, + isRunning: false, + }, + 'exec-1' + ) + + const entries = useTerminalConsoleStore.getState().getWorkflowEntries('wf-1') + expect(entries.find((entry) => entry.blockExecutionId === 'invoke-0')).toMatchObject({ + blockName: 'Function success', + output: { iteration: 0 }, + success: true, + error: undefined, + durationMs: 10, + startedAt: '2026-08-22T10:00:00.000Z', + endedAt: '2026-08-22T10:00:00.010Z', + isRunning: false, + }) + expect(entries.find((entry) => entry.blockExecutionId === 'invoke-1')).toMatchObject({ + blockName: 'Function failure', + output: {}, + success: false, + error: 'iteration-1-failure', + durationMs: 20, + startedAt: '2026-08-22T10:00:01.000Z', + endedAt: '2026-08-22T10:00:01.020Z', + isRunning: false, + }) + }) + + it('treats a replayed start as idempotent', () => { + const start = { + workflowId: 'wf-1', + blockId: 'function-1', + blockExecutionId: 'invoke-0', + blockName: 'Function', + blockType: 'function', + executionId: 'exec-1', + executionOrder: 1, + isRunning: true, + } + + const first = useTerminalConsoleStore.getState().addConsole(start) + const replay = useTerminalConsoleStore.getState().addConsole(start) + + expect(replay?.id).toBe(first?.id) + expect(useTerminalConsoleStore.getState().getWorkflowEntries('wf-1')).toHaveLength(1) + }) + + it('warns and leaves ambiguous legacy entries unchanged', () => { + const addConsole = useTerminalConsoleStore.getState().addConsole + for (const blockName of ['First', 'Second']) { + addConsole({ + workflowId: 'wf-1', + blockId: 'function-1', + blockName, + blockType: 'function', + executionId: 'exec-1', + executionOrder: 1, + iterationCurrent: 0, + iterationType: 'loop', + iterationContainerId: 'loop-1', + isRunning: true, + }) + } + + useTerminalConsoleStore.getState().updateConsole( + 'function-1', + { + executionOrder: 1, + iterationCurrent: 0, + iterationType: 'loop', + iterationContainerId: 'loop-1', + replaceOutput: { overwritten: true }, + success: true, + }, + 'exec-1' + ) + + expect( + useTerminalConsoleStore + .getState() + .getWorkflowEntries('wf-1') + .every((entry) => entry.output === undefined && entry.success === undefined) + ).toBe(true) + expect(storeLogger?.warn).toHaveBeenCalledWith( + 'Ignoring ambiguous legacy terminal update', + expect.objectContaining({ blockId: 'function-1', candidateCount: 2 }) + ) + }) + + it('enriches an exactly matched legacy child workflow entry with its instance id', () => { + useTerminalConsoleStore.getState().addConsole({ + workflowId: 'wf-1', + blockId: 'workflow-1', + blockName: 'Workflow', + blockType: 'workflow', + executionId: 'exec-1', + executionOrder: 3, + iterationCurrent: 1, + iterationType: 'loop', + iterationContainerId: 'loop-1', + isRunning: true, + }) + + useTerminalConsoleStore.getState().updateConsole( + 'workflow-1', + { + childWorkflowInstanceId: 'child-inst-1', + executionOrder: 3, + iterationCurrent: 1, + iterationType: 'loop', + iterationContainerId: 'loop-1', + }, + 'exec-1' + ) + + expect(useTerminalConsoleStore.getState().getWorkflowEntries('wf-1')[0]).toMatchObject({ + childWorkflowInstanceId: 'child-inst-1', + isRunning: true, + }) + }) + }) + describe('cancelRunningEntries', () => { it('flips a plain running entry to canceled', () => { useTerminalConsoleStore.getState().addConsole({ diff --git a/apps/sim/stores/terminal/console/store.ts b/apps/sim/stores/terminal/console/store.ts index 460ba524e01..dceca95f156 100644 --- a/apps/sim/stores/terminal/console/store.ts +++ b/apps/sim/stores/terminal/console/store.ts @@ -136,6 +136,21 @@ const matchesEntryForUpdate = ( return false } + if (update.iterationTotal !== undefined && entry.iterationTotal !== update.iterationTotal) { + return false + } + + if (update.iterationType !== undefined && entry.iterationType !== update.iterationType) { + return false + } + + if ( + update.parentIterations !== undefined && + JSON.stringify(entry.parentIterations ?? []) !== JSON.stringify(update.parentIterations) + ) { + return false + } + if ( update.childWorkflowBlockId !== undefined && entry.childWorkflowBlockId !== update.childWorkflowBlockId @@ -143,6 +158,13 @@ const matchesEntryForUpdate = ( return false } + if ( + update.childWorkflowName !== undefined && + entry.childWorkflowName !== update.childWorkflowName + ) { + return false + } + if ( update.childWorkflowInstanceId !== undefined && entry.childWorkflowInstanceId !== undefined && @@ -154,6 +176,62 @@ const matchesEntryForUpdate = ( return true } +function resolveEntryIdForUpdate( + state: ConsoleStore, + blockId: string, + executionId: string | undefined, + update: string | ConsoleUpdate, + warnOnAmbiguity = true +): string | undefined { + const blockExecutionId = typeof update === 'object' ? update.blockExecutionId : undefined + const directEntryId = blockExecutionId + ? state.entryIdByBlockExecutionId[blockExecutionId] + : undefined + + if (directEntryId) { + const location = state.entryLocationById[directEntryId] + const entry = location && state.workflowEntries[location.workflowId]?.[location.index] + if ( + entry?.id === directEntryId && + entry.blockId === blockId && + entry.executionId === executionId + ) { + return directEntryId + } + logger.warn('Ignoring terminal update whose invocation identity conflicts with its block', { + blockExecutionId, + blockId, + executionId, + }) + return undefined + } + + const candidateIds = + state.entryIdsByBlockExecution[getBlockExecutionKey(blockId, executionId)] ?? [] + const matchingIds = candidateIds.filter((entryId) => { + const location = state.entryLocationById[entryId] + const entry = location && state.workflowEntries[location.workflowId]?.[location.index] + return ( + entry?.id === entryId && + (!blockExecutionId || + !entry.blockExecutionId || + entry.blockExecutionId === blockExecutionId) && + matchesEntryForUpdate(entry, blockId, executionId, update) + ) + }) + + if (matchingIds.length === 1) return matchingIds[0] + if (matchingIds.length > 1 && warnOnAmbiguity) { + logger.warn('Ignoring ambiguous legacy terminal update', { + blockId, + executionId, + blockExecutionId, + candidateCount: matchingIds.length, + }) + } + return undefined +} + function cloneWorkflowEntries( workflowEntries: Record ): Record { @@ -164,10 +242,14 @@ function removeWorkflowIndexes( workflowId: string, entries: ConsoleEntry[], entryIdsByBlockExecution: Record, + entryIdByBlockExecutionId: Record, entryLocationById: Record ): void { for (const entry of entries) { delete entryLocationById[entry.id] + if (entry.blockExecutionId && entryIdByBlockExecutionId[entry.blockExecutionId] === entry.id) { + delete entryIdByBlockExecutionId[entry.blockExecutionId] + } const blockExecutionKey = getBlockExecutionKey(entry.blockId, entry.executionId) const existingIds = entryIdsByBlockExecution[blockExecutionKey] if (!existingIds) { @@ -187,10 +269,14 @@ function indexWorkflowEntries( workflowId: string, entries: ConsoleEntry[], entryIdsByBlockExecution: Record, + entryIdByBlockExecutionId: Record, entryLocationById: Record ): void { entries.forEach((entry, index) => { entryLocationById[entry.id] = { workflowId, index } + if (entry.blockExecutionId) { + entryIdByBlockExecutionId[entry.blockExecutionId] = entry.id + } const blockExecutionKey = getBlockExecutionKey(entry.blockId, entry.executionId) const existingIds = entryIdsByBlockExecution[blockExecutionKey] if (existingIds) { @@ -203,35 +289,58 @@ function indexWorkflowEntries( function rebuildWorkflowStateMaps(workflowEntries: Record) { const entryIdsByBlockExecution: Record = {} + const entryIdByBlockExecutionId: Record = {} const entryLocationById: Record = {} Object.entries(workflowEntries).forEach(([workflowId, entries]) => { - indexWorkflowEntries(workflowId, entries, entryIdsByBlockExecution, entryLocationById) + indexWorkflowEntries( + workflowId, + entries, + entryIdsByBlockExecution, + entryIdByBlockExecutionId, + entryLocationById + ) }) - return { entryIdsByBlockExecution, entryLocationById } + return { entryIdsByBlockExecution, entryIdByBlockExecutionId, entryLocationById } } function replaceWorkflowEntries( state: ConsoleStore, workflowId: string, nextEntries: ConsoleEntry[] -): Pick { +): Pick< + ConsoleStore, + 'workflowEntries' | 'entryIdsByBlockExecution' | 'entryIdByBlockExecutionId' | 'entryLocationById' +> { const workflowEntries = cloneWorkflowEntries(state.workflowEntries) const entryIdsByBlockExecution = { ...state.entryIdsByBlockExecution } + const entryIdByBlockExecutionId = { ...state.entryIdByBlockExecutionId } const entryLocationById = { ...state.entryLocationById } const previousEntries = workflowEntries[workflowId] ?? EMPTY_CONSOLE_ENTRIES - removeWorkflowIndexes(workflowId, previousEntries, entryIdsByBlockExecution, entryLocationById) + removeWorkflowIndexes( + workflowId, + previousEntries, + entryIdsByBlockExecution, + entryIdByBlockExecutionId, + entryLocationById + ) if (nextEntries.length === 0) { delete workflowEntries[workflowId] } else { workflowEntries[workflowId] = nextEntries - indexWorkflowEntries(workflowId, nextEntries, entryIdsByBlockExecution, entryLocationById) + indexWorkflowEntries( + workflowId, + nextEntries, + entryIdsByBlockExecution, + entryIdByBlockExecutionId, + entryLocationById + ) } - return { workflowEntries, entryIdsByBlockExecution, entryLocationById } + return { workflowEntries, entryIdsByBlockExecution, entryIdByBlockExecutionId, entryLocationById } } function appendWorkflowEntry( @@ -239,24 +348,38 @@ function appendWorkflowEntry( workflowId: string, newEntry: ConsoleEntry, trimmedEntries: ConsoleEntry[] -): Pick { +): Pick< + ConsoleStore, + 'workflowEntries' | 'entryIdsByBlockExecution' | 'entryIdByBlockExecutionId' | 'entryLocationById' +> { const workflowEntries = cloneWorkflowEntries(state.workflowEntries) const previousEntries = workflowEntries[workflowId] ?? EMPTY_CONSOLE_ENTRIES workflowEntries[workflowId] = trimmedEntries const entryLocationById = { ...state.entryLocationById } const entryIdsByBlockExecution = { ...state.entryIdsByBlockExecution } + const entryIdByBlockExecutionId = { ...state.entryIdByBlockExecutionId } const survivingIds = new Set(trimmedEntries.map((e) => e.id)) const droppedEntries = previousEntries.filter((e) => !survivingIds.has(e.id)) if (droppedEntries.length > 0) { - removeWorkflowIndexes(workflowId, droppedEntries, entryIdsByBlockExecution, entryLocationById) + removeWorkflowIndexes( + workflowId, + droppedEntries, + entryIdsByBlockExecution, + entryIdByBlockExecutionId, + entryLocationById + ) } trimmedEntries.forEach((entry, index) => { entryLocationById[entry.id] = { workflowId, index } }) + if (newEntry.blockExecutionId) { + entryIdByBlockExecutionId[newEntry.blockExecutionId] = newEntry.id + } + const blockExecutionKey = getBlockExecutionKey(newEntry.blockId, newEntry.executionId) const existingIds = entryIdsByBlockExecution[blockExecutionKey] if (existingIds) { @@ -267,7 +390,7 @@ function appendWorkflowEntry( entryIdsByBlockExecution[blockExecutionKey] = [newEntry.id] } - return { workflowEntries, entryIdsByBlockExecution, entryLocationById } + return { workflowEntries, entryIdsByBlockExecution, entryIdByBlockExecutionId, entryLocationById } } interface NotifyBlockErrorParams { @@ -336,6 +459,7 @@ export const useTerminalConsoleStore = create()( devtools((set, get) => ({ workflowEntries: {}, entryIdsByBlockExecution: {}, + entryIdByBlockExecutionId: {}, entryLocationById: {}, isOpen: false, _hasHydrated: false, @@ -345,6 +469,15 @@ export const useTerminalConsoleStore = create()( return get().getWorkflowEntries(entry.workflowId)[0] as ConsoleEntry | undefined } + if (entry.blockExecutionId) { + const existingId = get().entryIdByBlockExecutionId[entry.blockExecutionId] + const location = existingId ? get().entryLocationById[existingId] : undefined + const existingEntry = location + ? get().workflowEntries[location.workflowId]?.[location.index] + : undefined + if (existingId && existingEntry?.id === existingId) return existingEntry + } + const redactedEntry = { ...entry } if ( !isStreamingOutput(entry.output) && @@ -484,16 +617,12 @@ export const useTerminalConsoleStore = create()( updateConsole: (blockId: string, update: string | ConsoleUpdate, executionId?: string) => { set((state) => { - const candidateIds = - state.entryIdsByBlockExecution[getBlockExecutionKey(blockId, executionId)] ?? [] - if (candidateIds.length === 0) { - return state - } + const entryId = resolveEntryIdForUpdate(state, blockId, executionId, update) + if (!entryId) return state - const workflowId = state.entryLocationById[candidateIds[0]]?.workflowId - if (!workflowId) { - return state - } + const candidateIds = [entryId] + const workflowId = state.entryLocationById[entryId]?.workflowId + if (!workflowId) return state const currentEntries = state.workflowEntries[workflowId] ?? EMPTY_CONSOLE_ENTRIES let nextEntries: ConsoleEntry[] | null = null @@ -505,11 +634,8 @@ export const useTerminalConsoleStore = create()( const source = nextEntries ?? currentEntries const entry = source[location.index] if (!entry || entry.id !== candidateId) continue - if (!matchesEntryForUpdate(entry, blockId, executionId, update)) continue - if (!nextEntries) { - nextEntries = [...currentEntries] - } + if (!nextEntries) nextEntries = [...currentEntries] if (typeof update === 'string') { const newOutput = normalizeConsoleOutput(updateBlockOutput(entry.output, update)) @@ -542,103 +668,60 @@ export const useTerminalConsoleStore = create()( : normalizeConsoleOutput(mergedOutput) } - if (update.blockName !== undefined) { - updatedEntry.blockName = update.blockName - } - - if (update.blockType !== undefined) { - updatedEntry.blockType = update.blockType - } - - if (update.error !== undefined) { - updatedEntry.error = normalizeConsoleError(update.error) - } - + if (update.blockName !== undefined) updatedEntry.blockName = update.blockName + if (update.blockType !== undefined) updatedEntry.blockType = update.blockType + if (update.error !== undefined) updatedEntry.error = normalizeConsoleError(update.error) if (update.warning !== undefined) { updatedEntry.warning = normalizeConsoleError(update.warning) ?? undefined } - - if (update.success !== undefined) { - updatedEntry.success = update.success - } - - if (update.startedAt !== undefined) { - updatedEntry.startedAt = update.startedAt - } - - if (update.endedAt !== undefined) { - updatedEntry.endedAt = update.endedAt - } - - if (update.durationMs !== undefined) { - updatedEntry.durationMs = update.durationMs - } - + if (update.success !== undefined) updatedEntry.success = update.success + if (update.startedAt !== undefined) updatedEntry.startedAt = update.startedAt + if (update.endedAt !== undefined) updatedEntry.endedAt = update.endedAt + if (update.durationMs !== undefined) updatedEntry.durationMs = update.durationMs if (update.input !== undefined) { updatedEntry.input = typeof update.input === 'object' && update.input !== null ? normalizeConsoleInput(redactApiKeys(update.input)) : normalizeConsoleInput(update.input) } - - if (update.isRunning !== undefined) { - updatedEntry.isRunning = update.isRunning - } - - if (update.isCanceled !== undefined) { - updatedEntry.isCanceled = update.isCanceled - } - + if (update.isRunning !== undefined) updatedEntry.isRunning = update.isRunning + if (update.isCanceled !== undefined) updatedEntry.isCanceled = update.isCanceled if (update.iterationCurrent !== undefined) { updatedEntry.iterationCurrent = update.iterationCurrent } - if (update.iterationTotal !== undefined) { updatedEntry.iterationTotal = update.iterationTotal } - - if (update.iterationType !== undefined) { - updatedEntry.iterationType = update.iterationType - } - + if (update.iterationType !== undefined) updatedEntry.iterationType = update.iterationType if (update.iterationContainerId !== undefined) { updatedEntry.iterationContainerId = update.iterationContainerId } - if (update.parentIterations !== undefined) { updatedEntry.parentIterations = update.parentIterations } - if (update.childWorkflowBlockId !== undefined) { updatedEntry.childWorkflowBlockId = update.childWorkflowBlockId } - if (update.childWorkflowName !== undefined) { updatedEntry.childWorkflowName = update.childWorkflowName } - if (update.childWorkflowInstanceId !== undefined) { updatedEntry.childWorkflowInstanceId = update.childWorkflowInstanceId } - + if (update.blockExecutionId !== undefined) { + updatedEntry.blockExecutionId = update.blockExecutionId + } if (update.agentStreamThinking !== undefined) { updatedEntry.agentStreamThinking = update.agentStreamThinking } - - if (update.clearAgentStreamThinking) { - updatedEntry.agentStreamThinking = undefined - } - + if (update.clearAgentStreamThinking) updatedEntry.agentStreamThinking = undefined if (update.agentStreamToolCalls !== undefined) { updatedEntry.agentStreamToolCalls = update.agentStreamToolCalls } - if (update.agentStreamActive !== undefined) { updatedEntry.agentStreamActive = update.agentStreamActive } - // Settle live chrome whenever an entry stops running or stream activity ends. - // block:error / timeouts often skip stream:done and only flip isRunning. const shouldSettleAgentStream = update.isRunning === false || update.agentStreamActive === false || @@ -657,28 +740,30 @@ export const useTerminalConsoleStore = create()( nextEntries[location.index] = updatedEntry } - if (!nextEntries) { - return state - } + if (!nextEntries) return state - const workflowEntriesClone = cloneWorkflowEntries(state.workflowEntries) - workflowEntriesClone[workflowId] = nextEntries + const workflowEntries = cloneWorkflowEntries(state.workflowEntries) + workflowEntries[workflowId] = nextEntries + const entryIdByBlockExecutionId = + typeof update === 'object' && update.blockExecutionId + ? { ...state.entryIdByBlockExecutionId, [update.blockExecutionId]: entryId } + : state.entryIdByBlockExecutionId return { - workflowEntries: workflowEntriesClone, + workflowEntries, entryIdsByBlockExecution: state.entryIdsByBlockExecution, + entryIdByBlockExecutionId, entryLocationById: state.entryLocationById, } }) if (typeof update === 'object' && update.error) { - const matchingEntry = get() - .getWorkflowEntries( - get().entryLocationById[ - (get().entryIdsByBlockExecution[getBlockExecutionKey(blockId, executionId)] ?? - [])[0] ?? '' - ]?.workflowId ?? '' - ) - .find((entry) => matchesEntryForUpdate(entry, blockId, executionId, update)) + const state = get() + const matchingId = resolveEntryIdForUpdate(state, blockId, executionId, update, false) + const location = matchingId ? state.entryLocationById[matchingId] : undefined + const matchingEntry = location + ? state.workflowEntries[location.workflowId]?.[location.index] + : undefined + if (!matchingEntry) return notifyBlockError({ error: update.error, blockName: update.blockName || matchingEntry?.blockName || 'Unknown Block', diff --git a/apps/sim/stores/terminal/console/types.ts b/apps/sim/stores/terminal/console/types.ts index e3daa33f0e6..8070c75486d 100644 --- a/apps/sim/stores/terminal/console/types.ts +++ b/apps/sim/stores/terminal/console/types.ts @@ -11,6 +11,8 @@ export interface ConsoleEntry { blockName: string blockType: string executionId?: string + /** Unique identity for this individual block invocation. */ + blockExecutionId?: string startedAt?: string executionOrder: number endedAt?: string @@ -42,6 +44,7 @@ export interface ConsoleEntry { } export interface ConsoleUpdate { + blockExecutionId?: string content?: string output?: Partial replaceOutput?: NormalizedBlockOutput @@ -79,6 +82,7 @@ export interface ConsoleEntryLocation { export interface ConsoleStore { workflowEntries: Record entryIdsByBlockExecution: Record + entryIdByBlockExecutionId: Record entryLocationById: Record isOpen: boolean addConsole: (entry: Omit) => ConsoleEntry | undefined