fix(stream) handle task switching (#3609)

* Fix task switching causing stream to abort

* Process all task streams all the time

* Process task streams that are in the background

---------

Co-authored-by: Theodore Li <theo@sim.ai>

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -12,6 +12,7 @@ import { VFS_DIR_TO_RESOURCE } from '@/lib/copilot/resource-types'
 import { isWorkflowToolName } from '@/lib/copilot/workflow-tools'
 import { getNextWorkflowColor } from '@/lib/workflows/colors'
 import { invalidateResourceQueries } from '@/app/workspace/[workspaceId]/home/components/mothership-view/components/resource-registry'
+import { deploymentKeys } from '@/hooks/queries/deployments'
 import {
   type TaskChatHistory,
   type TaskStoredContentBlock,
@@ -21,6 +22,7 @@ import {
   taskKeys,
   useChatHistory,
 } from '@/hooks/queries/tasks'
+import { workflowKeys } from '@/hooks/queries/workflows'
 import { getTopInsertionSortOrder } from '@/hooks/queries/utils/top-insertion-sort-order'
 import { useExecutionStream } from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution/store'
@@ -74,6 +76,8 @@ const STATE_TO_STATUS: Record<string, ToolCallStatus> = {
   skipped: 'success',
 } as const
 
+const DEPLOY_TOOL_NAMES = new Set(['deploy_api', 'deploy_chat', 'deploy_mcp', 'redeploy'])
+
 function mapStoredBlock(block: TaskStoredContentBlock): ContentBlock {
   const mapped: ContentBlock = {
     type: block.type as ContentBlockType,
@@ -361,6 +365,15 @@ export function useChat(
 
   useEffect(() => {
     if (!chatHistory || appliedChatIdRef.current === chatHistory.id) return
+
+    const activeStreamId = chatHistory.activeStreamId
+    const snapshot = chatHistory.streamSnapshot
+
+    if (activeStreamId && !snapshot && !sendingRef.current) {
+      queryClient.invalidateQueries({ queryKey: taskKeys.detail(chatHistory.id) })
+      return
+    }
+
     appliedChatIdRef.current = chatHistory.id
     setMessages(chatHistory.messages.map(mapStoredMessage))
 
@@ -374,11 +387,6 @@ export function useChat(
       }
     }
 
-    // Kick off stream reconnection immediately if there's an active stream.
-    // The stream snapshot was fetched in parallel with the chat history (same
-    // API call), so there's no extra round-trip.
-    const activeStreamId = chatHistory.activeStreamId
-    const snapshot = chatHistory.streamSnapshot
     if (activeStreamId && !sendingRef.current) {
       const gen = ++streamGenRef.current
       const abortController = new AbortController()
@@ -396,8 +404,7 @@ export function useChat(
           const batchEvents = snapshot?.events ?? []
           const streamStatus = snapshot?.status ?? ''
 
-          if (!snapshot || (batchEvents.length === 0 && streamStatus === 'unknown')) {
-            // No snapshot available — stream buffer expired. Clean up.
+          if (batchEvents.length === 0 && streamStatus === 'unknown') {
             const cid = chatIdRef.current
             if (cid) {
               fetch('/api/mothership/chat/stop', {
@@ -462,7 +469,7 @@ export function useChat(
       }
       reconnect()
     }
-  }, [chatHistory, workspaceId])
+  }, [chatHistory, workspaceId, queryClient])
 
   useEffect(() => {
     if (resources.length === 0) {
@@ -686,6 +693,33 @@ export function useChat(
                     onResourceEventRef.current?.()
                   }
                 }
+
+                if (DEPLOY_TOOL_NAMES.has(tc.name) && tc.status === 'success') {
+                  const output = tc.result?.output as Record<string, unknown> | undefined
+                  const deployedWorkflowId = (output?.workflowId as string) ?? undefined
+                  if (deployedWorkflowId && typeof output?.isDeployed === 'boolean') {
+                    const isDeployed = output.isDeployed as boolean
+                    const serverDeployedAt = output.deployedAt
+                      ? new Date(output.deployedAt as string)
+                      : undefined
+                    useWorkflowRegistry
+                      .getState()
+                      .setDeploymentStatus(
+                        deployedWorkflowId,
+                        isDeployed,
+                        isDeployed ? (serverDeployedAt ?? new Date()) : undefined
+                      )
+                    queryClient.invalidateQueries({
+                      queryKey: deploymentKeys.info(deployedWorkflowId),
+                    })
+                    queryClient.invalidateQueries({
+                      queryKey: deploymentKeys.versions(deployedWorkflowId),
+                    })
+                    queryClient.invalidateQueries({
+                      queryKey: workflowKeys.list(workspaceId),
+                    })
+                  }
+                }
               }
 
               break
@@ -1116,11 +1150,6 @@ export function useChat(
   useEffect(() => {
     return () => {
       streamGenRef.current++
-      // Only drop the browser→Sim read; the Sim→Go stream stays open
-      // so the backend can finish persisting. Explicit abort is only
-      // triggered by the stop button via /api/copilot/chat/abort.
-      abortControllerRef.current?.abort()
-      abortControllerRef.current = null
       sendingRef.current = false
     }
   }, [])

apps/sim/lib/copilot/client-sse/handlers.ts

@@ -567,7 +567,6 @@ export const sseHandlers: Record<string, SSEHandler> = {
           }
         }
 
-        // Deploy tools: update deployment status in workflow registry
         if (
           targetState === ClientToolCallState.success &&
           (current.name === 'deploy_api' ||
@@ -579,21 +578,30 @@ export const sseHandlers: Record<string, SSEHandler> = {
             const resultPayload = asRecord(
               data?.result || eventData.result || eventData.data || data?.data
             )
-            const input = asRecord(current.params)
-            const workflowId =
-              (resultPayload?.workflowId as string) ||
-              (input?.workflowId as string) ||
-              useWorkflowRegistry.getState().activeWorkflowId
-            const isDeployed = resultPayload?.isDeployed !== false
-            if (workflowId) {
-              useWorkflowRegistry
-                .getState()
-                .setDeploymentStatus(workflowId, isDeployed, isDeployed ? new Date() : undefined)
-              logger.info('[SSE] Updated deployment status from tool result', {
-                toolName: current.name,
-                workflowId,
-                isDeployed,
-              })
+            if (typeof resultPayload?.isDeployed === 'boolean') {
+              const input = asRecord(current.params)
+              const workflowId =
+                (resultPayload?.workflowId as string) ||
+                (input?.workflowId as string) ||
+                useWorkflowRegistry.getState().activeWorkflowId
+              const isDeployed = resultPayload.isDeployed as boolean
+              const serverDeployedAt = resultPayload.deployedAt
+                ? new Date(resultPayload.deployedAt as string)
+                : undefined
+              if (workflowId) {
+                useWorkflowRegistry
+                  .getState()
+                  .setDeploymentStatus(
+                    workflowId,
+                    isDeployed,
+                    isDeployed ? (serverDeployedAt ?? new Date()) : undefined
+                  )
+                logger.info('[SSE] Updated deployment status from tool result', {
+                  toolName: current.name,
+                  workflowId,
+                  isDeployed,
+                })
+              }
             }
           } catch (err) {
             logger.warn('[SSE] Failed to hydrate deployment status', {

apps/sim/lib/copilot/orchestrator/tool-executor/deployment-tools/deploy.ts

@@ -87,7 +87,10 @@ export async function executeDeployChat(
         return { success: false, error: 'Unauthorized chat access' }
       }
       await db.delete(chat).where(eq(chat.id, existing[0].id))
-      return { success: true, output: { success: true, action: 'undeploy', isDeployed: false } }
+      return {
+        success: true,
+        output: { workflowId, success: true, action: 'undeploy', isChatDeployed: false },
+      }
     }
 
     const { hasAccess } = await checkWorkflowAccessForChatCreation(workflowId, context.userId)
@@ -199,9 +202,11 @@ export async function executeDeployChat(
     return {
       success: true,
       output: {
+        workflowId,
         success: true,
         action: 'deploy',
         isDeployed: true,
+        isChatDeployed: true,
         identifier,
         chatUrl: `${baseUrl}/chat/${identifier}`,
         apiEndpoint: `${baseUrl}/api/workflows/${workflowId}/run`,
@@ -355,6 +360,7 @@ export async function executeRedeploy(
       success: true,
       output: {
         workflowId,
+        isDeployed: true,
         deployedAt: result.deployedAt || null,
         version: result.version,
         apiEndpoint: `${baseUrl}/api/workflows/${workflowId}/run`,