fix(mothership): serialize headless run admission with chat soft delete

apps/sim/lib/knowledge/application/slack-search/assistant.integration.ts

@@ -139,8 +139,11 @@ function job() {
   }
 }
 
-/** Runs one Slack turn in a fresh private chat and returns the run it recorded. */
-async function slackTurn() {
+/**
+ * Runs one Slack turn in a fresh private chat and returns the run it recorded. `deletedAt`
+ * stands for a delete committed after the turn resolved its chat.
+ */
+async function slackTurn(deletedAt?: Date) {
   const chatId = generateId()
   const turnId = generateId()
   hoisted.chat = { id: chatId, model: 'default' }
@@ -150,6 +153,7 @@ async function slackTurn() {
     userId: hoisted.userId,
     organizationId: hoisted.organizationId,
     type: 'mothership',
+    deletedAt,
   })
   const controller = new AbortController()
   hoisted.controller = controller
@@ -237,6 +241,17 @@ describe('Slack Assistant run record', () => {
     expect(run.completedAt).not.toBeNull()
   })
 
+  it('never starts the worker for a turn whose chat was deleted before its run', async () => {
+    mockRunHeadlessCopilotLifecycle.mockClear()
+
+    const { run, outcome } = await slackTurn(new Date())
+
+    expect(mockRunHeadlessCopilotLifecycle).not.toHaveBeenCalled()
+    expect(outcome).toBeInstanceOf(Error)
+    expect(run.status).toBe('cancelled')
+    expect(run.toolAdmissionClosedAt).not.toBeNull()
+  })
+
   it('records a failed turn as an error', async () => {
     mockRunHeadlessCopilotLifecycle.mockResolvedValueOnce({
       success: false,

apps/sim/lib/knowledge/application/slack-search/assistant.ts

@@ -49,7 +49,7 @@ import {
   startAbortPoller,
   unregisterActiveStream,
 } from '@/lib/mothership/request/session/abort'
-import { isExplicitStopReason } from '@/lib/mothership/request/session/abort-reason'
+import { AbortReason, isExplicitStopReason } from '@/lib/mothership/request/session/abort-reason'
 import type { OrchestratorResult } from '@/lib/mothership/request/types'
 import { organizationRoutes } from '@/lib/navigation/paths'
 import { SlackSearchAssistantStream } from '@/lib/slack-search/assistant-stream'
@@ -206,6 +206,11 @@ export async function runSlackSearchAssistant(
     })
     if (!run) throw new Error('Could not persist Assistant execution')
     runId = run.id
+    /** Admission found the turn stopped or its chat deleted: it ends as that Stop would. */
+    if (run.status === 'cancelled') {
+      controller.abort(AbortReason.UserStop)
+      controller.signal.throwIfAborted()
+    }
     const responseStream = new SlackSearchAssistantStream({
       token: secret.botToken,
       channel: job.message.channelId,

apps/sim/lib/mothership/async-runs/repository.test.ts

@@ -148,9 +148,10 @@ describe('run admission and early Stop', () => {
       [copilotRequestStops.workspaceId, scope.workspaceId],
       [copilotRequestStops.streamId, scope.streamId],
     ]) {
+      /** The chat row is read first; the Stop lookup follows it. */
       expect(
         hasMockCondition(
-          dbChainMockFns.where.mock.calls[0]?.[0],
+          dbChainMockFns.where.mock.calls[1]?.[0],
           (condition) =>
             condition.type === 'eq' && condition.left === column && condition.right === value
         )

apps/sim/lib/mothership/async-runs/repository.ts

@@ -441,28 +441,31 @@ export async function createRunSegment(input: CreateRunSegmentInput) {
   )
 }
 
-/** Caller holds the Stop/admission lock; its related chat writes commit with this run. */
+/**
+ * Caller holds the Stop/admission lock; its related chat writes commit with this run. The
+ * chat row is read FOR SHARE, after that lock as chat admission takes it, so a soft delete
+ * either commits first and this run starts stopped, or waits and then stops this run.
+ */
 export async function insertRunSegment(tx: RunAdmissionTransaction, input: CreateRunSegmentInput) {
-  let workspaceId = input.workspaceId
-  let organizationId = input.organizationId
-  if (!workspaceId && !organizationId) {
-    const [chat] = await tx
-      .select({
-        workspaceId: copilotChats.workspaceId,
-        organizationId: copilotChats.organizationId,
-      })
-      .from(copilotChats)
-      .where(and(eq(copilotChats.id, input.chatId), eq(copilotChats.userId, input.userId)))
-      .limit(1)
-    workspaceId = chat?.workspaceId
-    organizationId = chat?.organizationId
-  }
+  const [chat] = await tx
+    .select({
+      workspaceId: copilotChats.workspaceId,
+      organizationId: copilotChats.organizationId,
+      deletedAt: copilotChats.deletedAt,
+    })
+    .from(copilotChats)
+    .where(and(eq(copilotChats.id, input.chatId), eq(copilotChats.userId, input.userId)))
+    .for('share')
+  const inheritsOwner = !input.workspaceId && !input.organizationId
+  const workspaceId = inheritsOwner ? chat?.workspaceId : input.workspaceId
+  const organizationId = inheritsOwner ? chat?.organizationId : input.organizationId
   const { table, predicate } = stopLocation({ ...input, workspaceId, organizationId })
   const [stop] = await tx
     .select({ stoppedAt: table.stoppedAt })
     .from(table)
     .where(predicate)
     .limit(1)
+  const closedAt = stop?.stoppedAt ?? chat?.deletedAt
   const [run] = await tx
     .insert(copilotRuns)
     .values({
@@ -481,8 +484,8 @@ export async function insertRunSegment(tx: RunAdmissionTransaction, input: Creat
       provider: input.provider ?? null,
       requestContext: input.requestContext ?? {},
       desktopDeviceId: input.desktopDeviceId ?? null,
-      status: stop ? 'cancelled' : (input.status ?? 'active'),
-      ...(stop ? { completedAt: sql`now()`, toolAdmissionClosedAt: stop.stoppedAt } : {}),
+      status: closedAt ? 'cancelled' : (input.status ?? 'active'),
+      ...(closedAt ? { completedAt: sql`now()`, toolAdmissionClosedAt: closedAt } : {}),
     })
     .returning()
   if (!run) throw new Error('Run admission did not return its persisted identity')

apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts

@@ -66,6 +66,7 @@ import { NextRequest } from 'next/server'
 import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { SIM_TOOL_EXECUTION_VERSION } from '@/lib/mothership/async-runs/lifecycle'
+import { createRunSegment } from '@/lib/mothership/async-runs/repository'
 import { admitChatTurn } from '@/lib/mothership/chat/application/admit-turn'
 import { deleteCopilotChat, softDeleteMothershipChat } from '@/lib/mothership/chat/deletion'
 import { chatPubSub } from '@/lib/mothership/chat-status'
@@ -203,6 +204,37 @@ async function seedRun(
   return { userId: owner.userId, chatId, streamId, runId, ...scope }
 }
 
+/** A chat with no run yet, as a task wake, inbox email, API or Slack turn finds it. */
+async function seedChat(userId: string, workspaceId: string) {
+  const chatId = generateId()
+  await db.insert(copilotChats).values({ id: chatId, userId, workspaceId, type: 'mothership' })
+  return chatId
+}
+
+/** Admits a headless turn's run the way those surfaces do, without the chat view's admission. */
+async function admitHeadlessRun(userId: string, chatId: string, workspaceId: string) {
+  const streamId = generateId()
+  testStreams.add(streamId)
+  const run = await createRunSegment({
+    id: generateId(),
+    executionId: generateId(),
+    chatId,
+    userId,
+    workspaceId,
+    streamId,
+    requestContext: { source: 'headless_lifecycle' },
+  })
+  return { run, seeded: { userId, chatId, streamId, runId: run.id, workspaceId } }
+}
+
+async function toolAdmissionClosedAt(run: SeededRun) {
+  const [row] = await db
+    .select({ closedAt: copilotRuns.toolAdmissionClosedAt })
+    .from(copilotRuns)
+    .where(eq(copilotRuns.id, run.runId))
+  return row?.closedAt ?? null
+}
+
 /** What the worker's heartbeat reads, through the outbound control transport. */
 async function runControl(run: SeededRun) {
   const result = await executeSimControl({
@@ -294,6 +326,23 @@ async function whileHolding<T>(chatId: string, run: (holder: number) => Promise<
   }
 }
 
+/** Queues admission and deletion behind a held chat row in the order `first` names. */
+async function queueOnChat<A, D>(
+  chatId: string,
+  first: 'admission' | 'deletion',
+  admit: () => Promise<A>,
+  remove: () => Promise<D>
+): Promise<{ admission: Promise<A>; deletion: Promise<D> }> {
+  return whileHolding(chatId, async (holder) => {
+    const admission = first === 'admission' ? admit() : undefined
+    const deletion = admission ? undefined : remove()
+    const waiter = await lockWaiterBehind(holder)
+    const queued = { admission: admission ?? admit(), deletion: deletion ?? remove() }
+    await lockWaiterBehind(holder, waiter)
+    return queued
+  })
+}
+
 const STOPPED = { status: 200, body: { stopped: true } }
 const LIVE = { status: 200, body: { stopped: false } }
 
@@ -674,12 +723,9 @@ describe.runIf(Boolean(redisUrl))('removing a run’s chat, owner or access stop
     async (first) => {
       const ownerId = await seedUser()
       const workspaceId = await seedWorkspace(ownerId)
-      const chatId = generateId()
+      const chatId = await seedChat(ownerId, workspaceId)
       const streamId = generateId()
       testStreams.add(streamId)
-      await db
-        .insert(copilotChats)
-        .values({ id: chatId, userId: ownerId, workspaceId, type: 'mothership' })
       const sendClaim = { normalizedKey: `run-removal:${generateId()}`, claimToken: generateId() }
       await db
         .insert(idempotencyKey)
@@ -725,20 +771,7 @@ describe.runIf(Boolean(redisUrl))('removing a run’s chat, owner or access stop
           .finally(() => releasePendingChatStream(chatId, streamId, lease))
       const remove = () => softDeleteMothershipChat({ chatId, userId: ownerId })
 
-      const { admission, deletion } = await whileHolding(chatId, async (holder) => {
-        if (first === 'admission') {
-          const admission = admit()
-          const waiter = await lockWaiterBehind(holder)
-          const deletion = remove()
-          await lockWaiterBehind(holder, waiter)
-          return { admission, deletion }
-        }
-        const deletion = remove()
-        const waiter = await lockWaiterBehind(holder)
-        const admission = admit()
-        await lockWaiterBehind(holder, waiter)
-        return { admission, deletion }
-      })
+      const { admission, deletion } = await queueOnChat(chatId, first, admit, remove)
 
       const admitted = await admission
       expect(await deletion).not.toBeNull()
@@ -751,4 +784,61 @@ describe.runIf(Boolean(redisUrl))('removing a run’s chat, owner or access stop
       expect(await runControl(run)).toEqual(STOPPED)
     }
   )
+
+  it('admits a headless run of a live chat as live', async () => {
+    const ownerId = await seedUser()
+    const workspaceId = await seedWorkspace(ownerId)
+
+    const { run, seeded } = await admitHeadlessRun(
+      ownerId,
+      await seedChat(ownerId, workspaceId),
+      workspaceId
+    )
+
+    expect(run).toMatchObject({ status: 'active', toolAdmissionClosedAt: null, completedAt: null })
+    expect(await runControl(seeded)).toEqual(LIVE)
+  })
+
+  it('admits a headless run of a chat deleted before it as already stopped', async () => {
+    const ownerId = await seedUser()
+    const workspaceId = await seedWorkspace(ownerId)
+    const chatId = await seedChat(ownerId, workspaceId)
+    expect(await softDeleteMothershipChat({ chatId, userId: ownerId })).not.toBeNull()
+
+    const { run, seeded } = await admitHeadlessRun(ownerId, chatId, workspaceId)
+
+    expect(run.status).toBe('cancelled')
+    expect(run.toolAdmissionClosedAt).not.toBeNull()
+    expect(run.completedAt).not.toBeNull()
+    expect(await runControl(seeded)).toEqual(STOPPED)
+  })
+
+  /**
+   * A headless run is admitted without writing the chat row, so only its read of the chat
+   * serializes it with a deletion: holding the row queues both in a chosen order.
+   */
+  it.each(['admission', 'deletion'] as const)(
+    'never leaves a headless run live while its chat is being deleted, when %s reaches the chat first',
+    async (first) => {
+      const ownerId = await seedUser()
+      const workspaceId = await seedWorkspace(ownerId)
+      const chatId = await seedChat(ownerId, workspaceId)
+      const admit = () => admitHeadlessRun(ownerId, chatId, workspaceId)
+      const remove = () => softDeleteMothershipChat({ chatId, userId: ownerId })
+
+      const { admission, deletion } = await queueOnChat(chatId, first, admit, remove)
+
+      const { run, seeded } = await admission
+      expect(await deletion).not.toBeNull()
+      if (first === 'admission') {
+        await expect.poll(pushedStops).toEqual([seeded.streamId])
+        expect(await hasStop(seeded)).toBe(true)
+      } else {
+        expect(run.status).toBe('cancelled')
+        expect(pushedStops()).toEqual([])
+      }
+      expect(await toolAdmissionClosedAt(seeded)).not.toBeNull()
+      expect(await runControl(seeded)).toEqual(STOPPED)
+    }
+  )
 })