address comments

Author: icecrasher321

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

@@ -27,7 +27,7 @@ import {
 } from '@/lib/execution/call-chain'
 import {
   createExecutionEventWriter,
-  finalizeExecutionStream,
+  flushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta,
   type TerminalExecutionStreamStatus,
 } from '@/lib/execution/event-buffer'
@@ -1296,15 +1296,14 @@ async function handleExecutePost(
             isManualAbortRegistered = false
           }
           if (finalMetaStatus && !terminalEventPublished) {
-            const terminalPublished = await finalizeExecutionStream(
+            const replayBufferFlushed = await flushExecutionStreamReplayBuffer(
               executionId,
-              eventWriter,
-              finalMetaStatus
+              eventWriter
             )
             reqLogger.error('Failed to publish terminal execution event durably', {
               executionId,
               status: finalMetaStatus,
-              replayBufferFlushed: terminalPublished,
+              replayBufferFlushed,
             })
             if (!isStreamClosed) {
               controller.error(new Error('Run buffer terminal event publish failed'))

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts

@@ -427,7 +427,7 @@ describe('workflow-execution-utils', () => {
         executionId: 'exec-1',
         executionOrder: 3,
         isRunning: true,
-        childWorkflowBlockId: 'child-inst-1',
+        childWorkflowBlockId: 'workflow-1',
         childWorkflowName: 'Workflow 1',
       })
       terminalConsoleMockFns.mockAddConsole({
@@ -489,7 +489,7 @@ describe('workflow-execution-utils', () => {
           success: true,
           isRunning: false,
           isCanceled: false,
-          childWorkflowBlockId: 'child-inst-1',
+          childWorkflowBlockId: 'workflow-1',
         }),
         'exec-1',
       ])
@@ -501,7 +501,7 @@ describe('workflow-execution-utils', () => {
           error: 'Request failed',
           isRunning: false,
           isCanceled: false,
-          childWorkflowBlockId: 'child-inst-1',
+          childWorkflowBlockId: 'workflow-1',
         }),
         'exec-1',
       ])
@@ -529,7 +529,7 @@ describe('workflow-execution-utils', () => {
         iterationCurrent: 0,
         iterationType: 'loop',
         iterationContainerId: 'loop-1',
-        childWorkflowBlockId: 'child-inst-1',
+        childWorkflowBlockId: 'workflow-1',
       })
       terminalConsoleMockFns.mockAddConsole({
         workflowId: 'wf-1',
@@ -542,7 +542,7 @@ describe('workflow-execution-utils', () => {
         iterationCurrent: 1,
         iterationType: 'loop',
         iterationContainerId: 'loop-1',
-        childWorkflowBlockId: 'child-inst-1',
+        childWorkflowBlockId: 'workflow-1',
       })
 
       const startedAt = new Date().toISOString()
@@ -632,7 +632,7 @@ describe('workflow-execution-utils', () => {
         executionId: 'exec-1',
         executionOrder: 3,
         isRunning: false,
-        childWorkflowBlockId: 'child-inst-1',
+        childWorkflowBlockId: 'workflow-1',
         childWorkflowInstanceId: 'nested-inst-1',
       })
       terminalConsoleMockFns.mockAddConsole({
@@ -643,7 +643,7 @@ describe('workflow-execution-utils', () => {
         executionId: 'exec-1',
         executionOrder: 1,
         isRunning: true,
-        childWorkflowBlockId: 'nested-inst-1',
+        childWorkflowBlockId: 'nested-workflow',
       })
 
       const startedAt = new Date().toISOString()
@@ -688,7 +688,7 @@ describe('workflow-execution-utils', () => {
       expect(updateConsole.mock.calls[1]).toEqual([
         'nested-api',
         expect.objectContaining({
-          childWorkflowBlockId: 'nested-inst-1',
+          childWorkflowBlockId: 'nested-workflow',
           success: true,
           isRunning: false,
           isCanceled: false,

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

@@ -512,6 +512,7 @@ export function reconcileFinalBlockLogs(
       reconcileChildTraceSpans(
         updateConsole,
         workflowId,
+        log.blockId,
         childWorkflowInstanceId,
         executionId,
         log.childTraceSpans
@@ -523,20 +524,21 @@ export function reconcileFinalBlockLogs(
 function reconcileChildTraceSpans(
   updateConsole: UpdateConsoleFn,
   workflowId: string,
+  childWorkflowBlockId: string,
   childWorkflowInstanceId: string,
   executionId: string,
   spans: TraceSpan[]
 ): void {
   for (const span of spans) {
     const matchingEntry = span.blockId
-      ? findConsoleEntryForSpan(workflowId, executionId, childWorkflowInstanceId, span)
+      ? findConsoleEntryForSpan(workflowId, executionId, childWorkflowBlockId, span)
       : undefined
     if (span.blockId) {
       const errorMessage = normalizeSpanError(span.output?.error)
       updateConsole(
         span.blockId,
         {
-          ...spanConsoleIdentity(span, childWorkflowInstanceId),
+          ...spanConsoleIdentity(span, childWorkflowBlockId),
           replaceOutput: (span.output ?? {}) as Record<string, unknown>,
           success: span.status !== 'error',
           ...(errorMessage !== undefined ? { error: errorMessage } : {}),
@@ -553,6 +555,7 @@ function reconcileChildTraceSpans(
       reconcileChildTraceSpans(
         updateConsole,
         workflowId,
+        matchingEntry?.blockId ?? childWorkflowBlockId,
         matchingEntry?.childWorkflowInstanceId ?? childWorkflowInstanceId,
         executionId,
         span.children
@@ -561,7 +564,7 @@ function reconcileChildTraceSpans(
   }
 }
 
-function spanConsoleIdentity(span: TraceSpan, childWorkflowInstanceId: string): ConsoleUpdate {
+function spanConsoleIdentity(span: TraceSpan, childWorkflowBlockId: string): ConsoleUpdate {
   const iterationContainerId = span.loopId ?? span.parallelId
   const iterationType = span.loopId ? 'loop' : span.parallelId ? 'parallel' : undefined
   return {
@@ -570,18 +573,18 @@ function spanConsoleIdentity(span: TraceSpan, childWorkflowInstanceId: string):
     ...(iterationType !== undefined && { iterationType }),
     ...(iterationContainerId !== undefined && { iterationContainerId }),
     ...(span.parentIterations !== undefined && { parentIterations: span.parentIterations }),
-    childWorkflowBlockId: childWorkflowInstanceId,
+    childWorkflowBlockId,
   }
 }
 
 function findConsoleEntryForSpan(
   workflowId: string,
   executionId: string,
-  childWorkflowInstanceId: string,
+  childWorkflowBlockId: string,
   span: TraceSpan
 ): ConsoleEntry | undefined {
   if (!span.blockId) return undefined
-  const identity = spanConsoleIdentity(span, childWorkflowInstanceId)
+  const identity = spanConsoleIdentity(span, childWorkflowBlockId)
   return useTerminalConsoleStore
     .getState()
     .getWorkflowEntries(workflowId)
@@ -615,6 +618,13 @@ function matchesConsoleIdentity(entry: ConsoleEntry, identity: ConsoleUpdate): b
   ) {
     return false
   }
+  if (
+    identity.childWorkflowInstanceId !== undefined &&
+    entry.childWorkflowInstanceId !== undefined &&
+    entry.childWorkflowInstanceId !== identity.childWorkflowInstanceId
+  ) {
+    return false
+  }
   return true
 }
 

apps/sim/hooks/use-execution-stream.ts

@@ -64,12 +64,8 @@ function isClientDisconnectError(error: any): boolean {
   return error.name === 'AbortError'
 }
 
-function isAbortDisconnectError(error: any): boolean {
-  return error.name === 'AbortError'
-}
-
 function isRecoverableStreamError(error: any): boolean {
-  if (isAbortDisconnectError(error)) return false
+  if (isClientDisconnectError(error)) return false
   const msg = (error.message ?? '').toLowerCase()
   return (
     msg.includes('network error') || msg.includes('failed to fetch') || msg.includes('load failed')
@@ -466,7 +462,7 @@ export function useExecutionStream() {
 
       await processSSEStream(response.body.getReader(), callbacks, 'Reconnect')
     } catch (error: any) {
-      if (isAbortDisconnectError(error)) return
+      if (isClientDisconnectError(error)) return
       logger.error('Reconnection stream error:', error)
       throw error
     } finally {

apps/sim/lib/execution/event-buffer.test.ts

@@ -27,7 +27,7 @@ vi.mock('@/lib/core/config/redis', () => ({
 
 import {
   createExecutionEventWriter,
-  finalizeExecutionStream,
+  flushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta,
   readExecutionEventsState,
 } from '@/lib/execution/event-buffer'
@@ -215,7 +215,7 @@ describe('execution event buffer', () => {
     await expect(writer.write(makeEvent('lost'))).rejects.toThrow('redis reservation failed')
     await writer.write(makeEvent('terminal'))
 
-    await expect(finalizeExecutionStream('exec-1', writer, 'complete')).resolves.toBe(true)
+    await expect(flushExecutionStreamReplayBuffer('exec-1', writer)).resolves.toBe(true)
     expect(persistedEntries.map((entry) => entry.eventId)).toEqual([101])
     expect(mockRedis.hset).not.toHaveBeenCalledWith(
       expect.any(String),
@@ -230,7 +230,7 @@ describe('execution event buffer', () => {
     const writer = createExecutionEventWriter('exec-1')
     await writer.write(makeEvent('terminal'))
 
-    await expect(finalizeExecutionStream('exec-1', writer, 'complete')).resolves.toBe(false)
+    await expect(flushExecutionStreamReplayBuffer('exec-1', writer)).resolves.toBe(false)
     expect(mockRedis.hset).not.toHaveBeenCalled()
   })
 
@@ -269,7 +269,7 @@ describe('execution event buffer', () => {
     const writer = createExecutionEventWriter('exec-1')
     await writer.write(makeEvent('terminal'))
 
-    await expect(finalizeExecutionStream('exec-1', writer, 'complete')).resolves.toBe(true)
+    await expect(flushExecutionStreamReplayBuffer('exec-1', writer)).resolves.toBe(true)
     expect(persistedEntries.map((entry) => entry.eventId)).toEqual([1])
     expect(mockRedis.hset).not.toHaveBeenCalledWith(
       expect.any(String),

apps/sim/lib/execution/event-buffer.ts

@@ -211,10 +211,9 @@ function createMemoryExecutionEventWriter(executionId: string): ExecutionEventWr
   }
 }
 
-export async function finalizeExecutionStream(
+export async function flushExecutionStreamReplayBuffer(
   executionId: string,
-  writer: ExecutionEventWriter,
-  status: TerminalExecutionStreamStatus
+  writer: ExecutionEventWriter
 ): Promise<boolean> {
   let writerClosed = false
   for (let attempt = 1; attempt <= FINALIZE_FLUSH_ATTEMPTS; attempt++) {
@@ -227,7 +226,6 @@ export async function finalizeExecutionStream(
     } catch (error) {
       logger.warn('Failed to flush execution stream replay buffer during finalization', {
         executionId,
-        status,
         attempt,
         error: toError(error).message,
       })

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -8,7 +8,7 @@ import type { Edge } from 'reactflow'
 import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core/execution-limits'
 import {
   createExecutionEventWriter,
-  finalizeExecutionStream,
+  flushExecutionStreamReplayBuffer,
   initializeExecutionStreamMeta,
   resetExecutionStreamBuffer,
   type TerminalExecutionStreamStatus,
@@ -1282,15 +1282,14 @@ export class PauseResumeManager {
     } finally {
       timeoutController?.cleanup()
       if (!terminalEventPublished) {
-        const terminalPublished = await finalizeExecutionStream(
+        const replayBufferFlushed = await flushExecutionStreamReplayBuffer(
           resumeExecutionId,
-          eventWriter,
-          finalMetaStatus
+          eventWriter
         )
         logger.warn('Failed to publish resume terminal event durably', {
           resumeExecutionId,
           status: finalMetaStatus,
-          replayBufferFlushed: terminalPublished,
+          replayBufferFlushed,
         })
         if (!executionError) {
           executionError = new Error(TERMINAL_PUBLISH_ERROR)

apps/sim/stores/terminal/console/store.ts

@@ -110,6 +110,14 @@ const matchesEntryForUpdate = (
     return false
   }
 
+  if (
+    update.childWorkflowInstanceId !== undefined &&
+    entry.childWorkflowInstanceId !== undefined &&
+    entry.childWorkflowInstanceId !== update.childWorkflowInstanceId
+  ) {
+    return false
+  }
+
   return true
 }