fix(scim): push Stop for runs a directory sync ends, after its transaction commits

A directory sync that withdrew a workspace grant or demoted an organization
admin recorded Stop for the user's live Chat runs but never pushed it, so the
runs ended only on the worker's next run-control read.

- Projection deltas carry the runs their withdrawals stopped; withdrawGrant and
  setOrganizationRole return them instead of dropping them
- Group writes and user updates deliver them once their transaction commits;
  the batched reconcile delivers each batch after it commits, so a rolled-back
  batch delivers nothing
- Member removal, transfer and external-collaborator removal results carry
  stoppedRuns as a required field, returned from inside the transaction

Author: waleedlatif1

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

@@ -151,7 +151,7 @@ export const POST = withRouteHandler(
         memberId: selfMember.id,
         spareSessionToken: session.session.token,
       })
-      deliverRunStops(removeResult.stoppedRuns ?? [])
+      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

@@ -272,7 +272,7 @@ export const DELETE = withRouteHandler(
         memberId,
         skipBillingLogic,
       })
-      deliverRunStops(result.stoppedRuns ?? [])
+      deliverRunStops(result.stoppedRuns)
 
       if (!result.success) {
         if (result.error === 'Cannot remove organization owner') {

apps/sim/ee/scim/lib/application/groups/manage-groups.test.ts

@@ -94,6 +94,7 @@ describe('replaceScimGroup', () => {
       memberIds.delete(scimUserId)
     )
     mocks.countMembers.mockImplementation(async () => memberIds.size)
+    mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [], stoppedRuns: [] })
     mocks.touch.mockImplementation(async () => {
       group = { ...group, updatedAt: new Date('2026-01-02T00:00:00.000Z') }
     })

apps/sim/ee/scim/lib/application/groups/manage-groups.ts

@@ -5,6 +5,8 @@ import { and, eq, ne } from 'drizzle-orm'
 import type { ScimPatchOperation } from '@/lib/api/contracts/scim'
 import { acquireOrganizationMutationLock } from '@/lib/billing/organizations/membership'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import type { StoppedRun } from '@/lib/mothership/async-runs/repository'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import {
   defineAuthorizedScimUseCase,
   type ScimUseCaseArgs,
@@ -50,16 +52,19 @@ import {
 /**
  * Every group write runs in one transaction under the organization lock, which
  * heads the documented lock order, so two full-membership PATCHes cannot both
- * compute from the same stale membership and keep members from both.
+ * compute from the same stale membership and keep members from both. The runs
+ * its projection stopped are delivered once that transaction commits.
  */
-function withGroupWrite<T>(
+async function withGroupWrite<T extends { stoppedRuns: StoppedRun[] }>(
   context: ScimUseCaseContext,
   work: (tx: DbTransaction) => Promise<T>
-): Promise<T> {
-  return db.transaction(async (tx) => {
+): Promise<Omit<T, 'stoppedRuns'>> {
+  const { stoppedRuns, ...result } = await db.transaction(async (tx) => {
     await acquireOrganizationMutationLock(tx, context.organizationId)
     return work(tx)
   })
+  deliverRunStops(stoppedRuns)
+  return result
 }
 
 async function assertDisplayNameAvailable(
@@ -205,7 +210,7 @@ export const createScimGroup = defineAuthorizedScimUseCase({
           })
         : 'no-match'
 
-      await reconcileUsersProjection(tx, {
+      const projection = await reconcileUsersProjection(tx, {
         connectionId: context.connection.id,
         organizationId: context.organizationId,
         scimUserIds: memberIds,
@@ -225,6 +230,7 @@ export const createScimGroup = defineAuthorizedScimUseCase({
         resource: toGroupResource({ ...created, members }, context.baseUrl),
         touchedUserIds: memberIds,
         renamed: true,
+        stoppedRuns: projection.stoppedRuns,
       }
     })
   },
@@ -295,7 +301,7 @@ export const replaceScimGroup = defineAuthorizedScimUseCase({
       await assertMemberCount(tx, current.id)
       if (touched.size > 0) await touchScimGroup(tx, current.id)
 
-      await reconcileUsersProjection(tx, {
+      const projection = await reconcileUsersProjection(tx, {
         connectionId: context.connection.id,
         organizationId: context.organizationId,
         scimUserIds: [...touched],
@@ -317,6 +323,7 @@ export const replaceScimGroup = defineAuthorizedScimUseCase({
         resource: toGroupResource({ ...refreshed, members }, context.baseUrl),
         touchedUserIds: [...touched],
         renamed: renamed || (current.externalId ?? null) !== (refreshed.externalId ?? null),
+        stoppedRuns: projection.stoppedRuns,
       }
     })
   },
@@ -428,7 +435,7 @@ export const patchScimGroup = defineAuthorizedScimUseCase({
       await assertMemberCount(tx, current.id)
       if (touched.size > 0) await touchScimGroup(tx, current.id)
 
-      await reconcileUsersProjection(tx, {
+      const projection = await reconcileUsersProjection(tx, {
         connectionId: context.connection.id,
         organizationId: context.organizationId,
         scimUserIds: [...touched],
@@ -447,6 +454,7 @@ export const patchScimGroup = defineAuthorizedScimUseCase({
           patch.kind === 'full' ? (patch.displayName ?? current.displayName) : current.displayName,
         touchedUserIds: [...touched],
         renamed,
+        stoppedRuns: projection.stoppedRuns,
       }
     })
   },
@@ -481,7 +489,7 @@ export const deleteScimGroup = defineAuthorizedScimUseCase({
        */
       const memberIds = await loadGroupMemberIds(tx, current.id)
       await deleteScimGroupRow(tx, current.id)
-      await reconcileUsersProjection(tx, {
+      const projection = await reconcileUsersProjection(tx, {
         connectionId: context.connection.id,
         organizationId: context.organizationId,
         scimUserIds: memberIds,
@@ -491,6 +499,7 @@ export const deleteScimGroup = defineAuthorizedScimUseCase({
         groupId: current.id,
         displayName: current.displayName,
         memberCount: memberIds.length,
+        stoppedRuns: projection.stoppedRuns,
       }
     })
   },

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

@@ -75,7 +75,7 @@ export const deprovisionScimUser = defineAuthorizedScimUseCase({
         memberId: membership.id,
         revokePersonalApiKeys: true,
       })
-      deliverRunStops(removal.stoppedRuns ?? [])
+      deliverRunStops(removal.stoppedRuns)
       if (!removal.success) {
         throw new ScimError(409, undefined, removal.error ?? 'The member could not be removed')
       }

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

@@ -215,6 +215,7 @@ export const provisionScimUser = defineAuthorizedScimUseCase({
           connectionId: context.connection.id,
           externalId: attributes.externalId,
         })
+        /** The new directory record has no provenance to withdraw, so this stops no runs. */
         await reconcileUserProjection(tx, {
           connectionId: context.connection.id,
           organizationId: context.organizationId,

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

@@ -133,7 +133,7 @@ describe('user updates', () => {
   beforeEach(() => {
     resetDbChainMock()
     mocks.loadGroups.mockResolvedValue(new Map())
-    mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [] })
+    mocks.reconcile.mockResolvedValue({ added: [], removed: [], raised: [], stoppedRuns: [] })
   })
 
   it('serializes on the organization and user locks before computing the update', async () => {

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

@@ -5,6 +5,8 @@ import { normalizeEmail } from '@sim/utils/string'
 import type { ScimPatchOperation } from '@/lib/api/contracts/scim'
 import { acquireOrganizationUserMutationLocks } from '@/lib/billing/organizations/membership'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import type { StoppedRun } from '@/lib/mothership/async-runs/repository'
+import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery'
 import { suspendMemberTx, unsuspendMemberTx } from '@/lib/organizations/members/lifecycle'
 import {
   invalidateAfterSessionRevocation,
@@ -42,18 +44,24 @@ import {
  * deactivation actually does.
  */
 
-export interface UpdateOutcome {
+interface UpdateOutcome {
   emailChanged: boolean
   deactivated: boolean
   reactivated: boolean
 }
 
+/** An applied update and the runs its projection stopped, delivered once the write commits. */
+interface AppliedUserUpdate {
+  outcome: UpdateOutcome
+  stoppedRuns: StoppedRun[]
+}
+
 async function applyUserUpdate(
   tx: DbTransaction,
   context: ScimUseCaseContext,
   current: ScimUserRecord,
   next: ScimUserAttributes
-): Promise<UpdateOutcome> {
+): Promise<AppliedUserUpdate> {
   const nextEmail = primaryEmail(next)
   const nextName = accountName(next)
   const emailChanged = accountEmailDiverged(current, next)
@@ -108,14 +116,17 @@ async function applyUserUpdate(
     active: next.active,
   })
 
-  await reconcileUserProjection(tx, {
+  const projection = await reconcileUserProjection(tx, {
     connectionId: context.connection.id,
     organizationId: context.organizationId,
     scimUserId: current.id,
     settings: context.connection.settings,
   })
 
-  return { emailChanged, deactivated, reactivated }
+  return {
+    outcome: { emailChanged, deactivated, reactivated },
+    stoppedRuns: projection.stoppedRuns,
+  }
 }
 
 export interface UpdateScimUserResult {
@@ -125,6 +136,27 @@ export interface UpdateScimUserResult {
   resource: ReturnType<typeof toUserResource>
 }
 
+/** Runs a user write in one transaction and delivers the runs it stopped once that commits. */
+async function writeUser(
+  context: ScimUseCaseContext,
+  scimUserId: string,
+  update: (tx: DbTransaction, current: ScimUserRecord) => Promise<AppliedUserUpdate | null>
+): Promise<UpdateScimUserResult> {
+  const { stoppedRuns, ...result } = await db.transaction(async (tx) => {
+    const current = await loadUserForUpdate(tx, context, scimUserId)
+    const applied = await update(tx, current)
+    return {
+      scimUserId: current.id,
+      userId: current.userId,
+      outcome: applied?.outcome ?? null,
+      resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl),
+      stoppedRuns: applied?.stoppedRuns ?? [],
+    }
+  })
+  deliverRunStops(stoppedRuns)
+  return result
+}
+
 /**
  * Loads the stored resource under the organization and user advisory locks.
  *
@@ -229,9 +261,7 @@ export const replaceScimUser = defineAuthorizedScimUseCase({
     input,
     context,
   }: ScimUseCaseArgs<ReplaceScimUserInput>): Promise<UpdateScimUserResult> {
-    return db.transaction(async (tx) => {
-      const current = await loadUserForUpdate(tx, context, input.scimUserId)
-
+    return writeUser(context, input.scimUserId, async (tx, current) => {
       /**
        * A replace keeps attributes Sim does not model that the directory sent on
        * a previous write but omitted now, so a partial mapping does not erase
@@ -249,16 +279,10 @@ export const replaceScimUser = defineAuthorizedScimUseCase({
        * changes nothing must not write, audit, or re-project, or a 2,000-user
        * organization produces 2,000 spurious audit rows per sync.
        */
-      const outcome =
-        userAttributesEqual(current.attributes, next) && !accountIdentityDiverged(current, next)
-          ? null
-          : await applyUserUpdate(tx, context, current, next)
-      return {
-        scimUserId: current.id,
-        userId: current.userId,
-        outcome,
-        resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl),
-      }
+      return userAttributesEqual(current.attributes, next) &&
+        !accountIdentityDiverged(current, next)
+        ? null
+        : applyUserUpdate(tx, context, current, next)
     })
   },
   projectAudit: ({ result }) => auditEntries(result),
@@ -277,9 +301,7 @@ export const patchScimUser = defineAuthorizedScimUseCase({
     input,
     context,
   }: ScimUseCaseArgs<PatchScimUserInput>): Promise<UpdateScimUserResult> {
-    return db.transaction(async (tx) => {
-      const current = await loadUserForUpdate(tx, context, input.scimUserId)
-
+    return writeUser(context, input.scimUserId, async (tx, current) => {
       const { next, changed } = applyUserPatch(current.attributes, input.operations)
 
       /**
@@ -289,16 +311,9 @@ export const patchScimUser = defineAuthorizedScimUseCase({
        * projection pass, and a `lastModified` bump for a request that meant
        * nothing.
        */
-      const outcome =
-        changed || accountIdentityDiverged(current, next)
-          ? await applyUserUpdate(tx, context, current, next)
-          : null
-      return {
-        scimUserId: current.id,
-        userId: current.userId,
-        outcome,
-        resource: await renderUpdated(tx, context.connection.id, current.id, context.baseUrl),
-      }
+      return changed || accountIdentityDiverged(current, next)
+        ? applyUserUpdate(tx, context, current, next)
+        : null
     })
   },
   projectAudit: ({ result }) => auditEntries(result),