Fix

Author: Sg312

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

@@ -7,7 +7,11 @@ import { z } from 'zod'
 import { getSession } from '@/lib/auth'
 import { resolveOrCreateChat } from '@/lib/copilot/chat-lifecycle'
 import { buildCopilotRequestPayload } from '@/lib/copilot/chat-payload'
-import { createSSEStream, SSE_RESPONSE_HEADERS } from '@/lib/copilot/chat-streaming'
+import {
+  createSSEStream,
+  SSE_RESPONSE_HEADERS,
+  waitForPendingChatStream,
+} from '@/lib/copilot/chat-streaming'
 import type { OrchestratorResult } from '@/lib/copilot/orchestrator/types'
 import { processContextsServer, resolveActiveResourceContext } from '@/lib/copilot/process-contents'
 import { createRequestTracker, createUnauthorizedResponse } from '@/lib/copilot/request-helpers'
@@ -244,6 +248,10 @@ export async function POST(req: NextRequest) {
       { selectedModel: '' }
     )
 
+    if (actualChatId) {
+      await waitForPendingChatStream(actualChatId)
+    }
+
     const stream = createSSEStream({
       requestPayload,
       userId: authenticatedUserId,

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

@@ -20,6 +20,40 @@ const logger = createLogger('CopilotChatStreaming')
 // reach them. Keyed by streamId, cleaned up when the stream completes.
 const activeStreams = new Map<string, AbortController>()
 
+// Tracks in-flight streams by chatId so that a subsequent request for the
+// same chat can wait until the previous stream (and its onComplete / Go-side
+// persistence) has fully settled before forwarding to Go.
+const pendingChatStreams = new Map<string, { promise: Promise<void>; resolve: () => void }>()
+
+function registerPendingChatStream(chatId: string): void {
+  let resolve: () => void
+  const promise = new Promise<void>((r) => {
+    resolve = r
+  })
+  pendingChatStreams.set(chatId, { promise, resolve: resolve! })
+}
+
+function resolvePendingChatStream(chatId: string): void {
+  const entry = pendingChatStreams.get(chatId)
+  if (entry) {
+    entry.resolve()
+    pendingChatStreams.delete(chatId)
+  }
+}
+
+/**
+ * Wait for any in-flight stream on `chatId` to finish before proceeding.
+ * Returns immediately if no stream is active. Gives up after `timeoutMs`.
+ */
+export async function waitForPendingChatStream(
+  chatId: string,
+  timeoutMs = 5_000
+): Promise<void> {
+  const entry = pendingChatStreams.get(chatId)
+  if (!entry) return
+  await Promise.race([entry.promise, new Promise<void>((r) => setTimeout(r, timeoutMs))])
+}
+
 export function abortActiveStream(streamId: string): boolean {
   const controller = activeStreams.get(streamId)
   if (!controller) return false
@@ -112,6 +146,10 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
   const abortController = new AbortController()
   activeStreams.set(streamId, abortController)
 
+  if (chatId) {
+    registerPendingChatStream(chatId)
+  }
+
   return new ReadableStream({
     async start(controller) {
       const encoder = new TextEncoder()
@@ -210,6 +248,9 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
         })
       } finally {
         activeStreams.delete(streamId)
+        if (chatId) {
+          resolvePendingChatStream(chatId)
+        }
         try {
           controller.close()
         } catch {
@@ -219,6 +260,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
     },
     cancel() {
       clientDisconnected = true
+      abortController.abort()
       if (eventWriter) {
         eventWriter.flush().catch(() => {})
       }

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

@@ -610,6 +610,24 @@ export async function executeToolAndReport(
   }
 }
 
+function abortAwareSleep(ms: number, abortSignal?: AbortSignal): Promise<void> {
+  return new Promise<void>((resolve) => {
+    if (abortSignal?.aborted) {
+      resolve()
+      return
+    }
+    const timer = setTimeout(resolve, ms)
+    abortSignal?.addEventListener(
+      'abort',
+      () => {
+        clearTimeout(timer)
+        resolve()
+      },
+      { once: true }
+    )
+  })
+}
+
 export async function waitForToolDecision(
   toolCallId: string,
   timeoutMs: number,
@@ -624,7 +642,7 @@ export async function waitForToolDecision(
     if (decision?.status) {
       return decision
     }
-    await new Promise((resolve) => setTimeout(resolve, interval))
+    await abortAwareSleep(interval, abortSignal)
     interval = Math.min(interval * TOOL_DECISION_POLL_BACKOFF, maxInterval)
   }
   return null
@@ -663,7 +681,7 @@ export async function waitForToolCompletion(
     ) {
       return decision
     }
-    await new Promise((resolve) => setTimeout(resolve, interval))
+    await abortAwareSleep(interval, abortSignal)
     interval = Math.min(interval * TOOL_DECISION_POLL_BACKOFF, maxInterval)
   }
   return null