From 1e8e4e78683b5b89b85cc585c95bd47394acb0e1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 20:07:39 -0700 Subject: [PATCH] fix(mothership): serialize headless run admission with chat soft delete --- .../slack-search/assistant.integration.ts | 19 ++- .../application/slack-search/assistant.ts | 7 +- .../mothership/async-runs/repository.test.ts | 3 +- .../lib/mothership/async-runs/repository.ts | 37 ++--- .../removal-stops-runs.integration.ts | 126 +++++++++++++++--- 5 files changed, 153 insertions(+), 39 deletions(-) diff --git a/apps/sim/lib/knowledge/application/slack-search/assistant.integration.ts b/apps/sim/lib/knowledge/application/slack-search/assistant.integration.ts index 8b0374cddcd..1a52458d9de 100644 --- a/apps/sim/lib/knowledge/application/slack-search/assistant.integration.ts +++ b/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, diff --git a/apps/sim/lib/knowledge/application/slack-search/assistant.ts b/apps/sim/lib/knowledge/application/slack-search/assistant.ts index b211965fbdd..dad4f2e0c7f 100644 --- a/apps/sim/lib/knowledge/application/slack-search/assistant.ts +++ b/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, diff --git a/apps/sim/lib/mothership/async-runs/repository.test.ts b/apps/sim/lib/mothership/async-runs/repository.test.ts index 7ae9662d040..dbbae17f179 100644 --- a/apps/sim/lib/mothership/async-runs/repository.test.ts +++ b/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 ) diff --git a/apps/sim/lib/mothership/async-runs/repository.ts b/apps/sim/lib/mothership/async-runs/repository.ts index 7f826b28f96..543606837a3 100644 --- a/apps/sim/lib/mothership/async-runs/repository.ts +++ b/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') diff --git a/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts b/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts index 484d6fdda3f..6d49149df9c 100644 --- a/apps/sim/lib/mothership/request/application/removal-stops-runs.integration.ts +++ b/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(chatId: string, run: (holder: number) => Promise< } } +/** Queues admission and deletion behind a held chat row in the order `first` names. */ +async function queueOnChat( + chatId: string, + first: 'admission' | 'deletion', + admit: () => Promise, + remove: () => Promise +): Promise<{ admission: Promise; deletion: Promise }> { + 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) + } + ) })