fix(mothership): reconcile stuck conversation_id against Redis lock to clear stuck-yellow task tiles

copilot_chats.conversation_id has no TTL/heartbeat, so when a stream
process dies before the clear path runs (pod OOM, SIGKILL, uncaught
throw, deploy mid-stream) the column is orphaned and the task tile
renders yellow forever. The Redis lock at copilot:chat-stream-lock:<chatId>
is the canonical liveness signal and self-heals via 60s TTL + 20s
heartbeat, but the mothership APIs weren't consulting it.

Adds read-time reconciliation: a batched MGET helper checks whether
each persisted conversation_id still has a live Redis lock, and both
GET /api/mothership/chats and GET /api/mothership/chats/[chatId]
rewrite the marker to null when the lock has expired. No DB writes;
stuck rows self-heal on next fetch.

Author: waleedlatif1

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

@@ -0,0 +1,201 @@
+/**
+ * @vitest-environment node
+ */
+import { copilotHttpMock, copilotHttpMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockGetAccessibleCopilotChat,
+  mockGetActiveChatStreamIds,
+  mockReadEvents,
+  mockReadFilePreviewSessions,
+  mockGetLatestRunForStream,
+} = vi.hoisted(() => ({
+  mockGetAccessibleCopilotChat: vi.fn(),
+  mockGetActiveChatStreamIds: vi.fn(),
+  mockReadEvents: vi.fn(),
+  mockReadFilePreviewSessions: vi.fn(),
+  mockGetLatestRunForStream: vi.fn(),
+}))
+
+vi.mock('@sim/db', () => ({ db: {} }))
+
+vi.mock('@sim/db/schema', () => ({
+  copilotChats: {
+    id: 'copilotChats.id',
+    userId: 'copilotChats.userId',
+    type: 'copilotChats.type',
+    updatedAt: 'copilotChats.updatedAt',
+    lastSeenAt: 'copilotChats.lastSeenAt',
+  },
+}))
+
+vi.mock('drizzle-orm', () => ({
+  and: vi.fn((...conditions: unknown[]) => ({ type: 'and', conditions })),
+  eq: vi.fn((field: unknown, value: unknown) => ({ type: 'eq', field, value })),
+  sql: Object.assign(
+    vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({
+      type: 'sql',
+      strings,
+      values,
+    })),
+    { raw: vi.fn() }
+  ),
+}))
+
+vi.mock('@/lib/copilot/request/http', () => copilotHttpMock)
+
+vi.mock('@/lib/copilot/chat/lifecycle', () => ({
+  getAccessibleCopilotChat: mockGetAccessibleCopilotChat,
+}))
+
+vi.mock('@/lib/copilot/request/session/abort', () => ({
+  getActiveChatStreamIds: mockGetActiveChatStreamIds,
+}))
+
+vi.mock('@/lib/copilot/request/session/buffer', () => ({
+  readEvents: mockReadEvents,
+}))
+
+vi.mock('@/lib/copilot/request/session/file-preview-session', () => ({
+  readFilePreviewSessions: mockReadFilePreviewSessions,
+}))
+
+vi.mock('@/lib/copilot/async-runs/repository', () => ({
+  getLatestRunForStream: mockGetLatestRunForStream,
+}))
+
+vi.mock('@/lib/copilot/request/session/types', () => ({
+  toStreamBatchEvent: (e: unknown) => e,
+}))
+
+vi.mock('@/lib/copilot/chat/effective-transcript', () => ({
+  buildEffectiveChatTranscript: ({ messages }: { messages: unknown[] }) => messages,
+}))
+
+vi.mock('@/lib/copilot/chat/persisted-message', () => ({
+  normalizeMessage: (m: unknown) => m,
+}))
+
+vi.mock('@/lib/copilot/tasks', () => ({
+  taskPubSub: { publishStatusChanged: vi.fn() },
+}))
+
+vi.mock('@/lib/posthog/server', () => ({
+  captureServerEvent: vi.fn(),
+}))
+
+import { GET } from '@/app/api/mothership/chats/[chatId]/route'
+
+function makeContext(chatId: string) {
+  return { params: Promise.resolve({ chatId }) }
+}
+
+function createRequest(chatId: string) {
+  return new NextRequest(`http://localhost:3000/api/mothership/chats/${chatId}`, {
+    method: 'GET',
+  })
+}
+
+describe('GET /api/mothership/chats/[chatId]', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    copilotHttpMockFns.mockAuthenticateCopilotRequestSessionOnly.mockResolvedValue({
+      userId: 'user-1',
+      isAuthenticated: true,
+    })
+    mockGetActiveChatStreamIds.mockResolvedValue(new Set<string>())
+    mockReadEvents.mockResolvedValue([])
+    mockReadFilePreviewSessions.mockResolvedValue([])
+    mockGetLatestRunForStream.mockResolvedValue(null)
+  })
+
+  it('clears conversationId when the redis lock has expired (stuck-yellow bug)', async () => {
+    mockGetAccessibleCopilotChat.mockResolvedValueOnce({
+      id: 'chat-stuck',
+      type: 'mothership',
+      title: 'Stuck',
+      messages: [],
+      resources: [],
+      conversationId: 'stream-orphaned',
+      createdAt: new Date('2026-05-11T12:00:00Z'),
+      updatedAt: new Date('2026-05-11T12:00:00Z'),
+    })
+    mockGetActiveChatStreamIds.mockResolvedValueOnce(new Set<string>())
+
+    const response = await GET(createRequest('chat-stuck'), makeContext('chat-stuck'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(mockGetActiveChatStreamIds).toHaveBeenCalledWith(['chat-stuck'])
+    expect(body.success).toBe(true)
+    expect(body.chat.conversationId).toBeNull()
+    expect(body.chat.streamSnapshot).toBeUndefined()
+    expect(mockReadEvents).not.toHaveBeenCalled()
+  })
+
+  it('returns the live conversationId when redis confirms the lock', async () => {
+    mockGetAccessibleCopilotChat.mockResolvedValueOnce({
+      id: 'chat-live',
+      type: 'mothership',
+      title: 'Live',
+      messages: [],
+      resources: [],
+      conversationId: 'stream-live',
+      createdAt: new Date('2026-05-11T12:00:00Z'),
+      updatedAt: new Date('2026-05-11T12:00:00Z'),
+    })
+    mockGetActiveChatStreamIds.mockResolvedValueOnce(new Set(['chat-live']))
+    mockGetLatestRunForStream.mockResolvedValueOnce({ status: 'active' })
+
+    const response = await GET(createRequest('chat-live'), makeContext('chat-live'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(body.chat.conversationId).toBe('stream-live')
+    expect(mockReadEvents).toHaveBeenCalledWith('stream-live', '0')
+    expect(body.chat.streamSnapshot).toBeDefined()
+    expect(body.chat.streamSnapshot.status).toBe('active')
+  })
+
+  it('skips the reconciliation lookup when conversationId is already null', async () => {
+    mockGetAccessibleCopilotChat.mockResolvedValueOnce({
+      id: 'chat-idle',
+      type: 'mothership',
+      title: 'Idle',
+      messages: [],
+      resources: [],
+      conversationId: null,
+      createdAt: new Date('2026-05-11T12:00:00Z'),
+      updatedAt: new Date('2026-05-11T12:00:00Z'),
+    })
+
+    const response = await GET(createRequest('chat-idle'), makeContext('chat-idle'))
+    expect(response.status).toBe(200)
+
+    expect(mockGetActiveChatStreamIds).not.toHaveBeenCalled()
+    const body = await response.json()
+    expect(body.chat.conversationId).toBeNull()
+  })
+
+  it('returns 404 when the chat does not exist', async () => {
+    mockGetAccessibleCopilotChat.mockResolvedValueOnce(null)
+
+    const response = await GET(createRequest('chat-missing'), makeContext('chat-missing'))
+    expect(response.status).toBe(404)
+    expect(mockGetActiveChatStreamIds).not.toHaveBeenCalled()
+  })
+
+  it('returns 401 when unauthenticated', async () => {
+    copilotHttpMockFns.mockAuthenticateCopilotRequestSessionOnly.mockResolvedValueOnce({
+      userId: null,
+      isAuthenticated: false,
+    })
+
+    const response = await GET(createRequest('chat-x'), makeContext('chat-x'))
+    expect(response.status).toBe(401)
+    expect(mockGetAccessibleCopilotChat).not.toHaveBeenCalled()
+    expect(mockGetActiveChatStreamIds).not.toHaveBeenCalled()
+  })
+})

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

@@ -20,6 +20,7 @@ import {
   createUnauthorizedResponse,
 } from '@/lib/copilot/request/http'
 import type { FilePreviewSession } from '@/lib/copilot/request/session'
+import { getActiveChatStreamIds } from '@/lib/copilot/request/session/abort'
 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'
@@ -52,23 +53,34 @@ export const GET = withRouteHandler(
         status: string
       } | null = null
 
-      if (chat.conversationId) {
+      // Reconcile the persisted stream marker against the canonical Redis
+      // lock. If `conversation_id` is set but no lock is held, the stream
+      // is no longer running (process died before finalize) — treat the
+      // marker as null so the client doesn't try to reconnect to a dead
+      // stream. Mirrors the same reconciliation in the task list route.
+      const activeIds = chat.conversationId
+        ? await getActiveChatStreamIds([chat.id])
+        : new Set<string>()
+      const liveConversationId =
+        chat.conversationId && activeIds.has(chat.id) ? chat.conversationId : null
+
+      if (liveConversationId) {
         try {
           const [events, previewSessions] = await Promise.all([
-            readEvents(chat.conversationId, '0'),
-            readFilePreviewSessions(chat.conversationId).catch((error) => {
+            readEvents(liveConversationId, '0'),
+            readFilePreviewSessions(liveConversationId).catch((error) => {
               logger.warn('Failed to read preview sessions for mothership chat', {
                 chatId,
-                conversationId: chat.conversationId,
+                conversationId: liveConversationId,
                 error: toError(error).message,
               })
               return []
             }),
           ])
-          const run = await getLatestRunForStream(chat.conversationId, userId).catch((error) => {
+          const run = await getLatestRunForStream(liveConversationId, userId).catch((error) => {
             logger.warn('Failed to fetch latest run for mothership chat snapshot', {
               chatId,
-              conversationId: chat.conversationId,
+              conversationId: liveConversationId,
               error: toError(error).message,
             })
             return null
@@ -87,7 +99,7 @@ export const GET = withRouteHandler(
         } catch (error) {
           logger.warn('Failed to read stream snapshot for mothership chat', {
             chatId,
-            conversationId: chat.conversationId,
+            conversationId: liveConversationId,
             error: toError(error).message,
           })
         }
@@ -100,7 +112,7 @@ export const GET = withRouteHandler(
         : []
       const effectiveMessages = buildEffectiveChatTranscript({
         messages: normalizedMessages,
-        activeStreamId: chat.conversationId || null,
+        activeStreamId: liveConversationId || null,
         ...(streamSnapshot ? { streamSnapshot } : {}),
       })
 
@@ -110,7 +122,7 @@ export const GET = withRouteHandler(
           id: chat.id,
           title: chat.title,
           messages: effectiveMessages,
-          conversationId: chat.conversationId || null,
+          conversationId: liveConversationId || null,
           resources: Array.isArray(chat.resources) ? chat.resources : [],
           createdAt: chat.createdAt,
           updatedAt: chat.updatedAt,