fix(chat): suppress mixed invocation outputs

Author: Bill Leoutsakos

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

@@ -7,6 +7,7 @@ import { createRoot, type Root } from 'react-dom/client'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
+  chatStoreState,
   executionStoreState,
   mockCancel,
   mockExecute,
@@ -22,6 +23,9 @@ const {
   workflowBlocks,
   workflowStoreState,
 } = vi.hoisted(() => {
+  const chatStoreState = {
+    selectedWorkflowOutputs: [] as string[],
+  }
   const workflowBlocks = {
     start: {
       id: 'start',
@@ -82,6 +86,7 @@ const {
   }
 
   return {
+    chatStoreState,
     executionStoreState,
     mockCancel: vi.fn(),
     mockExecute: vi.fn(),
@@ -167,9 +172,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<void>
+  }) => ({
     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 +235,7 @@ vi.mock('@/serializer', () => ({
 vi.mock('@/stores/chat/store', () => ({
   useChatStore: {
     getState: () => ({
-      getSelectedWorkflowOutput: () => [],
+      getSelectedWorkflowOutput: () => chatStoreState.selectedWorkflowOutputs,
     }),
   },
 }))
@@ -343,6 +357,7 @@ async function drainStream(value: unknown): Promise<void> {
 describe('useWorkflowExecution cancellation', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    chatStoreState.selectedWorkflowOutputs = []
     executionStoreState.getCurrentExecutionId.mockReturnValue('execution-1')
     mockRequestJson.mockResolvedValue({ success: true })
   })
@@ -402,6 +417,7 @@ describe('useWorkflowExecution cancellation', () => {
 describe('useWorkflowExecution attachment uploads', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    chatStoreState.selectedWorkflowOutputs = []
     executionStoreState.getCurrentExecutionId.mockReturnValue(null)
     mockResolveStartCandidates.mockReturnValue([])
     mockSelectBestTrigger.mockReturnValue([])
@@ -584,6 +600,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 new Promise((resolve) => setTimeout(resolve, 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')

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

@@ -784,6 +784,7 @@ export function useWorkflowExecution() {
           async start(controller) {
             const { encodeSSE } = await import('@/lib/core/utils/sse')
             const streamedChunks = new Map<string, string[]>()
+            const streamedBlockIds = new Set<string>()
             const streamReadingPromises: Promise<void>[] = []
 
             const safeEnqueue = (data: Uint8Array) => {
@@ -821,6 +822,7 @@ export function useWorkflowExecution() {
                     }
                     const chunk = new TextDecoder().decode(value)
                     if (streamKey) {
+                      if (blockId) streamedBlockIds.add(blockId)
                       streamedChunks.get(streamKey)!.push(chunk)
                     }
 
@@ -844,9 +846,9 @@ 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, blockExecutionId?: string) => {
               const streamKey = blockExecutionId ?? blockId
@@ -864,7 +866,7 @@ export function useWorkflowExecution() {
             ) => {
               const streamKey = blockExecutionId ?? blockId
               // Skip if this block already had streaming content (avoid duplicates)
-              if (streamedChunks.has(streamKey)) {
+              if (streamedChunks.has(streamKey) || streamedBlockIds.has(blockId)) {
                 logger.debug('[handleRunWorkflow] Skipping onBlockComplete for streaming block', {
                   blockId,
                 })