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

* 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. * test(mothership): clarify test name to reflect that getActiveChatStreamIds is called with empty candidateIds * address comments * fix state machine issue * cleanup code and fix types --------- Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>
2026-09-24 15:45:35 +08:00 · 2026-05-12 18:54:49 -07:00
parent d5c2ead5d4
commit 773cd84e3f
13 changed files with 955 additions and 37 deletions
@@ -0,0 +1,245 @@
+/**
+ * @vitest-environment node
+ */
+import { copilotHttpMock, copilotHttpMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockGetAccessibleCopilotChat,
+  mockReconcileChatStreamMarkers,
+  mockReadEvents,
+  mockReadFilePreviewSessions,
+  mockGetLatestRunForStream,
+} = vi.hoisted(() => ({
+  mockGetAccessibleCopilotChat: vi.fn(),
+  mockReconcileChatStreamMarkers: 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/chat/stream-liveness', () => ({
+  reconcileChatStreamMarkers: mockReconcileChatStreamMarkers,
+}))
+
+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,
+    })
+    mockReconcileChatStreamMarkers.mockImplementation(
+      async (candidates: Array<{ chatId: string; streamId: string | null }>) =>
+        new Map(
+          candidates.map((candidate) => [
+            candidate.chatId,
+            {
+              chatId: candidate.chatId,
+              streamId: candidate.streamId,
+              status: candidate.streamId ? 'active' : 'inactive',
+            },
+          ])
+        )
+    )
+    mockReadEvents.mockResolvedValue([])
+    mockReadFilePreviewSessions.mockResolvedValue([])
+    mockGetLatestRunForStream.mockResolvedValue(null)
+  })
+
+  it('clears activeStreamId 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'),
+    })
+    mockReconcileChatStreamMarkers.mockResolvedValueOnce(
+      new Map([['chat-stuck', { chatId: 'chat-stuck', streamId: null, status: 'inactive' }]])
+    )
+
+    const response = await GET(createRequest('chat-stuck'), makeContext('chat-stuck'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(mockReconcileChatStreamMarkers).toHaveBeenCalledWith(
+      [{ chatId: 'chat-stuck', streamId: 'stream-orphaned' }],
+      { repairVerifiedStaleMarkers: true }
+    )
+    expect(body.success).toBe(true)
+    expect(body.chat.activeStreamId).toBeNull()
+    expect(body.chat.streamSnapshot).toBeUndefined()
+    expect(mockReadEvents).not.toHaveBeenCalled()
+  })
+
+  it('returns the live activeStreamId 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'),
+    })
+    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.activeStreamId).toBe('stream-live')
+    expect(mockReadEvents).toHaveBeenCalledWith('stream-live', '0')
+    expect(body.chat.streamSnapshot).toBeDefined()
+    expect(body.chat.streamSnapshot.status).toBe('active')
+  })
+
+  it('uses the Redis lock owner when it differs from a stale persisted streamId', async () => {
+    mockGetAccessibleCopilotChat.mockResolvedValueOnce({
+      id: 'chat-mismatch',
+      type: 'mothership',
+      title: 'Mismatch',
+      messages: [],
+      resources: [],
+      conversationId: 'stream-stale',
+      createdAt: new Date('2026-05-11T12:00:00Z'),
+      updatedAt: new Date('2026-05-11T12:00:00Z'),
+    })
+    mockReconcileChatStreamMarkers.mockResolvedValueOnce(
+      new Map([
+        ['chat-mismatch', { chatId: 'chat-mismatch', streamId: 'stream-live', status: 'active' }],
+      ])
+    )
+
+    const response = await GET(createRequest('chat-mismatch'), makeContext('chat-mismatch'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(body.chat.activeStreamId).toBe('stream-live')
+    expect(mockReadEvents).toHaveBeenCalledWith('stream-live', '0')
+  })
+
+  it('returns null when the persisted stream marker 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(mockReconcileChatStreamMarkers).toHaveBeenCalledWith(
+      [{ chatId: 'chat-idle', streamId: null }],
+      { repairVerifiedStaleMarkers: true }
+    )
+    const body = await response.json()
+    expect(body.chat.activeStreamId).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(mockReconcileChatStreamMarkers).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(mockReconcileChatStreamMarkers).not.toHaveBeenCalled()
+  })
+})
@@ -14,6 +14,7 @@ import { getLatestRunForStream } from '@/lib/copilot/async-runs/repository'
 import { buildEffectiveChatTranscript } from '@/lib/copilot/chat/effective-transcript'
 import { getAccessibleCopilotChat } from '@/lib/copilot/chat/lifecycle'
 import { normalizeMessage } from '@/lib/copilot/chat/persisted-message'
+import { reconcileChatStreamMarkers } from '@/lib/copilot/chat/stream-liveness'
 import {
  authenticateCopilotRequestSessionOnly,
  createInternalServerErrorResponse,
@@ -52,23 +53,29 @@ export const GET = withRouteHandler(
        status: string
      } | null = null

-      if (chat.conversationId) {
+      const reconciledMarkers = await reconcileChatStreamMarkers(
+        [{ chatId: chat.id, streamId: chat.conversationId }],
+        { repairVerifiedStaleMarkers: true }
+      )
+      const liveStreamId = reconciledMarkers.get(chat.id)?.streamId ?? null
+
+      if (liveStreamId) {
        try {
          const [events, previewSessions] = await Promise.all([
-            readEvents(chat.conversationId, '0'),
-            readFilePreviewSessions(chat.conversationId).catch((error) => {
+            readEvents(liveStreamId, '0'),
+            readFilePreviewSessions(liveStreamId).catch((error) => {
              logger.warn('Failed to read preview sessions for mothership chat', {
                chatId,
-                conversationId: chat.conversationId,
+                streamId: liveStreamId,
                error: toError(error).message,
              })
              return []
            }),
          ])
-          const run = await getLatestRunForStream(chat.conversationId, userId).catch((error) => {
+          const run = await getLatestRunForStream(liveStreamId, userId).catch((error) => {
            logger.warn('Failed to fetch latest run for mothership chat snapshot', {
              chatId,
-              conversationId: chat.conversationId,
+              streamId: liveStreamId,
              error: toError(error).message,
            })
            return null
@@ -87,7 +94,7 @@ export const GET = withRouteHandler(
        } catch (error) {
          logger.warn('Failed to read stream snapshot for mothership chat', {
            chatId,
-            conversationId: chat.conversationId,
+            streamId: liveStreamId,
            error: toError(error).message,
          })
        }
@@ -100,7 +107,7 @@ export const GET = withRouteHandler(
        : []
      const effectiveMessages = buildEffectiveChatTranscript({
        messages: normalizedMessages,
-        activeStreamId: chat.conversationId || null,
+        activeStreamId: liveStreamId,
        ...(streamSnapshot ? { streamSnapshot } : {}),
      })

@@ -110,7 +117,7 @@ export const GET = withRouteHandler(
          id: chat.id,
          title: chat.title,
          messages: effectiveMessages,
-          conversationId: chat.conversationId || null,
+          activeStreamId: liveStreamId,
          resources: Array.isArray(chat.resources) ? chat.resources : [],
          createdAt: chat.createdAt,
          updatedAt: chat.updatedAt,
@@ -0,0 +1,225 @@
+/**
+ * @vitest-environment node
+ */
+import { copilotHttpMock, copilotHttpMockFns, permissionsMock } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockSelect, mockFrom, mockWhere, mockOrderBy, mockReconcileChatStreamMarkers } = vi.hoisted(
+  () => ({
+    mockSelect: vi.fn(),
+    mockFrom: vi.fn(),
+    mockWhere: vi.fn(),
+    mockOrderBy: vi.fn(),
+    mockReconcileChatStreamMarkers: vi.fn(),
+  })
+)
+
+vi.mock('@sim/db', () => ({
+  db: {
+    select: mockSelect,
+  },
+}))
+
+vi.mock('@sim/db/schema', () => ({
+  copilotChats: {
+    id: 'copilotChats.id',
+    title: 'copilotChats.title',
+    userId: 'copilotChats.userId',
+    workspaceId: 'copilotChats.workspaceId',
+    type: 'copilotChats.type',
+    updatedAt: 'copilotChats.updatedAt',
+    conversationId: 'copilotChats.conversationId',
+    lastSeenAt: 'copilotChats.lastSeenAt',
+  },
+}))
+
+vi.mock('drizzle-orm', () => ({
+  and: vi.fn((...conditions: unknown[]) => ({ type: 'and', conditions })),
+  desc: vi.fn((field: unknown) => ({ type: 'desc', field })),
+  eq: vi.fn((field: unknown, value: unknown) => ({ type: 'eq', field, value })),
+}))
+
+vi.mock('@/lib/copilot/request/http', () => copilotHttpMock)
+vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock)
+
+vi.mock('@/lib/copilot/chat/stream-liveness', () => ({
+  reconcileChatStreamMarkers: mockReconcileChatStreamMarkers,
+}))
+
+vi.mock('@/lib/copilot/tasks', () => ({
+  taskPubSub: { publishStatusChanged: vi.fn() },
+}))
+
+vi.mock('@/lib/posthog/server', () => ({
+  captureServerEvent: vi.fn(),
+}))
+
+import { GET } from '@/app/api/mothership/chats/route'
+
+function createRequest(workspaceId: string) {
+  return new NextRequest(`http://localhost:3000/api/mothership/chats?workspaceId=${workspaceId}`, {
+    method: 'GET',
+  })
+}
+
+describe('GET /api/mothership/chats', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+
+    copilotHttpMockFns.mockAuthenticateCopilotRequestSessionOnly.mockResolvedValue({
+      userId: 'user-1',
+      isAuthenticated: true,
+    })
+
+    mockOrderBy.mockResolvedValue([])
+    mockWhere.mockReturnValue({ orderBy: mockOrderBy })
+    mockFrom.mockReturnValue({ where: mockWhere })
+    mockSelect.mockReturnValue({ from: mockFrom })
+
+    mockReconcileChatStreamMarkers.mockImplementation(
+      async (candidates: Array<{ chatId: string; streamId: string | null }>) =>
+        new Map(
+          candidates.map((candidate) => [
+            candidate.chatId,
+            {
+              chatId: candidate.chatId,
+              streamId: candidate.streamId,
+              status: candidate.streamId ? 'active' : 'inactive',
+            },
+          ])
+        )
+    )
+  })
+
+  it('clears activeStreamId on chats whose redis lock has expired (stuck-yellow bug)', async () => {
+    const now = new Date('2026-05-11T12:00:00Z')
+    mockOrderBy.mockResolvedValueOnce([
+      {
+        id: 'chat-stuck',
+        title: 'Stuck chat',
+        updatedAt: now,
+        activeStreamId: 'stream-orphaned',
+        lastSeenAt: null,
+      },
+      {
+        id: 'chat-live',
+        title: 'Live chat',
+        updatedAt: now,
+        activeStreamId: 'stream-live',
+        lastSeenAt: null,
+      },
+      {
+        id: 'chat-idle',
+        title: 'Idle chat',
+        updatedAt: now,
+        activeStreamId: null,
+        lastSeenAt: null,
+      },
+    ])
+    mockReconcileChatStreamMarkers.mockResolvedValueOnce(
+      new Map([
+        ['chat-stuck', { chatId: 'chat-stuck', streamId: null, status: 'inactive' }],
+        ['chat-live', { chatId: 'chat-live', streamId: 'stream-live', status: 'active' }],
+        ['chat-idle', { chatId: 'chat-idle', streamId: null, status: 'inactive' }],
+      ])
+    )
+
+    const response = await GET(createRequest('ws-1'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(mockReconcileChatStreamMarkers).toHaveBeenCalledWith(
+      [
+        { chatId: 'chat-stuck', streamId: 'stream-orphaned' },
+        { chatId: 'chat-live', streamId: 'stream-live' },
+        { chatId: 'chat-idle', streamId: null },
+      ],
+      { repairVerifiedStaleMarkers: true }
+    )
+    expect(body.success).toBe(true)
+    expect(body.data).toEqual([
+      expect.objectContaining({ id: 'chat-stuck', activeStreamId: null }),
+      expect.objectContaining({ id: 'chat-live', activeStreamId: 'stream-live' }),
+      expect.objectContaining({ id: 'chat-idle', activeStreamId: null }),
+    ])
+  })
+
+  it('preserves chats when no chat has a stream marker set', async () => {
+    const now = new Date('2026-05-11T12:00:00Z')
+    mockOrderBy.mockResolvedValueOnce([
+      { id: 'chat-1', title: null, updatedAt: now, activeStreamId: null, lastSeenAt: null },
+      { id: 'chat-2', title: null, updatedAt: now, activeStreamId: null, lastSeenAt: null },
+    ])
+
+    const response = await GET(createRequest('ws-1'))
+    expect(response.status).toBe(200)
+
+    expect(mockReconcileChatStreamMarkers).toHaveBeenCalledWith(
+      [
+        { chatId: 'chat-1', streamId: null },
+        { chatId: 'chat-2', streamId: null },
+      ],
+      { repairVerifiedStaleMarkers: true }
+    )
+    const body = await response.json()
+    expect(body.data).toEqual([
+      expect.objectContaining({ id: 'chat-1', activeStreamId: null }),
+      expect.objectContaining({ id: 'chat-2', activeStreamId: null }),
+    ])
+  })
+
+  it('leaves activeStreamId untouched when redis confirms every lock is live', async () => {
+    const now = new Date('2026-05-11T12:00:00Z')
+    mockOrderBy.mockResolvedValueOnce([
+      { id: 'chat-a', title: null, updatedAt: now, activeStreamId: 'stream-a', lastSeenAt: null },
+      { id: 'chat-b', title: null, updatedAt: now, activeStreamId: 'stream-b', lastSeenAt: null },
+    ])
+
+    const response = await GET(createRequest('ws-1'))
+    const body = await response.json()
+
+    expect(body.data).toEqual([
+      expect.objectContaining({ id: 'chat-a', activeStreamId: 'stream-a' }),
+      expect.objectContaining({ id: 'chat-b', activeStreamId: 'stream-b' }),
+    ])
+  })
+
+  it('uses Redis lock owner when it differs from a stale activeStreamId', async () => {
+    const now = new Date('2026-05-11T12:00:00Z')
+    mockOrderBy.mockResolvedValueOnce([
+      {
+        id: 'chat-mismatch',
+        title: null,
+        updatedAt: now,
+        activeStreamId: 'stream-stale',
+        lastSeenAt: null,
+      },
+    ])
+    mockReconcileChatStreamMarkers.mockResolvedValueOnce(
+      new Map([
+        ['chat-mismatch', { chatId: 'chat-mismatch', streamId: 'stream-live', status: 'active' }],
+      ])
+    )
+
+    const response = await GET(createRequest('ws-1'))
+    expect(response.status).toBe(200)
+    const body = await response.json()
+
+    expect(body.data).toEqual([
+      expect.objectContaining({ id: 'chat-mismatch', activeStreamId: 'stream-live' }),
+    ])
+  })
+
+  it('returns 401 when unauthenticated', async () => {
+    copilotHttpMockFns.mockAuthenticateCopilotRequestSessionOnly.mockResolvedValueOnce({
+      userId: null,
+      isAuthenticated: false,
+    })
+
+    const response = await GET(createRequest('ws-1'))
+    expect(response.status).toBe(401)
+    expect(mockSelect).not.toHaveBeenCalled()
+    expect(mockReconcileChatStreamMarkers).not.toHaveBeenCalled()
+  })
+})
@@ -8,6 +8,7 @@ import {
  listMothershipChatsContract,
 } from '@/lib/api/contracts/mothership-tasks'
 import { parseRequest } from '@/lib/api/server'
+import { reconcileChatStreamMarkers } from '@/lib/copilot/chat/stream-liveness'
 import {
  authenticateCopilotRequestSessionOnly,
  createInternalServerErrorResponse,
@@ -55,7 +56,16 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
      )
      .orderBy(desc(copilotChats.updatedAt))

-    return NextResponse.json({ success: true, data: chats })
+    const streamMarkers = await reconcileChatStreamMarkers(
+      chats.map((c) => ({ chatId: c.id, streamId: c.activeStreamId })),
+      { repairVerifiedStaleMarkers: true }
+    )
+    const reconciled = chats.map((c) => {
+      const activeStreamId = streamMarkers.get(c.id)?.streamId ?? null
+      return activeStreamId === c.activeStreamId ? c : { ...c, activeStreamId }
+    })
+
+    return NextResponse.json({ success: true, data: reconciled })
  } catch (error) {
    logger.error('Error fetching mothership chats:', error)
    return createInternalServerErrorResponse('Failed to fetch chats')
@@ -3499,11 +3499,11 @@ export function useChat(
  processSSEStreamRef.current = processSSEStream

  const getActiveStreamIdForChat = useCallback(
-    async (chatId: string, signal?: AbortSignal): Promise<string | null> => {
+    async (
+      chatId: string,
+      signal?: AbortSignal
+    ): Promise<{ loaded: boolean; streamId: string | null }> => {
      const cached = queryClient.getQueryData<TaskChatHistory>(taskKeys.detail(chatId))
-      if (cached?.activeStreamId) {
-        return cached.activeStreamId
-      }

      try {
        const fetchSignal = combineAbortSignals(
@@ -3511,15 +3511,15 @@ export function useChat(
          createTimeoutSignal(CHAT_HISTORY_RECOVERY_TIMEOUT_MS)
        )
        const history = await fetchChatHistory(chatId, fetchSignal)
-        if (signal?.aborted || fetchSignal?.aborted) return null
+        if (signal?.aborted || fetchSignal?.aborted) return { loaded: false, streamId: null }
        queryClient.setQueryData(taskKeys.detail(chatId), history)
-        return history.activeStreamId ?? null
+        return { loaded: true, streamId: history.activeStreamId ?? null }
      } catch (error) {
        logger.warn('Failed to load chat history while recovering stream', {
          chatId,
          error: toError(error).message,
        })
-        return null
+        return { loaded: false, streamId: cached?.activeStreamId ?? null }
      }
    },
    [queryClient]
@@ -4032,12 +4032,12 @@ export function useChat(
          !recoveryController.signal.aborted

        const cached = queryClient.getQueryData<TaskChatHistory>(taskKeys.detail(chatId))
-        let streamId =
+        const fallbackStreamId =
          streamIdRef.current ?? activeTurnRef.current?.userMessageId ?? cached?.activeStreamId
-        if (!streamId) {
-          streamId =
-            (await getActiveStreamIdForChat(chatId, recoveryController.signal)) ?? undefined
-        }
+        const loadedStream = await getActiveStreamIdForChat(chatId, recoveryController.signal)
+        const streamId = loadedStream.loaded
+          ? (loadedStream.streamId ?? undefined)
+          : fallbackStreamId
        if (
          !isSameRecoverySubject() ||
          streamGenRef.current !== observedGeneration ||
@@ -100,7 +100,7 @@ describe('tasks query boundary parsing', () => {
          id: 'chat-1',
          title: 'Task history',
          messages: [],
-          conversationId: 'stream-1',
+          activeStreamId: 'stream-1',
          resources: [{ type: 'file', id: 'file-1', title: 'Spec.md' }],
          streamSnapshot: {
            events: [],
@@ -144,7 +144,7 @@ describe('tasks query boundary parsing', () => {
      )

    await expect(fetchChatHistory('chat-1')).rejects.toThrow(
-      'Invalid copilot chat response: chat.resources[0].type is invalid'
+      'Invalid chat response: chat.resources[0].type is invalid'
    )
  })

@@ -57,8 +57,6 @@ export const taskKeys = {
  detail: (chatId: string | undefined) => [...taskKeys.details(), chatId ?? ''] as const,
 }

-type ChatHistorySource = 'copilot' | 'mothership'
-
 function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value) && typeof value === 'object' && !Array.isArray(value)
 }
@@ -150,30 +148,28 @@ function parseStrictStreamSnapshot(
  return snapshot
 }

-function parseChatHistory(value: unknown, source: ChatHistorySource): TaskChatHistory {
-  const responseContext = `Invalid ${source} chat response`
+function parseChatHistory(value: unknown): TaskChatHistory {
+  const responseContext = 'Invalid chat response'
  const chatContext = `${responseContext}: chat`

  assertValid(isRecord(value), `${responseContext}: body must be an object`)
  assertValid(isRecord(value.chat), `${chatContext} must be an object`)

  const chat = value.chat
-  const activeStreamField = source === 'mothership' ? 'conversationId' : 'activeStreamId'
-  const activeStreamId = chat[activeStreamField]

  assertValid(typeof chat.id === 'string', `${chatContext}.id must be a string`)
  assertValid(isNullableString(chat.title), `${chatContext}.title must be a string or null`)
  assertValid(Array.isArray(chat.messages), `${chatContext}.messages must be an array`)
  assertValid(
-    isNullableString(activeStreamId),
-    `${chatContext}.${activeStreamField} must be a string or null`
+    isNullableString(chat.activeStreamId),
+    `${chatContext}.activeStreamId must be a string or null`
  )

  return {
    id: chat.id,
    title: chat.title,
    messages: normalizeMessages(chat.messages),
-    activeStreamId,
+    activeStreamId: chat.activeStreamId,
    resources: parseResources(chat.resources, `${chatContext}.resources`),
    streamSnapshot: parseStrictStreamSnapshot(chat.streamSnapshot, `${chatContext}.streamSnapshot`),
  }
@@ -233,7 +229,7 @@ export async function fetchChatHistory(
      params: { chatId },
      signal,
    })
-    return parseChatHistory(data, 'mothership')
+    return parseChatHistory(data)
  } catch (error) {
    if (!isApiClientError(error)) throw error
    // Fall through to the legacy copilot-shape alias on any HTTP error (typically 404
@@ -251,7 +247,7 @@ export async function fetchChatHistory(
    throw new Error('Failed to load chat')
  }

-  return parseChatHistory(await copilotRes.json(), 'copilot')
+  return parseChatHistory(await copilotRes.json())
 }

 /**
@@ -283,7 +283,7 @@ export const getMothershipChatResponseSchema = z.object({
      id: z.string(),
      title: z.string().nullable(),
      messages: z.array(z.unknown()),
-      conversationId: z.string().nullable(),
+      activeStreamId: z.string().nullable(),
      resources: z.array(z.unknown()),
      createdAt: z.union([z.string(), z.date()]).nullable().optional(),
      updatedAt: z.union([z.string(), z.date()]).nullable().optional(),
@@ -0,0 +1,154 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockAnd, mockEq, mockGetChatStreamLockOwners, mockSet, mockUpdate, mockWhere } = vi.hoisted(
+  () => ({
+    mockAnd: vi.fn((...conditions: unknown[]) => ({ type: 'and', conditions })),
+    mockEq: vi.fn((field: unknown, value: unknown) => ({ type: 'eq', field, value })),
+    mockGetChatStreamLockOwners: vi.fn(),
+    mockSet: vi.fn(),
+    mockUpdate: vi.fn(),
+    mockWhere: vi.fn(),
+  })
+)
+
+vi.mock('@sim/db', () => ({
+  db: { update: mockUpdate },
+}))
+
+vi.mock('@sim/db/schema', () => ({
+  copilotChats: {
+    id: 'copilotChats.id',
+    conversationId: 'copilotChats.conversationId',
+  },
+}))
+
+vi.mock('drizzle-orm', () => ({
+  and: mockAnd,
+  eq: mockEq,
+}))
+
+vi.mock('@/lib/copilot/request/session', () => ({
+  getChatStreamLockOwners: mockGetChatStreamLockOwners,
+}))
+
+import { reconcileChatStreamMarkers } from '@/lib/copilot/chat/stream-liveness'
+
+describe('reconcileChatStreamMarkers', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockSet.mockReturnValue({ where: mockWhere })
+    mockUpdate.mockReturnValue({ set: mockSet })
+    mockWhere.mockResolvedValue(undefined)
+    mockGetChatStreamLockOwners.mockResolvedValue({
+      status: 'verified',
+      ownersByChatId: new Map<string, string>(),
+    })
+  })
+
+  it('clears a persisted stream marker when Redis verifies no lock owner exists', async () => {
+    const markers = await reconcileChatStreamMarkers([
+      { chatId: 'chat-stuck', streamId: 'stream-orphaned' },
+    ])
+
+    expect(mockGetChatStreamLockOwners).toHaveBeenCalledWith(['chat-stuck'])
+    expect(markers.get('chat-stuck')).toEqual({
+      chatId: 'chat-stuck',
+      streamId: null,
+      status: 'inactive',
+    })
+  })
+
+  it('repairs a verified stale persisted stream marker when requested', async () => {
+    await reconcileChatStreamMarkers([{ chatId: 'chat-stuck', streamId: 'stream-orphaned' }], {
+      repairVerifiedStaleMarkers: true,
+    })
+
+    expect(mockUpdate).toHaveBeenCalled()
+    expect(mockSet).toHaveBeenCalledWith({ conversationId: null })
+    expect(mockWhere).toHaveBeenCalledWith(
+      mockAnd(
+        mockEq('copilotChats.id', 'chat-stuck'),
+        mockEq('copilotChats.conversationId', 'stream-orphaned')
+      )
+    )
+  })
+
+  it('uses the canonical Redis owner when the persisted stream marker is stale', async () => {
+    mockGetChatStreamLockOwners.mockResolvedValueOnce({
+      status: 'verified',
+      ownersByChatId: new Map([['chat-mismatch', 'stream-live']]),
+    })
+
+    const markers = await reconcileChatStreamMarkers([
+      { chatId: 'chat-mismatch', streamId: 'stream-stale' },
+    ])
+
+    expect(markers.get('chat-mismatch')).toEqual({
+      chatId: 'chat-mismatch',
+      streamId: 'stream-live',
+      status: 'active',
+    })
+  })
+
+  it('preserves persisted stream markers when Redis state is unknown', async () => {
+    mockGetChatStreamLockOwners.mockResolvedValueOnce({
+      status: 'unknown',
+      ownersByChatId: new Map<string, string>(),
+    })
+
+    const markers = await reconcileChatStreamMarkers([
+      { chatId: 'chat-remote', streamId: 'stream-remote' },
+    ])
+
+    expect(markers.get('chat-remote')).toEqual({
+      chatId: 'chat-remote',
+      streamId: 'stream-remote',
+      status: 'unknown',
+    })
+  })
+
+  it('preserves a persisted marker when unknown local owner differs', async () => {
+    mockGetChatStreamLockOwners.mockResolvedValueOnce({
+      status: 'unknown',
+      ownersByChatId: new Map([['chat-mismatch', 'stream-local']]),
+    })
+
+    const markers = await reconcileChatStreamMarkers([
+      { chatId: 'chat-mismatch', streamId: 'stream-persisted' },
+    ])
+
+    expect(markers.get('chat-mismatch')).toEqual({
+      chatId: 'chat-mismatch',
+      streamId: 'stream-persisted',
+      status: 'unknown',
+    })
+  })
+
+  it('treats a null persisted marker as inactive even when Redis still holds a lock (post-completion teardown window)', async () => {
+    mockGetChatStreamLockOwners.mockResolvedValueOnce({
+      status: 'verified',
+      ownersByChatId: new Map([['chat-starting', 'stream-starting']]),
+    })
+
+    const markers = await reconcileChatStreamMarkers([{ chatId: 'chat-starting', streamId: null }])
+
+    expect(markers.get('chat-starting')).toEqual({
+      chatId: 'chat-starting',
+      streamId: null,
+      status: 'inactive',
+    })
+  })
+
+  it('does not query locks when no chats have persisted stream markers', async () => {
+    const markers = await reconcileChatStreamMarkers([{ chatId: 'chat-idle', streamId: null }])
+
+    expect(markers.get('chat-idle')).toEqual({
+      chatId: 'chat-idle',
+      streamId: null,
+      status: 'inactive',
+    })
+  })
+})
@@ -0,0 +1,135 @@
+import { db } from '@sim/db'
+import { copilotChats } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
+import { and, eq } from 'drizzle-orm'
+import { getChatStreamLockOwners } from '@/lib/copilot/request/session'
+
+const logger = createLogger('ChatStreamLiveness')
+
+export interface ChatStreamMarkerCandidate {
+  chatId: string
+  streamId: string | null
+}
+
+export interface ReconciledChatStreamMarker {
+  chatId: string
+  streamId: string | null
+  status: 'active' | 'inactive' | 'unknown'
+}
+
+interface ReconcileChatStreamMarkersOptions {
+  repairVerifiedStaleMarkers?: boolean
+}
+
+/**
+ * Reconciles persisted chat stream markers against the runtime stream lock.
+ *
+ * Redis lock ownership is the canonical live-stream signal. When the lookup is
+ * verified, missing owners clear stale persisted markers and present owners win
+ * over stale DB values. When Redis state is unknown, persisted markers are
+ * preserved so a transient Redis failure in a multi-pod deployment does not
+ * incorrectly hide a live stream owned by another pod.
+ */
+export async function reconcileChatStreamMarkers(
+  candidates: ChatStreamMarkerCandidate[],
+  options: ReconcileChatStreamMarkersOptions = {}
+): Promise<Map<string, ReconciledChatStreamMarker>> {
+  const results = new Map<string, ReconciledChatStreamMarker>()
+
+  for (const candidate of candidates) {
+    if (candidate.streamId === null) {
+      results.set(candidate.chatId, {
+        chatId: candidate.chatId,
+        streamId: null,
+        status: 'inactive',
+      })
+      continue
+    }
+    results.set(candidate.chatId, {
+      chatId: candidate.chatId,
+      streamId: candidate.streamId,
+      status: 'unknown',
+    })
+  }
+
+  const candidatesWithMarkers = candidates.filter((candidate) => candidate.streamId !== null)
+  if (candidatesWithMarkers.length === 0) {
+    return results
+  }
+
+  const { status, ownersByChatId } = await getChatStreamLockOwners(
+    candidatesWithMarkers.map((candidate) => candidate.chatId)
+  )
+
+  for (const candidate of candidatesWithMarkers) {
+    const owner = ownersByChatId.get(candidate.chatId)
+    if (owner && (status === 'verified' || owner === candidate.streamId)) {
+      results.set(candidate.chatId, {
+        chatId: candidate.chatId,
+        streamId: owner,
+        status: 'active',
+      })
+      continue
+    }
+
+    if (status === 'verified') {
+      results.set(candidate.chatId, {
+        chatId: candidate.chatId,
+        streamId: null,
+        status: 'inactive',
+      })
+      continue
+    }
+
+    results.set(candidate.chatId, {
+      chatId: candidate.chatId,
+      streamId: candidate.streamId,
+      status: 'unknown',
+    })
+  }
+
+  if (options.repairVerifiedStaleMarkers) {
+    await repairVerifiedStaleMarkers(candidates, results)
+  }
+
+  return results
+}
+
+async function repairVerifiedStaleMarkers(
+  candidates: ChatStreamMarkerCandidate[],
+  results: Map<string, ReconciledChatStreamMarker>
+): Promise<void> {
+  const staleCandidates = candidates.filter(
+    (candidate): candidate is { chatId: string; streamId: string } => {
+      const result = results.get(candidate.chatId)
+      return (
+        candidate.streamId !== null && result?.status === 'inactive' && result.streamId === null
+      )
+    }
+  )
+
+  if (staleCandidates.length === 0) return
+
+  await Promise.all(
+    staleCandidates.map(async (candidate) => {
+      try {
+        await db
+          .update(copilotChats)
+          .set({ conversationId: null })
+          .where(
+            and(
+              eq(copilotChats.id, candidate.chatId),
+              eq(copilotChats.conversationId, candidate.streamId)
+            )
+          )
+      } catch (error) {
+        logger.warn('Failed to repair stale chat stream marker', {
+          chatId: candidate.chatId,
+          streamId: candidate.streamId,
+          error: toError(error).message,
+        })
+      }
+    })
+  )
+}
@@ -22,7 +22,12 @@ vi.mock('@/lib/copilot/request/otel', () => ({
    fn({ setAttribute: vi.fn() }),
 }))

-import { startAbortPoller } from '@/lib/copilot/request/session/abort'
+import {
+  acquirePendingChatStream,
+  getChatStreamLockOwners,
+  releasePendingChatStream,
+  startAbortPoller,
+} from '@/lib/copilot/request/session/abort'

 describe('startAbortPoller heartbeat', () => {
  beforeEach(() => {
@@ -159,3 +164,93 @@ describe('startAbortPoller heartbeat', () => {
    }
  })
 })
+
+describe('getChatStreamLockOwners', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(null)
+  })
+
+  it('returns a verified empty owner map when no chat ids are provided', async () => {
+    const result = await getChatStreamLockOwners([])
+    expect(result.status).toBe('verified')
+    expect(result.ownersByChatId.size).toBe(0)
+  })
+
+  it('returns Redis lock owners keyed by chat id', async () => {
+    const mget = vi.fn().mockResolvedValue(['stream-1', null, 'stream-3'])
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ mget } as never)
+
+    const result = await getChatStreamLockOwners(['chat-1', 'chat-2', 'chat-3'])
+
+    expect(mget).toHaveBeenCalledWith([
+      'copilot:chat-stream-lock:chat-1',
+      'copilot:chat-stream-lock:chat-2',
+      'copilot:chat-stream-lock:chat-3',
+    ])
+    expect(result.status).toBe('verified')
+    expect(result.ownersByChatId).toEqual(
+      new Map([
+        ['chat-1', 'stream-1'],
+        ['chat-3', 'stream-3'],
+      ])
+    )
+  })
+
+  it('returns a verified empty map when every lock has expired in Redis', async () => {
+    const mget = vi.fn().mockResolvedValue([null, null])
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ mget } as never)
+
+    const result = await getChatStreamLockOwners(['chat-stuck-1', 'chat-stuck-2'])
+
+    expect(result.status).toBe('verified')
+    expect(result.ownersByChatId.size).toBe(0)
+  })
+
+  it('trusts verified Redis null over a process-local pending stream', async () => {
+    const mget = vi.fn().mockResolvedValue([null])
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ mget } as never)
+    await acquirePendingChatStream('chat-local', 'stream-local')
+
+    try {
+      const result = await getChatStreamLockOwners(['chat-local'])
+
+      expect(result.status).toBe('verified')
+      expect(result.ownersByChatId.size).toBe(0)
+    } finally {
+      await releasePendingChatStream('chat-local', 'stream-local')
+    }
+  })
+
+  it('returns unknown status when Redis is unavailable', async () => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(null)
+
+    const result = await getChatStreamLockOwners(['chat-1', 'chat-2'])
+
+    expect(result.status).toBe('unknown')
+    expect(result.ownersByChatId.size).toBe(0)
+  })
+
+  it('preserves local pending stream owners when Redis is unavailable', async () => {
+    await acquirePendingChatStream('chat-local', 'stream-local')
+
+    try {
+      const result = await getChatStreamLockOwners(['chat-local', 'chat-remote'])
+
+      expect(result.status).toBe('unknown')
+      expect(result.ownersByChatId).toEqual(new Map([['chat-local', 'stream-local']]))
+    } finally {
+      await releasePendingChatStream('chat-local', 'stream-local')
+    }
+  })
+
+  it('returns unknown status without throwing when mget rejects', async () => {
+    const mget = vi.fn().mockRejectedValue(new Error('redis down'))
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ mget } as never)
+
+    const result = await getChatStreamLockOwners(['chat-1', 'chat-2'])
+
+    expect(result.status).toBe('unknown')
+    expect(result.ownersByChatId.size).toBe(0)
+  })
+})
@@ -35,6 +35,11 @@ const CHAT_STREAM_LOCK_TTL_SECONDS = 60
 */
 const CHAT_STREAM_LOCK_HEARTBEAT_INTERVAL_MS = 20_000

+export interface ChatStreamLockOwnersResult {
+  status: 'verified' | 'unknown'
+  ownersByChatId: Map<string, string>
+}
+
 function registerPendingChatStream(chatId: string, streamId: string): void {
  let resolve!: () => void
  const promise = new Promise<void>((r) => {
@@ -123,6 +128,50 @@ export async function getPendingChatStreamId(chatId: string): Promise<string | n
  }
 }

+/**
+ * Loads canonical stream lock owners for chat IDs.
+ *
+ * `status: 'verified'` means Redis was queried successfully, so a missing
+ * owner is authoritative. `status: 'unknown'` means only the process-local
+ * pending map is known, which is not enough to declare remote streams inactive
+ * in a multi-pod deployment.
+ */
+export async function getChatStreamLockOwners(
+  chatIds: string[]
+): Promise<ChatStreamLockOwnersResult> {
+  const localOwnersByChatId = new Map<string, string>()
+  if (chatIds.length === 0) {
+    return { status: 'verified', ownersByChatId: localOwnersByChatId }
+  }
+
+  for (const chatId of chatIds) {
+    const entry = pendingChatStreams.get(chatId)
+    if (entry?.streamId) localOwnersByChatId.set(chatId, entry.streamId)
+  }
+
+  const redis = getRedisClient()
+  if (!redis) {
+    return { status: 'unknown', ownersByChatId: localOwnersByChatId }
+  }
+
+  try {
+    const keys = chatIds.map(getChatStreamLockKey)
+    const values = await redis.mget(keys)
+    const redisOwnersByChatId = new Map<string, string>()
+    for (let i = 0; i < chatIds.length; i++) {
+      const owner = values[i]
+      if (owner) redisOwnersByChatId.set(chatIds[i], owner)
+    }
+    return { status: 'verified', ownersByChatId: redisOwnersByChatId }
+  } catch (error) {
+    logger.warn('Failed to load chat stream lock owners (batch)', {
+      count: chatIds.length,
+      error: toError(error).message,
+    })
+    return { status: 'unknown', ownersByChatId: localOwnersByChatId }
+  }
+}
+
 export async function releasePendingChatStream(chatId: string, streamId: string): Promise<void> {
  try {
    await releaseLock(getChatStreamLockKey(chatId), streamId)
@@ -1,9 +1,11 @@
+export type { ChatStreamLockOwnersResult } from './abort'
 export {
  AbortReason,
  type AbortReasonValue,
  abortActiveStream,
  acquirePendingChatStream,
  cleanupAbortMarker,
+  getChatStreamLockOwners,
  getPendingChatStreamId,
  isExplicitStopReason,
  registerActiveStream,