fix(mship): redis replay

Author: Sg312

apps/sim/app/api/mothership/chats/[chatId]/route.test.ts

@@ -173,7 +173,7 @@ describe('GET /api/mothership/chats/[chatId]', () => {
     expect(mockReadEvents).not.toHaveBeenCalled()
   })
 
-  it('returns the live activeStreamId when redis confirms the lock', async () => {
+  it('returns the live activeStreamId without shipping the event snapshot', async () => {
     mockGetAccessibleCopilotChat.mockResolvedValueOnce({
       id: 'chat-live',
       type: 'mothership',
@@ -191,9 +191,11 @@ describe('GET /api/mothership/chats/[chatId]', () => {
     const body = await response.json()
 
     expect(body.chat.activeStreamId).toBe('stream-live')
+    // Events are read only to synthesize the in-flight assistant turn for the
+    // initial paint; the client reconnects to the replay buffer for the rest.
     expect(mockReadEvents).toHaveBeenCalledWith('stream-live', '0')
-    expect(body.chat.streamSnapshot).toBeDefined()
-    expect(body.chat.streamSnapshot.status).toBe('active')
+    expect(body.chat.streamSnapshot).toBeUndefined()
+    expect(mockReadFilePreviewSessions).not.toHaveBeenCalled()
   })
 
   it('uses the Redis lock owner when it differs from a stale persisted streamId', async () => {

apps/sim/app/api/mothership/chats/[chatId]/route.ts

@@ -26,7 +26,6 @@ import {
 } from '@/lib/copilot/request/http'
 import type { FilePreviewSession } from '@/lib/copilot/request/session'
 import { readEvents } from '@/lib/copilot/request/session/buffer'
-import { readFilePreviewSessions } from '@/lib/copilot/request/session/file-preview-session'
 import { type StreamBatchEvent, toStreamBatchEvent } from '@/lib/copilot/request/session/types'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { captureServerEvent } from '@/lib/posthog/server'
@@ -50,7 +49,12 @@ export const GET = withRouteHandler(
         return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })
       }
 
-      let streamSnapshot: {
+      // The Redis replay buffer is read here only to synthesize the in-flight
+      // assistant turn for the initial paint. The raw events are NOT shipped
+      // to the client: when `activeStreamId` is set, the client reconnects to
+      // the replay buffer (from seq 0) via the stream resume endpoint, which
+      // is the source of truth for streaming state.
+      let liveTurnSnapshot: {
         events: StreamBatchEvent[]
         previewSessions: FilePreviewSession[]
         status: string
@@ -64,17 +68,7 @@ export const GET = withRouteHandler(
 
       if (liveStreamId) {
         try {
-          const [events, previewSessions] = await Promise.all([
-            readEvents(liveStreamId, '0'),
-            readFilePreviewSessions(liveStreamId).catch((error) => {
-              logger.warn('Failed to read preview sessions for mothership chat', {
-                chatId,
-                streamId: liveStreamId,
-                error: toError(error).message,
-              })
-              return []
-            }),
-          ])
+          const events = await readEvents(liveStreamId, '0')
           const run = await getLatestRunForStream(liveStreamId, userId).catch((error) => {
             logger.warn('Failed to fetch latest run for mothership chat snapshot', {
               chatId,
@@ -84,9 +78,9 @@ export const GET = withRouteHandler(
             return null
           })
 
-          streamSnapshot = {
+          liveTurnSnapshot = {
             events: events.map(toStreamBatchEvent),
-            previewSessions,
+            previewSessions: [],
             status:
               typeof run?.status === 'string'
                 ? run.status
@@ -111,7 +105,7 @@ export const GET = withRouteHandler(
       const effectiveMessages = buildEffectiveChatTranscript({
         messages: normalizedMessages,
         activeStreamId: liveStreamId,
-        ...(streamSnapshot ? { streamSnapshot } : {}),
+        ...(liveTurnSnapshot ? { streamSnapshot: liveTurnSnapshot } : {}),
       })
 
       return NextResponse.json({
@@ -124,7 +118,6 @@ export const GET = withRouteHandler(
           resources: Array.isArray(chat.resources) ? chat.resources : [],
           createdAt: chat.createdAt,
           updatedAt: chat.updatedAt,
-          ...(streamSnapshot ? { streamSnapshot } : {}),
         },
       })
     } catch (error) {

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

@@ -142,79 +142,61 @@ describe('reconcileLiveAssistantTurn', () => {
 })
 
 describe('selectReconnectReplayState', () => {
-  it('hydrates nonzero cursor replay from a cached live assistant that is ahead', () => {
-    const cachedBlock: ContentBlock = { type: 'text', content: 'Hello world' }
+  it('continues from a nonzero cursor when live streaming state exists in memory', () => {
+    const currentBlock: ContentBlock = { type: 'text', content: 'Hello world' }
 
     const result = selectReconnectReplayState({
       afterCursor: '4',
-      cachedLiveAssistant: {
-        content: 'Hello world',
-        contentBlocks: [cachedBlock],
-      },
-      currentContent: 'Hello',
-      currentBlocks: [],
+      currentContent: 'Hello world',
+      currentBlocks: [currentBlock],
     })
 
     expect(result).toEqual({
       afterCursor: '4',
-      content: 'Hello world',
-      contentBlocks: [cachedBlock],
       preserveExistingState: true,
-      source: 'cache',
+      source: 'live',
     })
   })
 
-  it('resets to replay from the beginning when a nonzero cursor has no usable live cache', () => {
+  it('continues when only blocks carry live state (e.g. tool-only turn)', () => {
     const result = selectReconnectReplayState({
       afterCursor: '4',
-      cachedLiveAssistant: null,
       currentContent: '',
-      currentBlocks: [],
+      currentBlocks: [{ type: 'tool_call', toolCall: { id: 't1', name: 'grep' } } as ContentBlock],
     })
 
     expect(result).toEqual({
-      afterCursor: '0',
-      content: '',
-      contentBlocks: [],
-      preserveExistingState: false,
-      source: 'reset',
+      afterCursor: '4',
+      preserveExistingState: true,
+      source: 'live',
     })
   })
 
-  it('resets when cached live content diverges from the local prefix', () => {
+  it('replays the buffer from seq 0 when a nonzero cursor has no live in-memory state', () => {
     const result = selectReconnectReplayState({
       afterCursor: '4',
-      cachedLiveAssistant: {
-        content: 'Goodbye world',
-        contentBlocks: [{ type: 'text', content: 'Goodbye world' }],
-      },
-      currentContent: 'Hello',
-      currentBlocks: [{ type: 'text', content: 'Hello' }],
+      currentContent: '',
+      currentBlocks: [],
     })
 
     expect(result).toEqual({
       afterCursor: '0',
-      content: '',
-      contentBlocks: [],
       preserveExistingState: false,
       source: 'reset',
     })
   })
 
-  it('resets current state for cursor zero replay', () => {
+  it('resets for cursor zero replay even when local state exists', () => {
     const currentBlock: ContentBlock = { type: 'text', content: 'Hello' }
 
     const result = selectReconnectReplayState({
       afterCursor: '0',
-      cachedLiveAssistant: null,
       currentContent: 'Hello',
       currentBlocks: [currentBlock],
     })
 
     expect(result).toEqual({
       afterCursor: '0',
-      content: '',
-      contentBlocks: [],
       preserveExistingState: false,
       source: 'reset',
     })