fix(deployed-chat): allow non-streaming responses in deployed chat, allow partial failure responses in deployed chat (#833)

* fix(deployed-chat): allow non-streaming responses in deployed chat, allow partial failure responses in deployed chat

* fix(csp): runtime variable resolution for CSP

* cleanup

---------

Co-authored-by: waleedlatif <waleedlatif@waleedlatifs-MacBook-Pro.local>

Author: waleedlatif1

apps/sim/app/api/chat/utils.test.ts

@@ -350,4 +350,77 @@ describe('Chat API Utils', () => {
       expect(result3.error).toBe('Email not authorized')
     })
   })
+
+  describe('Execution Result Processing', () => {
+    it('should process logs regardless of overall success status', () => {
+      // Test that logs are processed even when overall execution fails
+      // This is key for partial success scenarios
+      const executionResult = {
+        success: false, // Overall execution failed
+        output: {},
+        logs: [
+          {
+            blockId: 'agent1',
+            startedAt: '2023-01-01T00:00:00Z',
+            endedAt: '2023-01-01T00:00:01Z',
+            durationMs: 1000,
+            success: true,
+            output: { content: 'Agent 1 succeeded' },
+            error: undefined,
+          },
+          {
+            blockId: 'agent2',
+            startedAt: '2023-01-01T00:00:00Z',
+            endedAt: '2023-01-01T00:00:01Z',
+            durationMs: 500,
+            success: false,
+            output: null,
+            error: 'Agent 2 failed',
+          },
+        ],
+        metadata: { duration: 1000 },
+      }
+
+      // Test the key logic: logs should be processed regardless of overall success
+      expect(executionResult.success).toBe(false)
+      expect(executionResult.logs).toBeDefined()
+      expect(executionResult.logs).toHaveLength(2)
+
+      // First log should be successful
+      expect(executionResult.logs[0].success).toBe(true)
+      expect(executionResult.logs[0].output?.content).toBe('Agent 1 succeeded')
+
+      // Second log should be failed
+      expect(executionResult.logs[1].success).toBe(false)
+      expect(executionResult.logs[1].error).toBe('Agent 2 failed')
+    })
+
+    it('should handle ExecutionResult vs StreamingExecution types correctly', () => {
+      const executionResult = {
+        success: true,
+        output: { content: 'test' },
+        logs: [],
+        metadata: { duration: 100 },
+      }
+
+      // Test direct ExecutionResult
+      const directResult = executionResult
+      const extractedDirect = directResult
+      expect(extractedDirect).toBe(executionResult)
+
+      // Test StreamingExecution with embedded ExecutionResult
+      const streamingResult = {
+        stream: new ReadableStream(),
+        execution: executionResult,
+      }
+
+      // Simulate the type extraction logic from executeWorkflowForChat
+      const extractedFromStreaming =
+        streamingResult && typeof streamingResult === 'object' && 'execution' in streamingResult
+          ? streamingResult.execution
+          : streamingResult
+
+      expect(extractedFromStreaming).toBe(executionResult)
+    })
+  })
 })

apps/sim/app/api/chat/utils.ts

@@ -14,7 +14,7 @@ import { getBlock } from '@/blocks'
 import { db } from '@/db'
 import { chat, environment as envTable, userStats, workflow } from '@/db/schema'
 import { Executor } from '@/executor'
-import type { BlockLog } from '@/executor/types'
+import type { BlockLog, ExecutionResult } from '@/executor/types'
 import { Serializer } from '@/serializer'
 import { mergeSubblockState } from '@/stores/workflows/server-utils'
 import type { WorkflowState } from '@/stores/workflows/workflow/types'
@@ -549,6 +549,7 @@ export async function executeWorkflowForChat(
     async start(controller) {
       const encoder = new TextEncoder()
       const streamedContent = new Map<string, string>()
+      const streamedBlocks = new Set<string>() // Track which blocks have started streaming
 
       const onStream = async (streamingExecution: any): Promise<void> => {
         if (!streamingExecution.stream) return
@@ -557,6 +558,15 @@ export async function executeWorkflowForChat(
         const reader = streamingExecution.stream.getReader()
         if (blockId) {
           streamedContent.set(blockId, '')
+
+          // Add separator if this is not the first block to stream
+          if (streamedBlocks.size > 0) {
+            // Send separator before the new block starts
+            controller.enqueue(
+              encoder.encode(`data: ${JSON.stringify({ blockId, chunk: '\n\n' })}\n\n`)
+            )
+          }
+          streamedBlocks.add(blockId)
         }
         try {
           while (true) {
@@ -615,25 +625,117 @@ export async function executeWorkflowForChat(
         throw error
       }
 
-      if (result && 'success' in result) {
-        // Update streamed content and apply tokenization
-        if (result.logs) {
-          result.logs.forEach((log: BlockLog) => {
-            if (streamedContent.has(log.blockId)) {
-              const content = streamedContent.get(log.blockId)
-              if (log.output) {
-                log.output.content = content
+      // Handle both ExecutionResult and StreamingExecution types
+      const executionResult =
+        result && typeof result === 'object' && 'execution' in result
+          ? (result.execution as ExecutionResult)
+          : (result as ExecutionResult)
+
+      if (executionResult?.logs) {
+        // Update streamed content and apply tokenization - process regardless of overall success
+        // This ensures partial successes (some agents succeed, some fail) still return results
+
+        // Add newlines between different agent outputs for better readability
+        const processedOutputs = new Set<string>()
+        executionResult.logs.forEach((log: BlockLog) => {
+          if (streamedContent.has(log.blockId)) {
+            const content = streamedContent.get(log.blockId)
+            if (log.output && content) {
+              // Add newline separation between different outputs (but not before the first one)
+              const separator = processedOutputs.size > 0 ? '\n\n' : ''
+              log.output.content = separator + content
+              processedOutputs.add(log.blockId)
+            }
+          }
+        })
+
+        // Also process non-streamed outputs from selected blocks (like function blocks)
+        // This uses the same logic as the chat panel to ensure identical behavior
+        const nonStreamingLogs = executionResult.logs.filter(
+          (log: BlockLog) => !streamedContent.has(log.blockId)
+        )
+
+        // Extract the exact same functions used by the chat panel
+        const extractBlockIdFromOutputId = (outputId: string): string => {
+          return outputId.includes('_') ? outputId.split('_')[0] : outputId.split('.')[0]
+        }
+
+        const extractPathFromOutputId = (outputId: string, blockId: string): string => {
+          return outputId.substring(blockId.length + 1)
+        }
+
+        const parseOutputContentSafely = (output: any): any => {
+          if (!output?.content) {
+            return output
+          }
+
+          if (typeof output.content === 'string') {
+            try {
+              return JSON.parse(output.content)
+            } catch (e) {
+              // Fallback to original structure if parsing fails
+              return output
+            }
+          }
+
+          return output
+        }
+
+        // Filter outputs that have matching logs (exactly like chat panel)
+        const outputsToRender = selectedOutputIds.filter((outputId) => {
+          const blockIdForOutput = extractBlockIdFromOutputId(outputId)
+          return nonStreamingLogs.some((log) => log.blockId === blockIdForOutput)
+        })
+
+        // Process each selected output (exactly like chat panel)
+        for (const outputId of outputsToRender) {
+          const blockIdForOutput = extractBlockIdFromOutputId(outputId)
+          const path = extractPathFromOutputId(outputId, blockIdForOutput)
+          const log = nonStreamingLogs.find((l) => l.blockId === blockIdForOutput)
+
+          if (log) {
+            let outputValue: any = log.output
+
+            if (path) {
+              // Parse JSON content safely (exactly like chat panel)
+              outputValue = parseOutputContentSafely(outputValue)
+
+              const pathParts = path.split('.')
+              for (const part of pathParts) {
+                if (outputValue && typeof outputValue === 'object' && part in outputValue) {
+                  outputValue = outputValue[part]
+                } else {
+                  outputValue = undefined
+                  break
+                }
               }
             }
-          })
 
-          // Process all logs for streaming tokenization
-          const processedCount = processStreamingBlockLogs(result.logs, streamedContent)
-          logger.info(`[CHAT-API] Processed ${processedCount} blocks for streaming tokenization`)
+            if (outputValue !== undefined) {
+              // Add newline separation between different outputs
+              const separator = processedOutputs.size > 0 ? '\n\n' : ''
+
+              // Format the output exactly like the chat panel
+              const formattedOutput =
+                typeof outputValue === 'string' ? outputValue : JSON.stringify(outputValue, null, 2)
+
+              // Update the log content
+              if (!log.output.content) {
+                log.output.content = separator + formattedOutput
+              } else {
+                log.output.content = separator + formattedOutput
+              }
+              processedOutputs.add(log.blockId)
+            }
+          }
         }
 
-        const { traceSpans, totalDuration } = buildTraceSpans(result)
-        const enrichedResult = { ...result, traceSpans, totalDuration }
+        // Process all logs for streaming tokenization
+        const processedCount = processStreamingBlockLogs(executionResult.logs, streamedContent)
+        logger.info(`Processed ${processedCount} blocks for streaming tokenization`)
+
+        const { traceSpans, totalDuration } = buildTraceSpans(executionResult)
+        const enrichedResult = { ...executionResult, traceSpans, totalDuration }
         if (conversationId) {
           if (!enrichedResult.metadata) {
             enrichedResult.metadata = {
@@ -646,7 +748,7 @@ export async function executeWorkflowForChat(
         const executionId = uuidv4()
         logger.debug(`Generated execution ID for deployed chat: ${executionId}`)
 
-        if (result.success) {
+        if (executionResult.success) {
           try {
             await db
               .update(userStats)
@@ -669,12 +771,12 @@ export async function executeWorkflowForChat(
       }
 
       // Complete logging session (for both success and failure)
-      if (result && 'success' in result) {
-        const { traceSpans } = buildTraceSpans(result)
+      if (executionResult?.logs) {
+        const { traceSpans } = buildTraceSpans(executionResult)
         await loggingSession.safeComplete({
           endedAt: new Date().toISOString(),
-          totalDurationMs: result.metadata?.duration || 0,
-          finalOutput: result.output,
+          totalDurationMs: executionResult.metadata?.duration || 0,
+          finalOutput: executionResult.output,
           traceSpans,
         })
       }

apps/sim/app/chat/[subdomain]/hooks/use-chat-streaming.ts

@@ -3,6 +3,8 @@
 import { useRef, useState } from 'react'
 import { createLogger } from '@/lib/logs/console/logger'
 import type { ChatMessage } from '@/app/chat/[subdomain]/components/message/message'
+// No longer need complex output extraction - backend handles this
+import type { ExecutionResult } from '@/executor/types'
 
 const logger = createLogger('UseChatStreaming')
 
@@ -96,6 +98,8 @@ export function useChatStreaming() {
     let accumulatedText = ''
     let lastAudioPosition = 0
 
+    // Track which blocks have streamed content (like chat panel)
+    const messageIdMap = new Map<string, string>()
     const messageId = crypto.randomUUID()
     setMessages((prev) => [
       ...prev,
@@ -148,13 +152,49 @@ export function useChatStreaming() {
               const { blockId, chunk: contentChunk, event: eventType } = json
 
               if (eventType === 'final' && json.data) {
+                // The backend has already processed and combined all outputs
+                // We just need to extract the combined content and use it
+                const result = json.data as ExecutionResult
+
+                // Collect all content from logs that have output.content (backend processed)
+                let combinedContent = ''
+                if (result.logs) {
+                  const contentParts: string[] = []
+
+                  // Get content from all logs that have processed content
+                  result.logs.forEach((log) => {
+                    if (log.output?.content && typeof log.output.content === 'string') {
+                      // The backend already includes proper separators, so just collect the content
+                      contentParts.push(log.output.content)
+                    }
+                  })
+
+                  // Join without additional separators since backend already handles this
+                  combinedContent = contentParts.join('')
+                }
+
+                // Update the existing streaming message with the final combined content
                 setMessages((prev) =>
-                  prev.map((msg) => (msg.id === messageId ? { ...msg, isStreaming: false } : msg))
+                  prev.map((msg) =>
+                    msg.id === messageId
+                      ? {
+                          ...msg,
+                          content: combinedContent || accumulatedText, // Use combined content or fallback to streamed
+                          isStreaming: false,
+                        }
+                      : msg
+                  )
                 )
+
                 return
               }
 
               if (blockId && contentChunk) {
+                // Track that this block has streamed content (like chat panel)
+                if (!messageIdMap.has(blockId)) {
+                  messageIdMap.set(blockId, messageId)
+                }
+
                 accumulatedText += contentChunk
                 setMessages((prev) =>
                   prev.map((msg) =>