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

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) =>