fix(mothership): discard the worker copy of a fork that is not published

When anything after the worker copy failed, most often the final transaction
that publishes the chat, the worker kept a conversation Sim had no chat for.
The fork now asks the worker to clean that chat up, through the same cleanup
endpoint chat deletion uses. It is best effort and a no-op when the worker
never committed.

Author: waleedlatif1

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

@@ -342,6 +342,19 @@ describe('POST /api/mothership/chats/[chatId]/fork', () => {
     expect(dbChainMockFns.delete).not.toHaveBeenCalled()
   })
 
+  it('discards the worker copy when the fork cannot be published', async () => {
+    dbChainMockFns.transaction.mockRejectedValueOnce(new Error('connection lost'))
+    const res = await POST(createRequest('chat-1'), createRouteContext({ chatId: 'chat-1' }))
+    expect(res.status).toBe(500)
+    const [fork, discard] = mockFetchGo.mock.calls
+    expect(fork[0]).toBe('http://mothership.test/api/chats/fork')
+    expect(discard[0]).toBe('http://mothership.test/api/tasks/cleanup')
+    expect(JSON.parse(discard[1].body)).toEqual({
+      chatIds: [JSON.parse(fork[1].body).newChatId],
+    })
+    expect(mockPublishStatusChanged).not.toHaveBeenCalled()
+  })
+
   it('copies pre-cut uploads and drops only post-cut ghosts', async () => {
     // The source chat owns two more uploads (apple pre-cut, banana post-cut)
     // beside the kept one, plus one shared workspace-file resource. The fork

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

@@ -19,7 +19,10 @@ import {
   planChatFileCopies,
 } from '@/lib/mothership/chat/fork-chat-files'
 import { planForkInlineImages } from '@/lib/mothership/chat/fork-inline-images'
-import { copyWorkerConversation } from '@/lib/mothership/chat/fork-worker'
+import {
+  copyWorkerConversation,
+  discardWorkerConversation,
+} from '@/lib/mothership/chat/fork-worker'
 import { loadCopilotChatMessages } from '@/lib/mothership/chat/lifecycle'
 import { appendCopilotChatMessages } from '@/lib/mothership/chat/messages-store'
 import {
@@ -88,6 +91,8 @@ export const forkChat = defineAuthorizedChatUseCase({
     const { chatId, upToMessageId } = input
     let preparedBlobs: ChatBlobCopyTask[] = []
     let published = false
+    const newId = generateId()
+    let workerCopyRequested = false
     try {
       const messages = await loadCopilotChatMessages(chatId)
       const forkIdx = messages.findIndex((m) => m.id === upToMessageId)
@@ -111,7 +116,6 @@ export const forkChat = defineAuthorizedChatUseCase({
       /** The source chat's chat-owned file ids (no cut) — the "is this resource a ghost?" test set for the rewrite. */
       const chatOwnedFileIds = new Set(chatOwnedFiles.map((row) => row.id))
 
-      const newId = generateId()
       /** Strip a leading "Fork | " so titles don't stack prefixes when forking a forked chat. */
       const baseTitle = (parent.title ?? 'New chat').replace(/^Fork \| /, '')
       const title = `Fork | ${baseTitle}`
@@ -129,6 +133,7 @@ export const forkChat = defineAuthorizedChatUseCase({
       ).filter((resource) => resource.type !== 'file' || !failedIds.has(resource.id))
       const cutUser = [...forkedMessages].reverse().find((message) => message.role === 'user')
       if (!cutUser) throw new Error('The fork has no user message')
+      workerCopyRequested = true
       await copyWorkerConversation({
         sourceChatId: chatId,
         newChatId: newId,
@@ -198,6 +203,8 @@ export const forkChat = defineAuthorizedChatUseCase({
         ...(failed > 0 ? { failedFileCopies: failed } : {}),
       }
     } catch (error) {
+      if (!published && workerCopyRequested)
+        await discardWorkerConversation({ newChatId: newId, userId })
       if (!published) {
         await mapWithConcurrency(preparedBlobs, 4, async (task) => {
           try {

apps/sim/lib/mothership/chat/fork-worker.ts

@@ -1,10 +1,14 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { isRetryableNetworkError } from '@/lib/core/errors/retryable-infrastructure'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { type ForkChatRequest, ForkChatResponse } from '@/lib/mothership/generated/protocol'
 import { fetchGo } from '@/lib/mothership/request/go/fetch'
 import { mothershipRequestHeaders } from '@/lib/mothership/request/headers'
 import { getMothershipBaseURL } from '@/lib/mothership/server/agent-url'
 
+const logger = createLogger('ForkWorker')
+
 /**
  * How long one attempt waits for the copy. The worker refuses a cut above 50,000 events
  * with 413, and measured locally it copies 15,000 events in about 2 s and 50,000 in about
@@ -16,6 +20,9 @@ import { getMothershipBaseURL } from '@/lib/mothership/server/agent-url'
  */
 const ATTEMPT_TIMEOUT_MS = 120_000
 
+/** Cleanup is a small archive-and-purge of one chat, never a copy. */
+const DISCARD_TIMEOUT_MS = 10_000
+
 /** Gateway failures: the request may never have reached a worker, so one more attempt is safe. */
 const RETRYABLE_STATUSES = new Set([502, 503, 504])
 
@@ -76,3 +83,41 @@ export async function copyWorkerConversation(request: ForkChatRequest): Promise<
     return
   }
 }
+
+/**
+ * Removes the worker's copy of a fork that will not be published, so the worker keeps no
+ * conversation Sim has no chat for. Best effort: a failure is logged, never thrown, since
+ * the caller is already reporting the fork's own failure.
+ *
+ * The worker rolls back a copy whose caller disconnected, so after a timed-out attempt
+ * this is normally a no-op. A worker that commits in the instant between its last
+ * disconnect check and its commit, or one deployed before that rollback existed, can
+ * commit after this call runs; that conversation then stays on the worker, unreachable
+ * from Sim.
+ */
+export async function discardWorkerConversation(
+  request: Pick<ForkChatRequest, 'newChatId' | 'userId'>
+): Promise<void> {
+  try {
+    const baseURL = await getMothershipBaseURL({ userId: request.userId })
+    const response = await fetchGo(`${baseURL}/api/tasks/cleanup`, {
+      method: 'POST',
+      headers: mothershipRequestHeaders(),
+      body: JSON.stringify({ chatIds: [request.newChatId] }),
+      signal: AbortSignal.timeout(DISCARD_TIMEOUT_MS),
+      spanName: 'sim → worker /api/tasks/cleanup',
+      operation: 'discard_fork',
+    })
+    await response.body?.cancel().catch(() => undefined)
+    if (!response.ok)
+      logger.warn('Worker refused to discard an unpublished fork', {
+        chatId: request.newChatId,
+        status: response.status,
+      })
+  } catch (error) {
+    logger.warn('Failed to discard an unpublished fork on the worker', {
+      chatId: request.newChatId,
+      error: getErrorMessage(error),
+    })
+  }
+}