fix(mothership): order fork publication against a purge of its source

A fork can share keys with its source (organization attachments, files whose
copy failed), and chat cleanup deletes a shared key once no remaining chat
references it, checking after it deletes the source row. The fork's publish
transaction now holds the source row with FOR KEY SHARE: a purge that already
removed it refuses the fork (404, worker copy discarded), and one that has not
waits for the commit and then sees the fork's references.

Author: waleedlatif1

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

@@ -185,6 +185,7 @@ describe('POST /api/mothership/chats/[chatId]/fork', () => {
     queueTableRows(copilotChats, [chat])
     queueTableRows(copilotChats, [chat])
     queueTableRows(member, [{ role: 'member' }])
+    queueTableRows(copilotChats, [{ id: chat.id }])
     const attachments = [
       {
         id: 'upload-1',
@@ -215,6 +216,26 @@ describe('POST /api/mothership/chats/[chatId]/fork', () => {
     expect(mockAssertActiveWorkspaceAccess).not.toHaveBeenCalled()
   })
 
+  it('refuses to publish a fork whose source chat was purged while it copied', async () => {
+    // The fork shares its attachment keys with the source, and cleanup deletes an unreferenced
+    // key only after deleting the source row: publishing without the source would leave the
+    // fork pointing at bytes that cleanup is about to delete.
+    dbChainMockFns.limit.mockReset()
+    const chat = { ...parentRow, workspaceId: null, organizationId: 'org-1', resources: [] }
+    queueTableRows(copilotChats, [chat])
+    queueTableRows(copilotChats, [chat])
+    queueTableRows(member, [{ role: 'member' }])
+    queueTableRows(copilotChats, [])
+    const res = await POST(createRequest('chat-1'), createRouteContext({ chatId: 'chat-1' }))
+    expect(res.status).toBe(404)
+    expect(mockAppendCopilotChatMessages).not.toHaveBeenCalled()
+    expect(mockPublishStatusChanged).not.toHaveBeenCalled()
+    expect(mockFetchGo.mock.calls.map(([url]) => url)).toEqual([
+      'http://mothership.test/api/chats/fork',
+      'http://mothership.test/api/tasks/cleanup',
+    ])
+  })
+
   it.each(['membership', 'capability'] as const)(
     'denies organization forks after %s revocation before copying',
     async (revocation) => {

apps/sim/lib/mothership/chat/application/fork.ts

@@ -156,6 +156,20 @@ export const forkChat = defineAuthorizedChatUseCase({
 
       /** Publish only after both the file bytes and the worker conversation are prepared. */
       await db.transaction(async (tx) => {
+        /**
+         * The fork can share keys with its source (organization attachments, files whose copy
+         * failed), and chat cleanup deletes a shared key once no remaining chat references it,
+         * checking only after it deletes the source row. Holding the source row until commit
+         * orders the two: a purge that already removed it refuses this fork, and one that has
+         * not waits for this commit and then sees the fork's references.
+         */
+        const [source] = await tx
+          .select({ id: copilotChats.id })
+          .from(copilotChats)
+          .where(eq(copilotChats.id, chatId))
+          .for('key share')
+          .limit(1)
+        if (!source) throw new OrchestrationError('not_found', 'Chat not found')
         const [row] = await tx
           .insert(copilotChats)
           .values({