Fixes

Author: Sg312

apps/sim/app/api/copilot/chat/abort/route.ts

@@ -1,9 +1,12 @@
 import { NextResponse } from 'next/server'
-import { abortActiveStream } from '@/lib/copilot/chat-streaming'
+import { getLatestRunForStream } from '@/lib/copilot/async-runs/repository'
+import { abortActiveStream, waitForPendingChatStream } from '@/lib/copilot/chat-streaming'
 import { SIM_AGENT_API_URL } from '@/lib/copilot/constants'
 import { authenticateCopilotRequestSessionOnly } from '@/lib/copilot/request-helpers'
 import { env } from '@/lib/core/config/env'
 
+const GO_EXPLICIT_ABORT_TIMEOUT_MS = 3000
+
 export async function POST(request: Request) {
   const { userId: authenticatedUserId, isAuthenticated } =
     await authenticateCopilotRequestSessionOnly()
@@ -14,25 +17,48 @@ export async function POST(request: Request) {
 
   const body = await request.json().catch(() => ({}))
   const streamId = typeof body.streamId === 'string' ? body.streamId : ''
+  let chatId = typeof body.chatId === 'string' ? body.chatId : ''
 
   if (!streamId) {
     return NextResponse.json({ error: 'streamId is required' }, { status: 400 })
   }
 
+  if (!chatId) {
+    const run = await getLatestRunForStream(streamId, authenticatedUserId).catch(() => null)
+    if (run?.chatId) {
+      chatId = run.chatId
+    }
+  }
+
   try {
     const headers: Record<string, string> = { 'Content-Type': 'application/json' }
     if (env.COPILOT_API_KEY) {
       headers['x-api-key'] = env.COPILOT_API_KEY
     }
-    await fetch(`${SIM_AGENT_API_URL}/api/streams/explicit-abort`, {
+    const controller = new AbortController()
+    const timeout = setTimeout(() => controller.abort(), GO_EXPLICIT_ABORT_TIMEOUT_MS)
+    const response = await fetch(`${SIM_AGENT_API_URL}/api/streams/explicit-abort`, {
       method: 'POST',
       headers,
-      body: JSON.stringify({ messageId: streamId }),
-    })
+      signal: controller.signal,
+      body: JSON.stringify({
+        messageId: streamId,
+        userId: authenticatedUserId,
+        ...(chatId ? { chatId } : {}),
+      }),
+    }).finally(() => clearTimeout(timeout))
+    if (!response.ok) {
+      throw new Error(`Explicit abort marker request failed: ${response.status}`)
+    }
   } catch {
     // best effort: local abort should still proceed even if Go marker fails
   }
 
   const aborted = await abortActiveStream(streamId)
+  if (chatId) {
+    await waitForPendingChatStream(chatId, GO_EXPLICIT_ABORT_TIMEOUT_MS + 1000, streamId).catch(
+      () => false
+    )
+  }
   return NextResponse.json({ aborted })
 }

apps/sim/app/api/copilot/chat/route.ts

@@ -10,6 +10,7 @@ import { buildCopilotRequestPayload } from '@/lib/copilot/chat-payload'
 import {
   acquirePendingChatStream,
   createSSEStream,
+  releasePendingChatStream,
   requestChatTitle,
   SSE_RESPONSE_HEADERS,
 } from '@/lib/copilot/chat-streaming'
@@ -108,6 +109,10 @@ const ChatMessageSchema = z.object({
  */
 export async function POST(req: NextRequest) {
   const tracker = createRequestTracker()
+  let actualChatId: string | undefined
+  let pendingChatStreamAcquired = false
+  let pendingChatStreamHandedOff = false
+  let pendingChatStreamID: string | undefined
 
   try {
     // Get session to access user information including name
@@ -200,7 +205,7 @@ export async function POST(req: NextRequest) {
 
     let currentChat: any = null
     let conversationHistory: any[] = []
-    let actualChatId = chatId
+    actualChatId = chatId
     const selectedModel = model || 'claude-opus-4-6'
 
     if (chatId || createNewChat) {
@@ -335,6 +340,21 @@ export async function POST(req: NextRequest) {
       })
     } catch {}
 
+    if (stream && actualChatId) {
+      const acquired = await acquirePendingChatStream(actualChatId, userMessageIdToUse)
+      if (!acquired) {
+        return NextResponse.json(
+          {
+            error:
+              'A response is already in progress for this chat. Wait for it to finish or use Stop.',
+          },
+          { status: 409 }
+        )
+      }
+      pendingChatStreamAcquired = true
+      pendingChatStreamID = userMessageIdToUse
+    }
+
     if (actualChatId) {
       const userMsg = {
         id: userMessageIdToUse,
@@ -364,19 +384,6 @@ export async function POST(req: NextRequest) {
       }
     }
 
-    if (stream && actualChatId) {
-      const acquired = await acquirePendingChatStream(actualChatId, userMessageIdToUse)
-      if (!acquired) {
-        return NextResponse.json(
-          {
-            error:
-              'A response is already in progress for this chat. Wait for it to finish or use Stop.',
-          },
-          { status: 409 }
-        )
-      }
-    }
-
     if (stream) {
       const executionId = crypto.randomUUID()
       const runId = crypto.randomUUID()
@@ -482,6 +489,7 @@ export async function POST(req: NextRequest) {
           },
         },
       })
+      pendingChatStreamHandedOff = true
 
       return new Response(sseStream, { headers: SSE_RESPONSE_HEADERS })
     }
@@ -587,6 +595,14 @@ export async function POST(req: NextRequest) {
       },
     })
   } catch (error) {
+    if (
+      actualChatId &&
+      pendingChatStreamAcquired &&
+      !pendingChatStreamHandedOff &&
+      pendingChatStreamID
+    ) {
+      await releasePendingChatStream(actualChatId, pendingChatStreamID).catch(() => {})
+    }
     const duration = tracker.getDuration()
 
     if (error instanceof z.ZodError) {

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

@@ -441,18 +441,34 @@ export function useChat(
   }, [isHomePage])
 
   useEffect(() => {
-    if (!chatHistory || appliedChatIdRef.current === chatHistory.id) return
+    if (!chatHistory) return
 
     const activeStreamId = chatHistory.activeStreamId
     const snapshot = chatHistory.streamSnapshot
+    const staleKey = `${chatHistory.id}:stale-stream`
 
     if (activeStreamId && !snapshot && !sendingRef.current) {
-      appliedChatIdRef.current = chatHistory.id
+      if (appliedChatIdRef.current === staleKey) return
+      appliedChatIdRef.current = staleKey
+      setMessages(chatHistory.messages.map(mapStoredMessage))
+
+      const persistedResources = chatHistory.resources.filter((r) => r.id !== 'streaming-file')
+      if (persistedResources.length > 0) {
+        setResources(persistedResources)
+        setActiveResourceId(persistedResources[persistedResources.length - 1].id)
+        for (const resource of persistedResources) {
+          if (resource.type !== 'workflow') continue
+          ensureWorkflowInRegistry(resource.id, resource.title, workspaceId)
+        }
+      }
+
       setError(RECONNECT_TAIL_ERROR)
       queryClient.invalidateQueries({ queryKey: taskKeys.detail(chatHistory.id) })
       return
     }
 
+    if (appliedChatIdRef.current === chatHistory.id) return
+
     appliedChatIdRef.current = chatHistory.id
     const mappedMessages = chatHistory.messages.map(mapStoredMessage)
     const shouldPreserveActiveStreamingMessage =

apps/sim/lib/copilot/async-runs/repository.ts

@@ -86,6 +86,20 @@ export async function getLatestRunForExecution(executionId: string) {
   return run ?? null
 }
 
+export async function getLatestRunForStream(streamId: string, userId?: string) {
+  const conditions = userId
+    ? and(eq(copilotRuns.streamId, streamId), eq(copilotRuns.userId, userId))
+    : eq(copilotRuns.streamId, streamId)
+  const [run] = await db
+    .select()
+    .from(copilotRuns)
+    .where(conditions)
+    .orderBy(desc(copilotRuns.startedAt))
+    .limit(1)
+
+  return run ?? null
+}
+
 export async function getRunSegment(runId: string) {
   const [run] = await db.select().from(copilotRuns).where(eq(copilotRuns.id, runId)).limit(1)
   return run ?? null

apps/sim/lib/copilot/chat-streaming.ts

@@ -8,6 +8,7 @@ import type { OrchestrateStreamOptions } from '@/lib/copilot/orchestrator'
 import { orchestrateCopilotStream } from '@/lib/copilot/orchestrator'
 import {
   createStreamEventWriter,
+  getStreamMeta,
   resetStreamBuffer,
   setStreamMeta,
 } from '@/lib/copilot/orchestrator/stream/buffer'
@@ -66,15 +67,46 @@ function getStreamAbortKey(streamId: string): string {
  */
 export async function waitForPendingChatStream(
   chatId: string,
-  timeoutMs = 5_000
+  timeoutMs = 5_000,
+  expectedStreamId?: string
 ): Promise<boolean> {
-  const entry = pendingChatStreams.get(chatId)
-  if (!entry) return true
+  const redis = getRedisClient()
+  const deadline = Date.now() + timeoutMs
+
+  for (;;) {
+    const entry = pendingChatStreams.get(chatId)
+    const localPending = !!entry && (!expectedStreamId || entry.streamId === expectedStreamId)
+
+    if (redis) {
+      try {
+        const ownerStreamId = await redis.get(getChatStreamLockKey(chatId))
+        const lockReleased =
+          !ownerStreamId || (expectedStreamId !== undefined && ownerStreamId !== expectedStreamId)
+        if (!localPending && lockReleased) {
+          return true
+        }
+      } catch (error) {
+        logger.warn('Failed to check distributed chat stream lock while waiting', {
+          chatId,
+          expectedStreamId,
+          error: error instanceof Error ? error.message : String(error),
+        })
+      }
+    } else if (!localPending) {
+      return true
+    }
 
-  return await Promise.race([
-    entry.promise.then(() => true),
-    new Promise<boolean>((r) => setTimeout(() => r(false), timeoutMs)),
-  ])
+    if (Date.now() >= deadline) return false
+    await new Promise((resolve) => setTimeout(resolve, 200))
+  }
+}
+
+export async function releasePendingChatStream(chatId: string, streamId: string): Promise<void> {
+  const redis = getRedisClient()
+  if (redis) {
+    await releaseLock(getChatStreamLockKey(chatId), streamId).catch(() => false)
+  }
+  resolvePendingChatStream(chatId, streamId)
 }
 
 export async function acquirePendingChatStream(
@@ -86,14 +118,36 @@ export async function acquirePendingChatStream(
   if (redis) {
     const deadline = Date.now() + timeoutMs
     for (;;) {
-      const acquired = await acquireLock(
-        getChatStreamLockKey(chatId),
-        streamId,
-        CHAT_STREAM_LOCK_TTL_SECONDS
-      )
-      if (acquired) {
-        registerPendingChatStream(chatId, streamId)
-        return true
+      try {
+        const acquired = await acquireLock(
+          getChatStreamLockKey(chatId),
+          streamId,
+          CHAT_STREAM_LOCK_TTL_SECONDS
+        )
+        if (acquired) {
+          registerPendingChatStream(chatId, streamId)
+          return true
+        }
+        if (!pendingChatStreams.has(chatId)) {
+          const ownerStreamId = await redis.get(getChatStreamLockKey(chatId))
+          if (ownerStreamId) {
+            const ownerMeta = await getStreamMeta(ownerStreamId)
+            const ownerTerminal =
+              ownerMeta?.status === 'complete' ||
+              ownerMeta?.status === 'error' ||
+              ownerMeta?.status === 'cancelled'
+            if (ownerTerminal) {
+              await releaseLock(getChatStreamLockKey(chatId), ownerStreamId).catch(() => false)
+              continue
+            }
+          }
+        }
+      } catch (error) {
+        logger.warn('Distributed chat stream lock failed; retrying distributed coordination', {
+          chatId,
+          streamId,
+          error: error instanceof Error ? error.message : String(error),
+        })
       }
       if (Date.now() >= deadline) return false
       await new Promise((resolve) => setTimeout(resolve, 200))

apps/sim/lib/copilot/orchestrator/sse/handlers/tool-execution.ts

@@ -117,6 +117,20 @@ const FORMAT_TO_CONTENT_TYPE: Record<OutputFormat, string> = {
   html: 'text/html',
 }
 
+function normalizeOutputWorkspaceFileName(outputPath: string): string {
+  const trimmed = outputPath.trim().replace(/^\/+/, '')
+  const withoutPrefix = trimmed.startsWith('files/') ? trimmed.slice('files/'.length) : trimmed
+  if (!withoutPrefix) {
+    throw new Error('outputPath must include a file name, e.g. "files/result.json"')
+  }
+  if (withoutPrefix.includes('/')) {
+    throw new Error(
+      'outputPath must target a flat workspace file, e.g. "files/result.json". Nested paths like "files/reports/result.json" are not supported.'
+    )
+  }
+  return withoutPrefix
+}
+
 function resolveOutputFormat(fileName: string, explicit?: string): OutputFormat {
   if (explicit && explicit in FORMAT_TO_CONTENT_TYPE) return explicit as OutputFormat
   const ext = fileName.slice(fileName.lastIndexOf('.')).toLowerCase()
@@ -153,10 +167,10 @@ async function maybeWriteOutputToFile(
 
   const explicitFormat =
     (params?.outputFormat as string | undefined) ?? (args?.outputFormat as string | undefined)
-  const fileName = outputPath.replace(/^files\//, '')
-  const format = resolveOutputFormat(fileName, explicitFormat)
 
   try {
+    const fileName = normalizeOutputWorkspaceFileName(outputPath)
+    const format = resolveOutputFormat(fileName, explicitFormat)
     if (context.abortSignal?.aborted) {
       throw new Error('Request aborted before tool mutation could be applied')
     }
@@ -193,12 +207,16 @@ async function maybeWriteOutputToFile(
       },
     }
   } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
     logger.warn('Failed to write tool output to file', {
       toolName,
       outputPath,
-      error: err instanceof Error ? err.message : String(err),
+      error: message,
     })
-    return result
+    return {
+      success: false,
+      error: `Failed to write output file: ${message}`,
+    }
   }
 }