feat(transport): replace shared chat transport with mothership-stream module

Author: Sg312

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

@@ -1,6 +1,6 @@
 import { NextResponse } from 'next/server'
 import { getLatestRunForStream } from '@/lib/copilot/async-runs/repository'
-import { abortActiveStream, waitForPendingChatStream } from '@/lib/copilot/chat-streaming'
+import { abortActiveStream } 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'
@@ -55,10 +55,5 @@ export async function POST(request: Request) {
   }
 
   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

@@ -8,15 +8,12 @@ import { getSession } from '@/lib/auth'
 import { getAccessibleCopilotChat, resolveOrCreateChat } from '@/lib/copilot/chat-lifecycle'
 import { buildCopilotRequestPayload } from '@/lib/copilot/chat-payload'
 import {
-  acquirePendingChatStream,
   createSSEStream,
-  releasePendingChatStream,
   requestChatTitle,
   SSE_RESPONSE_HEADERS,
 } from '@/lib/copilot/chat-streaming'
 import { COPILOT_REQUEST_MODES } from '@/lib/copilot/models'
 import { orchestrateCopilotStream } from '@/lib/copilot/orchestrator'
-import { getStreamMeta, readStreamEvents } from '@/lib/copilot/orchestrator/stream/buffer'
 import type { OrchestratorResult } from '@/lib/copilot/orchestrator/types'
 import { resolveActiveResourceContext } from '@/lib/copilot/process-contents'
 import {
@@ -110,9 +107,6 @@ 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
@@ -340,21 +334,6 @@ 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,
@@ -401,7 +380,6 @@ export async function POST(req: NextRequest) {
         titleProvider: provider,
         requestId: tracker.requestId,
         workspaceId: resolvedWorkspaceId,
-        pendingChatStreamAlreadyRegistered: Boolean(actualChatId && stream),
         orchestrateOptions: {
           userId: authenticatedUserId,
           workflowId,
@@ -489,7 +467,6 @@ export async function POST(req: NextRequest) {
           },
         },
       })
-      pendingChatStreamHandedOff = true
 
       return new Response(sseStream, { headers: SSE_RESPONSE_HEADERS })
     }
@@ -595,14 +572,6 @@ 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) {
@@ -649,30 +618,6 @@ export async function GET(req: NextRequest) {
         return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })
       }
 
-      let streamSnapshot: {
-        events: Array<{ eventId: number; streamId: string; event: Record<string, unknown> }>
-        status: string
-      } | null = null
-
-      if (chat.conversationId) {
-        try {
-          const [meta, events] = await Promise.all([
-            getStreamMeta(chat.conversationId),
-            readStreamEvents(chat.conversationId, 0),
-          ])
-          streamSnapshot = {
-            events: events || [],
-            status: meta?.status || 'unknown',
-          }
-        } catch (err) {
-          logger.warn('Failed to read stream snapshot for chat', {
-            chatId,
-            conversationId: chat.conversationId,
-            error: err instanceof Error ? err.message : String(err),
-          })
-        }
-      }
-
       const transformedChat = {
         id: chat.id,
         title: chat.title,
@@ -681,11 +626,10 @@ export async function GET(req: NextRequest) {
         messageCount: Array.isArray(chat.messages) ? chat.messages.length : 0,
         planArtifact: chat.planArtifact || null,
         config: chat.config || null,
-        conversationId: chat.conversationId || null,
+        activeStreamId: chat.conversationId || null,
         resources: Array.isArray(chat.resources) ? chat.resources : [],
         createdAt: chat.createdAt,
         updatedAt: chat.updatedAt,
-        ...(streamSnapshot ? { streamSnapshot } : {}),
       }
 
       logger.info(`Retrieved chat ${chatId}`)

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

@@ -5,24 +5,37 @@
 import { NextRequest } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { getStreamMeta, readStreamEvents, authenticateCopilotRequestSessionOnly } = vi.hoisted(
-  () => ({
-    getStreamMeta: vi.fn(),
-    readStreamEvents: vi.fn(),
-    authenticateCopilotRequestSessionOnly: vi.fn(),
-  })
-)
+const {
+  getLatestRunForStream,
+  readEnvelopes,
+  checkForReplayGap,
+  authenticateCopilotRequestSessionOnly,
+} = vi.hoisted(() => ({
+  getLatestRunForStream: vi.fn(),
+  readEnvelopes: vi.fn(),
+  checkForReplayGap: vi.fn(),
+  authenticateCopilotRequestSessionOnly: vi.fn(),
+}))
+
+vi.mock('@/lib/copilot/async-runs/repository', () => ({
+  getLatestRunForStream,
+}))
 
-vi.mock('@/lib/copilot/orchestrator/stream/buffer', () => ({
-  getStreamMeta,
-  readStreamEvents,
+vi.mock('@/lib/copilot/mothership-stream', () => ({
+  readEnvelopes,
+  checkForReplayGap,
+  encodeSSEEnvelope: (event: Record<string, unknown>) =>
+    new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`),
+  SSE_RESPONSE_HEADERS: {
+    'Content-Type': 'text/event-stream',
+  },
 }))
 
 vi.mock('@/lib/copilot/request-helpers', () => ({
   authenticateCopilotRequestSessionOnly,
 }))
 
-import { GET } from '@/app/api/copilot/chat/stream/route'
+import { GET } from './route'
 
 describe('copilot chat stream replay route', () => {
   beforeEach(() => {
@@ -31,29 +44,32 @@ describe('copilot chat stream replay route', () => {
       userId: 'user-1',
       isAuthenticated: true,
     })
-    readStreamEvents.mockResolvedValue([])
+    readEnvelopes.mockResolvedValue([])
+    checkForReplayGap.mockResolvedValue(null)
   })
 
-  it('stops replay polling when stream meta becomes cancelled', async () => {
-    getStreamMeta
+  it('stops replay polling when run becomes cancelled', async () => {
+    getLatestRunForStream
       .mockResolvedValueOnce({
         status: 'active',
-        userId: 'user-1',
+        executionId: 'exec-1',
+        id: 'run-1',
       })
       .mockResolvedValueOnce({
         status: 'cancelled',
-        userId: 'user-1',
+        executionId: 'exec-1',
+        id: 'run-1',
       })
 
     const response = await GET(
-      new NextRequest('http://localhost:3000/api/copilot/chat/stream?streamId=stream-1')
+      new NextRequest('http://localhost:3000/api/copilot/chat/stream?streamId=stream-1&after=0')
     )
 
     const reader = response.body?.getReader()
     expect(reader).toBeTruthy()
 
     const first = await reader!.read()
     expect(first.done).toBe(true)
-    expect(getStreamMeta).toHaveBeenCalledTimes(2)
+    expect(getLatestRunForStream).toHaveBeenCalledTimes(2)
   })
 })