feat(mothership): erase private memory when a member leaves or an organization is deleted

apps/sim/lib/billing/organizations/membership.ts

@@ -61,6 +61,7 @@ import { isRetryableTransactionError } from '@/lib/db/transaction'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { acquireInvitationMutationLocks } from '@/lib/invitations/locks'
 import { type StoppedRun, stopLiveRunsTx } from '@/lib/mothership/async-runs/repository'
+import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox'
 import { requireMemberManagementAuthority } from '@/lib/organizations/members/authority'
 import {
   revokePersonalApiKeysTx,
@@ -1192,6 +1193,13 @@ export async function transferUserBetweenOrganizations(
         if (deleted.length === 0) {
           throw new Error('Member could not be transferred because their role changed')
         }
+        await enqueueMemoryCleanup(tx, [
+          {
+            kind: 'organization_member',
+            organizationId: params.sourceOrganizationId,
+            userId: params.userId,
+          },
+        ])
 
         const cancelledInvitations = invitationIds.length
           ? await tx
@@ -1407,6 +1415,7 @@ export async function removeUserFromOrganization(
             'The membership changed before removal. Refresh and try again.'
           )
         }
+        await enqueueMemoryCleanup(tx, [{ kind: 'organization_member', organizationId, userId }])
 
         if (!skipBillingLogic) {
           await enqueueOutboxEvent(tx, MEMBER_BILLING_RECONCILIATION_EVENT_TYPE, {
@@ -1679,6 +1688,7 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: {
           )
 
         await setOrgMemberUsageLimit(organizationId, userId, null, undefined, tx)
+        await enqueueMemoryCleanup(tx, [{ kind: 'organization_member', organizationId, userId }])
 
         const cancelledInvitations = invitationIds.length
           ? await tx

apps/sim/lib/mothership/generated/memory-notes-access.ts

@@ -53,6 +53,14 @@ export const MemoryCleanupInput = z.strictObject({
     z.strictObject({ kind: z.literal("private_owner"), userId: z.string().min(1).max(200) }),
     z.strictObject({ kind: z.literal("workspace"), workspaceId: z.uuid() }),
     z.strictObject({ kind: z.literal("project"), projectId: z.uuid() }),
+    // Erase private memory held in an organization: all of it when the organization is
+    // deleted, one person's when they leave it (D106).
+    z.strictObject({ kind: z.literal("organization"), organizationId: z.string().min(1).max(200) }),
+    z.strictObject({
+      kind: z.literal("organization_member"),
+      organizationId: z.string().min(1).max(200),
+      userId: z.string().min(1).max(200),
+    }),
   ]),
 });
 export type MemoryCleanupInput = z.infer<typeof MemoryCleanupInput>;

apps/sim/lib/mothership/memories/offboarding.integration.ts

@@ -0,0 +1,145 @@
+import { db } from '@sim/db'
+import { member, organization, outboxEvent, permissions, user, workspace } from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
+import { generateId } from '@sim/utils/id'
+import { and, eq, inArray, sql } from 'drizzle-orm'
+import { afterAll, beforeAll, expect, it } from 'vitest'
+
+readTestDatabaseUrl()
+
+const ownerId = generateId()
+const memberId = generateId()
+const externalId = generateId()
+const organizationId = generateId()
+const membershipId = generateId()
+const workspaceId = generateId()
+let runtime: {
+  membership: typeof import('@/lib/billing/organizations/membership')
+  resources: typeof import('@/lib/organizations/resource-cleanup')
+  event: typeof import('@/lib/mothership/memories/cleanup-event')
+}
+
+beforeAll(async () => {
+  // The cleanup outbox reads its worker credential and destination when it loads.
+  process.env.SIM_AGENT_API_URL = 'http://127.0.0.1:9'
+  process.env.COPILOT_API_KEY = 'memory-offboarding-fixture-key'
+  const [membership, resources, event] = await Promise.all([
+    import('@/lib/billing/organizations/membership'),
+    import('@/lib/organizations/resource-cleanup'),
+    import('@/lib/mothership/memories/cleanup-event'),
+  ])
+  runtime = { membership, resources, event }
+  await db.insert(user).values(
+    [ownerId, memberId, externalId].map((id) => ({
+      id,
+      name: 'Offboarding fixture',
+      email: `${id}@memory.test`,
+      emailVerified: true,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+    }))
+  )
+  await db
+    .insert(organization)
+    .values({ id: organizationId, name: 'Offboarding', slug: organizationId })
+  await db.insert(member).values([
+    { id: generateId(), organizationId, userId: ownerId, role: 'owner', createdAt: new Date() },
+    { id: membershipId, organizationId, userId: memberId, role: 'member', createdAt: new Date() },
+  ])
+  // The external collaborator works in one of the organization's workspaces without membership.
+  await insertWorkspaceFixture(db, {
+    id: workspaceId,
+    name: 'Offboarding',
+    ownerId,
+    billedAccountUserId: ownerId,
+    organizationId,
+    workspaceMode: 'organization',
+  })
+  await db.insert(permissions).values({
+    id: generateId(),
+    userId: externalId,
+    entityType: 'workspace',
+    entityId: workspaceId,
+    permissionType: 'write',
+  })
+}, 60_000)
+
+afterAll(async () => {
+  await db
+    .delete(outboxEvent)
+    .where(
+      and(
+        eq(outboxEvent.eventType, runtime.event.MEMORY_CLEANUP_EVENT),
+        sql`${outboxEvent.payload}->'request'->'target'->>'organizationId' = ${organizationId}`
+      )
+    )
+  await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId))
+  await db.delete(organization).where(eq(organization.id, organizationId))
+  await db.delete(user).where(inArray(user.id, [ownerId, memberId, externalId]))
+})
+
+/** The memory erasures queued for this organization, as `kind:userId` (or `kind`). */
+async function queued(): Promise<string[]> {
+  const rows = await db
+    .select({ payload: outboxEvent.payload })
+    .from(outboxEvent)
+    .where(
+      and(
+        eq(outboxEvent.eventType, runtime.event.MEMORY_CLEANUP_EVENT),
+        sql`${outboxEvent.payload}->'request'->'target'->>'organizationId' = ${organizationId}`
+      )
+    )
+  return rows
+    .map(({ payload }) => {
+      const target = (payload as { request: { target: { kind: string; userId?: string } } }).request
+        .target
+      return target.userId ? `${target.kind}:${target.userId}` : target.kind
+    })
+    .sort()
+}
+
+it('queues private memory erasure with each committed departure and organization deletion, never on rollback', async () => {
+  // A removal whose actor lacks authority rolls back with nothing queued.
+  await expect(
+    runtime.membership.removeUserFromOrganization({
+      organizationId,
+      userId: memberId,
+      memberId: membershipId,
+      actorUserId: externalId,
+      skipBillingLogic: true,
+      onError: 'throw',
+    })
+  ).rejects.toThrow()
+  expect(await queued()).toEqual([])
+
+  await runtime.membership.removeUserFromOrganization({
+    organizationId,
+    userId: memberId,
+    memberId: membershipId,
+    skipBillingLogic: true,
+    onError: 'throw',
+  })
+  expect(
+    await runtime.membership.removeExternalUserFromOrganizationWorkspaces({
+      organizationId,
+      userId: externalId,
+    })
+  ).toMatchObject({ success: true, workspaceAccessRevoked: 1 })
+  expect(await queued()).toEqual(
+    [`organization_member:${externalId}`, `organization_member:${memberId}`].sort()
+  )
+
+  // Organization deletion queues the erasure in the delete transaction: a rollback drops it.
+  await expect(
+    db.transaction(async (tx) => {
+      await runtime.resources.enqueueOrganizationResourceCleanup(tx, organizationId)
+      throw new Error('rollback fixture')
+    })
+  ).rejects.toThrow('rollback fixture')
+  expect(await queued()).not.toContain('organization')
+  await db.transaction(async (tx) => {
+    await runtime.resources.enqueueOrganizationResourceCleanup(tx, organizationId)
+  })
+  expect(await queued()).toContain('organization')
+}, 30_000)

apps/sim/lib/organizations/resource-cleanup.ts

@@ -8,6 +8,7 @@ import { env } from '@/lib/core/config/env'
 import { enqueueOutboxEvent, type OutboxHandlerRegistry } from '@/lib/core/outbox/service'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import type { DbTransaction } from '@/lib/db/types'
+import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox'
 import { ORGANIZATION_RESOURCE_CLEANUP_EVENT } from '@/lib/organizations/resource-cleanup-event'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 
@@ -35,7 +36,8 @@ const organizationResourceCleanupSchema = z.discriminatedUnion('kind', [
  * Captures canonical bindings before the org FK cascade removes them. The org
  * row lock excludes new direct resources; a rebound resource is checked again
  * by the worker. Each keyset page becomes its own bounded outbox event in the
- * caller's delete transaction, so rollback cannot leave cleanup work behind.
+ * caller's delete transaction, so rollback cannot leave cleanup work behind. The
+ * members' private Mothership memory in the organization is erased the same way.
  */
 export async function enqueueOrganizationResourceCleanup(
   tx: DbTransaction,
@@ -105,6 +107,8 @@ export async function enqueueOrganizationResourceCleanup(
     chatCursor = chats[chats.length - 1].id
     if (chats.length < CLEANUP_BATCH_SIZE) break
   }
+
+  await enqueueMemoryCleanup(tx, [{ kind: 'organization', organizationId }])
 }
 
 export const organizationResourceCleanupOutboxHandlers = {