fix(mothership): stop live runs when their chat, owner or access is removed (#9089)

* fix(mothership): stop live runs when their chat, owner or access is removed

Deleting a chat, an account or a workspace, or removing someone's access,
left the worker's run going: no path issued a Stop, and run control answered
the missing chat or owner with a 404/403 the worker can only retry.

- Run control answers Stop when the run's chat is deleted, its owner or
  workspace is gone, or the owner lost access; only a failure to read is an error
- Chat delete, copilot chat delete, account deletion, workspace archive,
  member removal, organization removal, transfer and external-collaborator
  removal record Stop in their own transaction and push it after commit
- Stop delivery is shared with the user's Stop
- A takeover never claims a run whose chat was deleted

* fix(mothership): deliver removal stops from callers outside page graphs

Membership and workspace lifecycle modules sit in page server graphs, so they return the runs they stopped and their callers deliver the Stop. Fixes the organization run-control scope narrowing and shrinks the baselines this change paid down.

* fix(mothership): keep stop delivery complete when its rows or marker are gone

- Read a stopped run's client workflows and sandbox commands inside the removal's transaction, before a hard delete cascades them away
- Abort this process's controller even when the abort marker cannot be written, and keep delivering the rest of the Stop
- Account deletion no longer depends on the run-routing lookup
- The operator archive repair keeps the delivery out of its script graph
- Race test orders admission and deletion with a row-lock barrier; delivery assertions only count the test's own runs

* fix(mothership): signal a stopped run before reading its work, and deliver each removal stop once

- The user's Stop signals the worker and controller while its workflow and sandbox ownership is still being read
- Workspace deletion starts delivering its Stop as soon as the archive commits, before the slower post-archive cleanup
- External-collaborator removal delivers the Stop it recorded even when no access was left to remove
- Leaving a workspace that also ends the organization membership delivers each run's Stop once

* fix(mothership): never answer Stop for a refused delegation or an admin who keeps access

- Run control rethrows a refusal of the caller's own delegation (expired mid-request or out of scope) instead of answering Stop; only the chat, owner or access standing answers Stop
- Withdrawing a workspace grant stops the user's runs only when no effective access remains, so an organization admin keeps their run

* fix(mothership): stop an organization admin's runs when a demotion ends their access

Withdrawing a workspace grant and demoting an organization admin both end access only where no effective access remains, so the check runs after either: a role change that drops admin stops the member's runs in the organization's workspaces they can no longer reach, whatever order a directory sync withdraws the grant and the role in.

* fix(mothership): keep stopped-run identities out of removal results, check access only where runs live

- Organization member removal delivers its stopped runs and strips them from the result it returns to the route and audit
- Withdrawing a grant or demoting an admin checks remaining access only in workspaces where the user has a live run, instead of every organization workspace under the mutation lock

Author: waleedlatif1

apps/sim/app/api/copilot/chat/abort/route.test.ts

@@ -135,6 +135,30 @@ describe('POST /api/copilot/chat/abort', () => {
     expect(mockAbortWorkflow).toHaveBeenCalledExactlyOnceWith('client-execution')
   })
 
+  it('still cancels the run’s workflows when the abort marker cannot be written', async () => {
+    mockUnsettledWorkflows.mockResolvedValue(['client-execution'])
+    mockAbortActiveStream.mockRejectedValueOnce(new Error('redis unavailable'))
+    const response = await POST(abortRequest())
+    expect(response.status).toBe(200)
+    expect(mockRequestExplicitStreamAbort).toHaveBeenCalled()
+    expect(mockCancelWorkflow).toHaveBeenCalledExactlyOnceWith('client-execution')
+  })
+
+  it('signals the worker before the run’s workflow ownership has been read', async () => {
+    let releaseRead: (ids: string[]) => void = () => {}
+    mockUnsettledWorkflows.mockReturnValue(
+      new Promise<string[]>((resolve) => {
+        releaseRead = resolve
+      })
+    )
+    const response = POST(abortRequest())
+    await vi.waitFor(() => expect(mockRequestExplicitStreamAbort).toHaveBeenCalled())
+    expect(mockAbortActiveStream).toHaveBeenCalled()
+    releaseRead(['client-execution'])
+    expect((await response).status).toBe(200)
+    expect(mockCancelWorkflow).toHaveBeenCalledExactlyOnceWith('client-execution')
+  })
+
   it('does not signal workflows when the atomic Stop and admission closure cannot commit', async () => {
     mockRequestRunStop.mockRejectedValueOnce(new Error('database unavailable'))
     const response = await POST(abortRequest())

apps/sim/app/api/copilot/chat/delete/route.ts

@@ -1,12 +1,10 @@
-import { db } from '@sim/db'
-import { copilotChats } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { and, eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { deleteCopilotChatContract } from '@/lib/api/contracts/copilot'
 import { parseRequest } from '@/lib/api/server'
 import { getSession } from '@/lib/auth'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { deleteCopilotChat } from '@/lib/mothership/chat/deletion'
 import { getAccessibleCopilotChatAuth } from '@/lib/mothership/chat/lifecycle'
 import { chatPubSub } from '@/lib/mothership/chat-status'
 
@@ -45,10 +43,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest) => {
       )
     }
 
-    const [deleted] = await db
-      .delete(copilotChats)
-      .where(and(eq(copilotChats.id, parsed.chatId), eq(copilotChats.userId, session.user.id)))
-      .returning({ workspaceId: copilotChats.workspaceId })
+    const deleted = await deleteCopilotChat({ chatId: parsed.chatId, userId: session.user.id })
 
     if (!deleted) {
       return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })

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

@@ -11,6 +11,7 @@ import {
 } from '@/lib/api/contracts/mothership-chats'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { softDeleteMothershipChat } from '@/lib/mothership/chat/deletion'
 import { buildEffectiveChatTranscript } from '@/lib/mothership/chat/effective-transcript'
 import {
   getAccessibleCopilotChatAuth,
@@ -255,21 +256,7 @@ export const DELETE = withRouteHandler(
         return NextResponse.json({ success: true })
       }
 
-      const [deletedChat] = await db
-        .update(copilotChats)
-        .set({ deletedAt: new Date() })
-        .where(
-          and(
-            eq(copilotChats.id, chatId),
-            eq(copilotChats.userId, userId),
-            eq(copilotChats.type, 'mothership'),
-            isNull(copilotChats.deletedAt)
-          )
-        )
-        .returning({
-          workspaceId: copilotChats.workspaceId,
-          organizationId: copilotChats.organizationId,
-        })
+      const deletedChat = await softDeleteMothershipChat({ chatId, userId })
 
       if (!deletedChat) {
         return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })

apps/sim/app/api/mothership/runs/control/route.test.ts

@@ -15,6 +15,7 @@ import {
   workspaceContextMockFns,
 } from '@sim/testing/mocks/workspace-context.mock'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { OrchestrationError } from '@/lib/core/orchestration/types'
 
 vi.unmock('@/lib/mothership/request/http')
 vi.mock('@/lib/auth/ban', () => authBanMock)
@@ -73,17 +74,44 @@ describe('worker run control boundary', () => {
     expect(mocks.run).not.toHaveBeenCalled()
     expect(mocks.stopped).not.toHaveBeenCalled()
   })
-  it('conceals a chat belonging to another actor', async () => {
-    expect((await POST(request({ 'x-mothership-user-id': 'other-actor' }))).status).toBe(404)
+  it("answers Stop for a chat that is not the actor's, without reading its state", async () => {
+    const response = await POST(request({ 'x-mothership-user-id': 'other-actor' }))
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ stopped: true })
     expect(mocks.stopped).not.toHaveBeenCalled()
   })
+  it('answers Stop once the chat is deleted', async () => {
+    resetDbChainMock()
+    queueTableRows(schemaMock.copilotChats, [])
+    const response = await POST(request())
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ stopped: true })
+    expect(mocks.stopped).not.toHaveBeenCalled()
+  })
+  it('answers Stop once the actor lost workspace access', async () => {
+    mocks.authorize.mockRejectedValue(
+      new OrchestrationError('forbidden', 'Insufficient workspace permissions')
+    )
+    const response = await POST(request())
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ stopped: true })
+    expect(mocks.stopped).not.toHaveBeenCalled()
+  })
+  it('lets a live run continue', async () => {
+    mocks.stopped.mockResolvedValue(false)
+    const response = await POST(request())
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ stopped: false })
+  })
   it.each([
     null,
     { chatId: 'other-chat', workspaceId: 'workspace' },
     { chatId, workspaceId: 'other-workspace' },
-  ])('refuses a missing or mismatched canonical run: %j', async (run) => {
+  ])('answers Stop for a missing or mismatched canonical run: %j', async (run) => {
     mocks.run.mockResolvedValue(run)
-    expect((await POST(request())).status).toBe(404)
+    const response = await POST(request())
+    expect(response.status).toBe(200)
+    expect(await response.json()).toEqual({ stopped: true })
     expect(mocks.stopped).not.toHaveBeenCalled()
   })
   it('does not turn a database outage into permission to continue', async () => {

apps/sim/app/api/organizations/[id]/transfer-ownership/route.ts

@@ -13,6 +13,7 @@ import {
   transferOrganizationOwnership,
 } from '@/lib/billing/organizations/membership'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 
 const logger = createLogger('TransferOwnershipAPI')
 
@@ -150,6 +151,7 @@ export const POST = withRouteHandler(
         memberId: selfMember.id,
         spareSessionToken: session.session.token,
       })
+      deliverRunStops(removeResult.stoppedRuns ?? [])
 
       if (!removeResult.success) {
         logger.error('Transfer succeeded but self-removal failed', {

apps/sim/app/api/v1/admin/organizations/[id]/members/[memberId]/route.ts

@@ -43,6 +43,7 @@ import {
 } from '@/lib/billing/organizations/membership'
 import { isBillingEnabled } from '@/lib/core/config/env-flags'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 import {
   adminBadRequestResponse,
@@ -271,6 +272,7 @@ export const DELETE = withRouteHandler(
         memberId,
         skipBillingLogic,
       })
+      deliverRunStops(result.stoppedRuns ?? [])
 
       if (!result.success) {
         if (result.error === 'Cannot remove organization owner') {

apps/sim/app/api/v1/admin/workspaces/[id]/members/[memberId]/route.ts

@@ -33,6 +33,7 @@ import {
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import { ProjectConflictError } from '@/lib/projects/errors'
 import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
 import { getWorkspaceById } from '@/lib/workspaces/permissions/utils'
@@ -281,12 +282,13 @@ export const DELETE = withRouteHandler(
           expectedPermissionId: existingMember.id,
         })
         if (!result.revoked) {
-          if (result.reason === 'membership-changed') return false
+          if (result.reason === 'membership-changed') return null
           throw new WorkspaceBillingAccountRemovalError()
         }
-        return true
+        return result
       })
       if (!removed) return adminNotFoundResponse('Workspace member')
+      deliverRunStops(removed.stoppedRuns)
 
       logger.info(`Admin API: Removed member ${memberId} from workspace ${workspaceId}`, {
         userId: existingMember.userId,

apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts

@@ -50,6 +50,7 @@ import {
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { syncWorkspaceEnvCredentials } from '@/lib/credentials/environment'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import { ProjectConflictError } from '@/lib/projects/errors'
 import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
 import { getWorkspaceById } from '@/lib/workspaces/permissions/utils'
@@ -397,12 +398,13 @@ export const DELETE = withRouteHandler(
           expectedPermissionId: existingPermission.id,
         })
         if (!result.revoked) {
-          if (result.reason === 'membership-changed') return false
+          if (result.reason === 'membership-changed') return null
           throw new WorkspaceBillingAccountRemovalError()
         }
-        return true
+        return result
       })
       if (!removed) return adminNotFoundResponse('Workspace member')
+      deliverRunStops(removed.stoppedRuns)
 
       logger.info(`Admin API: Removed user ${userId} from workspace ${workspaceId}`)
 

apps/sim/app/api/workspaces/[id]/route.ts

@@ -17,6 +17,7 @@ const logger = createLogger('WorkspaceByIdAPI')
 import { db } from '@sim/db'
 import { permissions, workspace } from '@sim/db/schema'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import {
   getEffectiveWorkspacePermission,
   getUserEntityPermissions,
@@ -284,6 +285,7 @@ export const DELETE = withRouteHandler(
 
       const archiveResult = await archiveWorkspace(workspaceId, {
         requestId: `workspace-${workspaceId}`,
+        deliverStops: deliverRunStops,
       })
 
       /** Recorded first: a retry that finishes an earlier partial deletion still archives it. */

apps/sim/ee/scim/lib/application/users/deprovision-user.ts

@@ -5,6 +5,7 @@ import { createLogger } from '@sim/logger'
 import { and, eq } from 'drizzle-orm'
 import { removeUserFromOrganization } from '@/lib/billing/organizations/membership'
 import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import {
   defineAuthorizedScimUseCase,
   type ScimUseCaseArgs,
@@ -74,6 +75,7 @@ export const deprovisionScimUser = defineAuthorizedScimUseCase({
         memberId: membership.id,
         revokePersonalApiKeys: true,
       })
+      deliverRunStops(removal.stoppedRuns ?? [])
       if (!removal.success) {
         throw new ScimError(409, undefined, removal.error ?? 'The member could not be removed')
       }