diff --git a/apps/sim/app/api/v1/admin/workspaces/[id]/members/[memberId]/route.ts b/apps/sim/app/api/v1/admin/workspaces/[id]/members/[memberId]/route.ts
index be48377a5ca..8cd732d3cc4 100644
--- a/apps/sim/app/api/v1/admin/workspaces/[id]/members/[memberId]/route.ts
+++ b/apps/sim/app/api/v1/admin/workspaces/[id]/members/[memberId]/route.ts
@@ -33,14 +33,9 @@ import {
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { revokeWorkspaceCredentialMembershipsTx } from '@/lib/credentials/access'
-import { removeWorkspaceSkillMembershipsTx } from '@/lib/skills/access'
+import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
 import { getWorkspaceById } from '@/lib/workspaces/permissions/utils'
-import {
-  reassignWorkflowOwnershipForWorkspaceMemberRemovalTx,
-  transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx,
-  WorkspaceBillingAccountRemovalError,
-} from '@/lib/workspaces/utils'
+import { WorkspaceBillingAccountRemovalError } from '@/lib/workspaces/utils'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 import {
   badRequestResponse,
@@ -278,28 +273,19 @@ export const DELETE = withRouteHandler(
         )
       }
 
-      await db.transaction(async (tx) => {
-        await transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx({
-          tx,
+      const removed = await db.transaction(async (tx) => {
+        const result = await revokeWorkspaceAccessTx(tx, {
           workspaceId,
-          departingUserId: existingMember.userId,
+          userId: existingMember.userId,
+          expectedPermissionId: existingMember.id,
         })
-
-        const workflowOwnershipReassignment =
-          await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
-            tx,
-            workspaceIds: [workspaceId],
-            departingUserId: existingMember.userId,
-          })
-        if (workflowOwnershipReassignment.unresolved.length > 0) {
+        if (!result.revoked) {
+          if (result.reason === 'membership-changed') return false
           throw new WorkspaceBillingAccountRemovalError()
         }
-
-        await tx.delete(permissions).where(eq(permissions.id, memberId))
-
-        await revokeWorkspaceCredentialMembershipsTx(tx, workspaceId, existingMember.userId)
-        await removeWorkspaceSkillMembershipsTx(tx, workspaceId, existingMember.userId)
+        return true
       })
+      if (!removed) return notFoundResponse('Workspace member')
 
       logger.info(`Admin API: Removed member ${memberId} from workspace ${workspaceId}`, {
         userId: existingMember.userId,
diff --git a/apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts b/apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts
index 1b3b429a9d3..0f5682492a6 100644
--- a/apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts
+++ b/apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts
@@ -49,15 +49,10 @@ import {
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { revokeWorkspaceCredentialMembershipsTx } from '@/lib/credentials/access'
 import { syncWorkspaceEnvCredentials } from '@/lib/credentials/environment'
-import { removeWorkspaceSkillMembershipsTx } from '@/lib/skills/access'
+import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
 import { getWorkspaceById } from '@/lib/workspaces/permissions/utils'
-import {
-  reassignWorkflowOwnershipForWorkspaceMemberRemovalTx,
-  transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx,
-  WorkspaceBillingAccountRemovalError,
-} from '@/lib/workspaces/utils'
+import { WorkspaceBillingAccountRemovalError } from '@/lib/workspaces/utils'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 import {
   badRequestResponse,
@@ -395,26 +390,8 @@ export const DELETE = withRouteHandler(
       }
 
       await db.transaction(async (tx) => {
-        await transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx({
-          tx,
-          workspaceId,
-          departingUserId: userId,
-        })
-
-        const workflowOwnershipReassignment =
-          await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
-            tx,
-            workspaceIds: [workspaceId],
-            departingUserId: userId,
-          })
-        if (workflowOwnershipReassignment.unresolved.length > 0) {
-          throw new WorkspaceBillingAccountRemovalError()
-        }
-
-        await tx.delete(permissions).where(eq(permissions.id, existingPermission.id))
-
-        await revokeWorkspaceCredentialMembershipsTx(tx, workspaceId, userId)
-        await removeWorkspaceSkillMembershipsTx(tx, workspaceId, userId)
+        const result = await revokeWorkspaceAccessTx(tx, { workspaceId, userId: userId })
+        if (!result.revoked) throw new WorkspaceBillingAccountRemovalError()
       })
 
       logger.info(`Admin API: Removed user ${userId} from workspace ${workspaceId}`)
diff --git a/apps/sim/background/cleanup-soft-deletes.ts b/apps/sim/background/cleanup-soft-deletes.ts
index ba955e7e2ab..7688f100673 100644
--- a/apps/sim/background/cleanup-soft-deletes.ts
+++ b/apps/sim/background/cleanup-soft-deletes.ts
@@ -19,6 +19,7 @@ import { and, asc, eq, inArray, isNotNull, isNull, lt, sql } from 'drizzle-orm'
 import { type CleanupJobPayload, runCleanupWithLimits } from '@/lib/billing/cleanup-dispatcher'
 import {
   decrementStorageUsageForBillingContextInTx,
+  lockWorkspaceStorageForMutationInTx,
   resolveStorageBillingContext,
   type StorageBillingContext,
 } from '@/lib/billing/storage'
@@ -334,6 +335,7 @@ async function deleteExpiredBillableWorkspaceFileRows(
     for (const batch of chunkArray(workspaceRows, DEFAULT_DELETE_CHUNK_SIZE)) {
       try {
         const deletedCount = await db.transaction(async (tx) => {
+          await lockWorkspaceStorageForMutationInTx(tx, workspaceId)
           await releaseWorkspaceFileVersionsForPurgeInTx(
             tx,
             batch.map(({ id }) => id),
diff --git a/apps/sim/lib/billing/organizations/membership-external-removal.test.ts b/apps/sim/lib/billing/organizations/membership-external-removal.test.ts
index a0484192798..3951dbc3860 100644
--- a/apps/sim/lib/billing/organizations/membership-external-removal.test.ts
+++ b/apps/sim/lib/billing/organizations/membership-external-removal.test.ts
@@ -1,4 +1,4 @@
-import { credential, knowledgeBase, member, workspaceFiles } from '@sim/db/schema'
+import { credential, knowledgeBase, member, user, workspaceFiles } from '@sim/db/schema'
 import { dbChainMockFns, hasMockCondition, queueTableRows, resetDbChainMock } from '@sim/testing'
 import {
   organizationMemberLimitsMock,
@@ -39,6 +39,10 @@ describe('external organization access removal', () => {
   it('preserves organization indexes and cached documents when a creator leaves with no workspaces', async () => {
     queueTableRows(member, [{ id: 'membership', userId: 'departing', role: 'admin' }])
     queueTableRows(member, [{ userId: 'surviving-owner' }])
+    queueTableRows(user, [{ email: 'departing@example.test' }])
+    queueTableRows(user, [{ email: 'departing@example.test' }])
+    queueTableRows(user, [{ id: 'surviving-owner' }])
+    queueTableRows(workspaceFiles, [{ id: 'cached-file', context: 'knowledge-base' }])
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'membership' }])
 
     const result = await removeUserFromOrganization({
@@ -46,6 +50,7 @@ describe('external organization access removal', () => {
       organizationId: 'organization-a',
       memberId: 'membership',
       skipBillingLogic: true,
+      onError: 'throw',
     })
 
     expect(result.success).toBe(true)
diff --git a/apps/sim/lib/billing/organizations/membership.ts b/apps/sim/lib/billing/organizations/membership.ts
index e93f4bcd0e2..5ac80e94210 100644
--- a/apps/sim/lib/billing/organizations/membership.ts
+++ b/apps/sim/lib/billing/organizations/membership.ts
@@ -10,7 +10,6 @@ import {
   account,
   credential,
   invitation,
-  knowledgeBase,
   member,
   organization,
   permissionGroupMember,
@@ -19,7 +18,6 @@ import {
   user,
   userStats,
   workspace,
-  workspaceFiles,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
@@ -39,6 +37,7 @@ import {
 import { acquireUserBillingIdentityLock } from '@/lib/billing/organizations/billing-identity-lock'
 import { setOrgMemberUsageLimit } from '@/lib/billing/organizations/member-limits'
 import { MEMBER_BILLING_RECONCILIATION_EVENT_TYPE } from '@/lib/billing/organizations/membership-reconciliation-event'
+import { reassignOrganizationSharedResourcesTx } from '@/lib/billing/organizations/resource-handoff'
 import { isPaid, sqlIsPro } from '@/lib/billing/plan-helpers'
 import { changeOrganizationWorkspaceBilledAccountsInTx } from '@/lib/billing/storage/payer-transfer'
 import {
@@ -63,10 +62,8 @@ import {
 } from '@/lib/organizations/members/revocation'
 import { reassignOrganizationProjects } from '@/lib/projects/membership'
 import { removeWorkspaceSkillMembershipsTx } from '@/lib/skills/access'
-import {
-  reassignWorkflowOwnershipForWorkspaceMemberRemovalTx,
-  WorkspaceBillingAccountRemovalError,
-} from '@/lib/workspaces/utils'
+import { reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx } from '@/lib/workspaces/resource-handoff'
+import { WorkspaceBillingAccountRemovalError } from '@/lib/workspaces/utils'
 import { endDirectoryMembershipTx } from '@/ee/scim/lib/identity/end-directory-membership'
 
 export { acquireUserBillingIdentityLock } from '@/lib/billing/organizations/billing-identity-lock'
@@ -546,14 +543,8 @@ async function reassignOwnedOrganizationResourcesTx({
   organizationId: string
   workspaceIds: string[]
 }) {
-  const [ownerMembership] = await tx
-    .select({ userId: member.userId })
-    .from(member)
-    .where(and(eq(member.organizationId, organizationId), eq(member.role, 'owner')))
-    .limit(1)
-
-  const ownerId = ownerMembership?.userId
-  if (!ownerId || ownerId === userId) return 0
+  const ownerId = await reassignOrganizationSharedResourcesTx(tx, organizationId, userId)
+  if (!ownerId) return 0
 
   await reassignOrganizationProjects(tx, {
     organizationId,
@@ -562,28 +553,6 @@ async function reassignOwnedOrganizationResourcesTx({
     workspaceIds,
   })
 
-  /** Creator attribution must survive account deletion without changing document ACLs. */
-  await tx
-    .update(knowledgeBase)
-    .set({ userId: ownerId, updatedAt: new Date() })
-    .where(
-      and(
-        eq(knowledgeBase.organizationId, organizationId),
-        isNull(knowledgeBase.workspaceId),
-        eq(knowledgeBase.userId, userId)
-      )
-    )
-  await tx
-    .update(workspaceFiles)
-    .set({ userId: ownerId, updatedAt: new Date() })
-    .where(
-      and(
-        eq(workspaceFiles.organizationId, organizationId),
-        isNull(workspaceFiles.workspaceId),
-        eq(workspaceFiles.userId, userId)
-      )
-    )
-
   if (workspaceIds.length === 0) return 0
 
   const reassignedWorkspaces = await tx
@@ -1205,13 +1174,13 @@ export async function transferUserBetweenOrganizations(
           workspaceIds,
         })
         if (workspaceIds.length > 0) {
-          const workflowOwnershipReassignment =
-            await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
+          const resourceOwnershipReassignment =
+            await reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
               tx,
               workspaceIds,
               departingUserId: params.userId,
             })
-          if (workflowOwnershipReassignment.unresolved.length > 0) {
+          if (resourceOwnershipReassignment.unresolved.length > 0) {
             throw new WorkspaceBillingAccountRemovalError()
           }
           const deletedPermissions = await tx
@@ -1434,13 +1403,13 @@ export async function removeUserFromOrganization(
           }
         }
 
-        const workflowOwnershipReassignment =
-          await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
+        const resourceOwnershipReassignment =
+          await reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
             tx,
             workspaceIds,
             departingUserId: userId,
           })
-        if (workflowOwnershipReassignment.unresolved.length > 0) {
+        if (resourceOwnershipReassignment.unresolved.length > 0) {
           throw new WorkspaceBillingAccountRemovalError()
         }
 
@@ -1651,13 +1620,13 @@ export async function removeExternalUserFromOrganizationWorkspaces(params: {
           }
         }
 
-        const workflowOwnershipReassignment =
-          await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
+        const resourceOwnershipReassignment =
+          await reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
             tx,
             workspaceIds,
             departingUserId: userId,
           })
-        if (workflowOwnershipReassignment.unresolved.length > 0) {
+        if (resourceOwnershipReassignment.unresolved.length > 0) {
           throw new WorkspaceBillingAccountRemovalError()
         }
 
diff --git a/apps/sim/lib/billing/organizations/resource-handoff.ts b/apps/sim/lib/billing/organizations/resource-handoff.ts
new file mode 100644
index 00000000000..6d1efb390ff
--- /dev/null
+++ b/apps/sim/lib/billing/organizations/resource-handoff.ts
@@ -0,0 +1,105 @@
+import {
+  knowledgeBase,
+  member,
+  permissionGroup,
+  ssoProvider,
+  user,
+  workspaceFiles,
+} from '@sim/db/schema'
+import { and, asc, eq, isNull, ne, sql } from 'drizzle-orm'
+import { OrchestrationError } from '@/lib/core/orchestration/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import { handoffFileCreatorsInTx } from '@/lib/uploads/contexts/workspace/creator-handoff'
+
+/** Organization references survive even when their creator no longer has a membership row. */
+export async function listSharedResourceOrganizationIdsForUser(executor: DbOrTx, userId: string) {
+  const rows = await executor.execute<{ organizationId: string }>(sql`
+    SELECT DISTINCT "organizationId" FROM (
+      SELECT ${knowledgeBase.organizationId} AS "organizationId" FROM ${knowledgeBase}
+        WHERE ${knowledgeBase.userId} = ${userId} AND ${knowledgeBase.workspaceId} IS NULL
+      UNION ALL SELECT ${workspaceFiles.organizationId} FROM ${workspaceFiles}
+        WHERE ${workspaceFiles.userId} = ${userId} AND ${workspaceFiles.workspaceId} IS NULL
+      UNION ALL SELECT ${permissionGroup.organizationId} FROM ${permissionGroup} WHERE ${permissionGroup.createdBy} = ${userId}
+      UNION ALL SELECT ${ssoProvider.organizationId} FROM ${ssoProvider} WHERE ${ssoProvider.userId} = ${userId}
+    ) owned WHERE "organizationId" IS NOT NULL ORDER BY "organizationId"
+  `)
+  return rows.map((row) => row.organizationId)
+}
+
+/** Retains organization-owned configuration without rewriting ACLs, credentials, or historical actors. */
+export async function reassignOrganizationSharedResourcesTx(
+  tx: DbTransaction,
+  organizationId: string,
+  departingUserId: string
+): Promise<string | null> {
+  const [successor] = await tx
+    .select({ userId: member.userId })
+    .from(member)
+    .where(
+      and(
+        eq(member.organizationId, organizationId),
+        eq(member.role, 'owner'),
+        ne(member.userId, departingUserId)
+      )
+    )
+    .orderBy(asc(member.userId))
+    .limit(1)
+    .for('share')
+  if (!successor) {
+    const owned = await listSharedResourceOrganizationIdsForUser(tx, departingUserId)
+    if (owned.includes(organizationId)) {
+      throw new OrchestrationError(
+        'conflict',
+        'Shared organization resources need an active organization owner before this account can leave or be deleted.'
+      )
+    }
+    return null
+  }
+  const [account] = await tx
+    .select({ id: user.id })
+    .from(user)
+    .where(eq(user.id, successor.userId))
+    .for('key share')
+  if (!account)
+    throw new OrchestrationError(
+      'conflict',
+      'The organization owner changed. Nothing was changed; try again.'
+    )
+  const userId = successor.userId
+  const updatedAt = new Date()
+  await tx
+    .update(knowledgeBase)
+    .set({ userId, updatedAt })
+    .where(
+      and(
+        eq(knowledgeBase.organizationId, organizationId),
+        isNull(knowledgeBase.workspaceId),
+        eq(knowledgeBase.userId, departingUserId)
+      )
+    )
+  await handoffFileCreatorsInTx(
+    tx,
+    and(
+      eq(workspaceFiles.organizationId, organizationId),
+      isNull(workspaceFiles.workspaceId),
+      eq(workspaceFiles.userId, departingUserId)
+    ),
+    userId
+  )
+  await tx
+    .update(permissionGroup)
+    .set({ createdBy: userId, updatedAt })
+    .where(
+      and(
+        eq(permissionGroup.organizationId, organizationId),
+        eq(permissionGroup.createdBy, departingUserId)
+      )
+    )
+  await tx
+    .update(ssoProvider)
+    .set({ userId })
+    .where(
+      and(eq(ssoProvider.organizationId, organizationId), eq(ssoProvider.userId, departingUserId))
+    )
+  return userId
+}
diff --git a/apps/sim/lib/billing/storage/index.ts b/apps/sim/lib/billing/storage/index.ts
index 4be6f5cd100..b96d27b95f2 100644
--- a/apps/sim/lib/billing/storage/index.ts
+++ b/apps/sim/lib/billing/storage/index.ts
@@ -15,6 +15,7 @@ export {
   incrementAdmittedStorageUsageForBillingContextInTx,
   incrementStorageUsageForBillingContextInTx,
   type LegacyStorageUsageDelta,
+  lockWorkspaceStorageForMutationInTx,
   maybeNotifyStorageLimitForBillingContext,
   type WorkspaceStorageUsageDelta,
 } from './tracking'
diff --git a/apps/sim/lib/billing/storage/tracking.ts b/apps/sim/lib/billing/storage/tracking.ts
index e583883810f..dad16efefb6 100644
--- a/apps/sim/lib/billing/storage/tracking.ts
+++ b/apps/sim/lib/billing/storage/tracking.ts
@@ -430,6 +430,29 @@ export async function applyStorageUsageDeltasInTx(
   return destinationUpdatedUsage
 }
 
+/** Locks the canonical workspace before existing-file mutations that later change its storage ledger. */
+export async function lockWorkspaceStorageForMutationInTx(
+  tx: DbOrTx,
+  workspaceId: string
+): Promise<LockedWorkspaceStorage> {
+  const [workspacePayer] = await tx
+    .select({
+      billedAccountUserId: workspace.billedAccountUserId,
+      organizationId: workspace.organizationId,
+      storageUsedBytes: workspace.storageUsedBytes,
+    })
+    .from(workspace)
+    .where(eq(workspace.id, workspaceId))
+    .for('no key update')
+    .limit(1)
+
+  if (!workspacePayer) {
+    throw new Error(`Workspace ${workspaceId} not found for storage accounting`)
+  }
+
+  return { id: workspaceId, ...workspacePayer }
+}
+
 /**
  * Mutates the durable workspace total and its current routed payer as one
  * transaction. The workspace row is the serialization point shared with payer
@@ -447,25 +470,9 @@ async function mutateWorkspaceStorageUsage(
   maximumUsage: number | undefined,
   context: StorageBillingContext
 ): Promise<WorkspaceStorageMutationResult> {
-  const [workspacePayer] = await tx
-    .select({
-      billedAccountUserId: workspace.billedAccountUserId,
-      organizationId: workspace.organizationId,
-      storageUsedBytes: workspace.storageUsedBytes,
-    })
-    .from(workspace)
-    .where(eq(workspace.id, workspaceId))
-    .for('no key update')
-    .limit(1)
-
-  if (!workspacePayer) {
-    throw new Error(`Workspace ${workspaceId} not found for storage accounting`)
-  }
+  const workspacePayer = await lockWorkspaceStorageForMutationInTx(tx, workspaceId)
 
-  const billingEntity = assertWorkspaceStorageContext(
-    { id: workspaceId, ...workspacePayer },
-    context
-  )
+  const billingEntity = assertWorkspaceStorageContext(workspacePayer, context)
   const currentPayerUsage = await lockStorageUsageForMutation(tx, billingEntity)
 
   if (mutation === 'decrement' && workspacePayer.storageUsedBytes < bytes) {
diff --git a/apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts b/apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts
index 21ab15dacce..6d0ec157e72 100644
--- a/apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts
+++ b/apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts
@@ -7,12 +7,15 @@ import { db, dbFor } from '@sim/db'
 import {
   organization,
   outboxEvent,
+  permissions,
   user,
   workspace,
   workspaceFileSecretProvenance,
   workspaceFiles,
   workspaceFileVersion,
 } from '@sim/db/schema'
+import { sha256Hex } from '@sim/security/hash'
+import { createDeferred } from '@sim/testing/helpers/deferred'
 import { generateId } from '@sim/utils/id'
 import { and, asc, eq, inArray, sql } from 'drizzle-orm'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
@@ -31,16 +34,21 @@ import {
 } from '@/lib/knowledge/__integration__/seed-source-access-fixture'
 import { createFileReadTransport } from '@/lib/mothership/agent-cli/file-read-transport'
 import { runCli } from '@/lib/mothership/agent-cli/run-cli'
+import { handoffFileCreatorsInTx } from '@/lib/uploads/contexts/workspace/creator-handoff'
 import { WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT } from '@/lib/uploads/contexts/workspace/file-outbox-events'
 import {
   deleteWorkspaceFileVersion,
   fetchWorkspaceFileBuffer,
   getWorkspaceFile,
   getWorkspaceFileWithCurrentVersion,
+  purgeCreatedWorkspaceFile,
   updateWorkspaceFileContent,
   uploadWorkspaceFile,
 } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
-import type { WorkspaceFileSecretProvenance } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
+import {
+  replaceWorkspaceFileSecretProvenanceInTx,
+  type WorkspaceFileSecretProvenance,
+} from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
 import {
   getCurrentWorkspaceFileVersion,
   getWorkspaceFileVersion,
@@ -55,7 +63,9 @@ import {
   readWorkspaceFileVersionText,
   revertWorkspaceFileVersion,
 } from '@/lib/workspace-files/application/file-versions'
+import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
 import { runCleanupFileVersions } from '@/background/cleanup-file-versions'
+import { runCleanupSoftDeletes } from '@/background/cleanup-soft-deletes'
 import { projectResolvedSecretModelContent } from '@/executor/utils/resolved-secret-content-projection'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 
@@ -114,6 +124,50 @@ describe('workspace file version history in PostgreSQL', () => {
     return (await fetchWorkspaceFileBuffer({ ...file, key }, { maxBytes: 1024 })).toString()
   }
 
+  async function runContendedFileOperations(
+    fileId: string,
+    operations: (() => Promise<unknown>)[]
+  ) {
+    const ready = createDeferred<number>()
+    const release = createDeferred<void>()
+    const blocker = db.transaction(async (tx) => {
+      await tx.select().from(workspaceFiles).where(eq(workspaceFiles.id, fileId)).for('update')
+      const [connection] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+      ready.resolve(connection.pid)
+      await release.promise
+    })
+    const blockerPid = await ready.promise
+    const pending: Promise<PromiseSettledResult<unknown>[]>[] = []
+    try {
+      for (const operation of operations) {
+        pending.push(Promise.allSettled([operation()]))
+        await expect
+          .poll(
+            async () =>
+              (
+                await db.execute(
+                  sql`WITH RECURSIVE blocked(pid) AS (
+                      SELECT pid FROM pg_stat_activity
+                      WHERE ${blockerPid} = ANY(pg_blocking_pids(pid))
+                      UNION
+                      SELECT activity.pid FROM pg_stat_activity activity
+                      JOIN blocked ON blocked.pid = ANY(pg_blocking_pids(activity.pid))
+                    ) SELECT pid FROM blocked`
+                )
+              ).length,
+            { timeout: 5000 }
+          )
+          .toBe(pending.length)
+      }
+    } finally {
+      release.resolve()
+      await blocker
+      expect(
+        (await Promise.all(pending)).flat().filter((result) => result.status === 'rejected')
+      ).toEqual([])
+    }
+  }
+
   it('lists a never-rewritten file as an implicit version 1 attributed to its uploader', async () => {
     const fixture = await seedFile('original')
     const file = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
@@ -132,6 +186,454 @@ describe('workspace file version history in PostgreSQL', () => {
     expect(await versionRows(fixture.fileId)).toEqual([])
   })
 
+  it('keeps original attribution, bytes and revision through repeated creator handoff and a later edit', async () => {
+    const fixture = await seedFile('original')
+    const [before] = await db
+      .select()
+      .from(workspaceFiles)
+      .where(eq(workspaceFiles.id, fixture.fileId))
+    const provenance = {
+      status: 'exact' as const,
+      entries: [
+        {
+          sourceUserId: fixture.aliceId,
+          sourceWorkspaceId: fixture.workspaceId,
+          name: 'HISTORY_FIXTURE',
+          encryptedValue: (await encryptSecret('synthetic-history-value')).encrypted,
+        },
+      ],
+    }
+    await db.transaction((tx) =>
+      replaceWorkspaceFileSecretProvenanceInTx(
+        tx,
+        fixture.fileId,
+        before.contentUpdatedAt,
+        provenance
+      )
+    )
+    const sidecarsBefore = await db
+      .select()
+      .from(workspaceFileSecretProvenance)
+      .where(eq(workspaceFileSecretProvenance.fileId, fixture.fileId))
+    const [workspaceBefore] = await db
+      .select({ bytes: workspace.storageUsedBytes })
+      .from(workspace)
+      .where(eq(workspace.id, fixture.workspaceId))
+    for (const successor of [fixture.bobId, fixture.aliceId, fixture.bobId]) {
+      await db.transaction((tx) =>
+        handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), successor)
+      )
+    }
+    const [after] = await db
+      .select()
+      .from(workspaceFiles)
+      .where(eq(workspaceFiles.id, fixture.fileId))
+    expect(after).toMatchObject({
+      key: before.key,
+      contentUpdatedAt: before.contentUpdatedAt,
+      uploadedAt: before.uploadedAt,
+      sizeBytes: before.sizeBytes,
+      userId: fixture.bobId,
+    })
+    const preserved = await versionRows(fixture.fileId)
+    expect(preserved).toHaveLength(1)
+    expect(preserved[0]).toMatchObject({
+      secretProvenanceStatus: 'exact',
+      secretProvenanceEntries: provenance.entries,
+    })
+    expect(
+      await db
+        .select()
+        .from(workspaceFileSecretProvenance)
+        .where(eq(workspaceFileSecretProvenance.fileId, fixture.fileId))
+    ).toEqual(sidecarsBefore)
+    expect(
+      await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+    ).toEqual([workspaceBefore])
+    await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.bobId,
+      Buffer.from('second'),
+      undefined,
+      { version: { source: 'api', authorUserId: fixture.bobId } }
+    )
+    const rows = await versionRows(fixture.fileId)
+    expect(rows.map((row) => [row.version, row.source, row.authorUserIds])).toEqual([
+      [1, 'upload', [fixture.aliceId]],
+      [2, 'api', [fixture.bobId]],
+    ])
+    expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, rows[0].key)).toBe(
+      'original'
+    )
+    await db.transaction((tx) =>
+      handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), fixture.aliceId)
+    )
+    expect(await versionRows(fixture.fileId)).toEqual(rows)
+  })
+
+  it('does not manufacture original authorship for previously overwritten unrecorded content', async () => {
+    const fixture = await seedFile('overwritten')
+    await db
+      .update(workspaceFiles)
+      .set({ contentUpdatedAt: new Date(Date.now() + 5000) })
+      .where(eq(workspaceFiles.id, fixture.fileId))
+    await db.transaction((tx) =>
+      handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), fixture.bobId)
+    )
+    expect(await versionRows(fixture.fileId)).toMatchObject([
+      { version: 1, source: 'unknown', authorUserIds: [] },
+    ])
+  })
+
+  it.each([false, true])(
+    'preserves empty history across handoff and first-write version one (unknown=%s)',
+    async (unknown) => {
+      const fixture = await seedFile('')
+      if (unknown)
+        await db
+          .update(workspaceFiles)
+          .set({ contentUpdatedAt: new Date(Date.now() + 5000) })
+          .where(eq(workspaceFiles.id, fixture.fileId))
+      const before = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+      if (!before) throw new Error('file missing')
+      const [accounting] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      const historyBefore = await queryWorkspaceFileVersions(before, {
+        sortOrder: 'asc',
+        limit: 10,
+      })
+      expect(historyBefore.versions).toMatchObject([
+        {
+          version: 1,
+          source: unknown ? 'unknown' : 'upload',
+          authorUserIds: unknown ? [] : [fixture.aliceId],
+          size: 0,
+        },
+      ])
+      for (const successor of [fixture.bobId, fixture.aliceId, fixture.bobId]) {
+        await db.transaction((tx) =>
+          handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), successor)
+        )
+      }
+      const after = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+      if (!after) throw new Error('file missing')
+      expect(after.uploadedBy).toBe(fixture.bobId)
+      const historyAfter = await queryWorkspaceFileVersions(after, { sortOrder: 'asc', limit: 10 })
+      expect(historyAfter.versions).toEqual(historyBefore.versions)
+      expect(
+        await db
+          .select({ bytes: workspace.storageUsedBytes })
+          .from(workspace)
+          .where(eq(workspace.id, fixture.workspaceId))
+      ).toEqual([accounting])
+      expect(await objectExists(fixture.firstKey)).toBe(true)
+      await updateWorkspaceFileContent(
+        fixture.workspaceId,
+        fixture.fileId,
+        fixture.bobId,
+        Buffer.from('first'),
+        undefined,
+        {
+          version: { source: 'api', authorUserId: fixture.bobId },
+          secretProvenancePolicy: { mode: 'replace', provenance: { status: 'unknown' } },
+        }
+      )
+      expect(await versionRows(fixture.fileId)).toMatchObject([
+        {
+          version: 1,
+          source: 'api',
+          authorUserIds: [fixture.bobId],
+          contentHash: sha256Hex('first'),
+          restoredFromVersion: null,
+          secretProvenanceStatus: 'unknown',
+          secretProvenanceEntries: [],
+        },
+      ])
+      const [written] = await versionRows(fixture.fileId)
+      expect(written.createdAt).toEqual(written.updatedAt)
+      expect(written.createdAt).not.toEqual(historyBefore.versions[0].createdAt)
+      expect(await objectExists(fixture.firstKey)).toBe(false)
+      expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, written.key)).toBe('first')
+    }
+  )
+
+  it.each(['api', 'revert'] as const)(
+    'keeps an explicitly recorded empty %s version through handoff and a later write',
+    async (source) => {
+      const fixture = await seedFile('')
+      await updateWorkspaceFileContent(
+        fixture.workspaceId,
+        fixture.fileId,
+        fixture.aliceId,
+        Buffer.alloc(0),
+        undefined,
+        {
+          version: {
+            source: 'api',
+            authorUserId: fixture.aliceId,
+          },
+        }
+      )
+      if (source === 'revert') {
+        await updateWorkspaceFileContent(
+          fixture.workspaceId,
+          fixture.fileId,
+          fixture.aliceId,
+          Buffer.from('intermediate'),
+          undefined,
+          { version: { source: 'api', authorUserId: fixture.aliceId } }
+        )
+        await revertWorkspaceFileVersion.execute({
+          principal: { kind: 'session', userId: fixture.aliceId, sessionId: generateId() },
+          input: { fileId: fixture.fileId, assertedWorkspaceId: fixture.workspaceId, version: 1 },
+        })
+      }
+      const previousRows = await versionRows(fixture.fileId)
+      const recorded = previousRows[previousRows.length - 1]
+      await db.transaction((tx) =>
+        handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), fixture.bobId)
+      )
+      await updateWorkspaceFileContent(
+        fixture.workspaceId,
+        fixture.fileId,
+        fixture.bobId,
+        Buffer.from('later'),
+        undefined,
+        { version: { source: 'api', authorUserId: fixture.bobId } }
+      )
+      const rows = await versionRows(fixture.fileId)
+      expect(rows.map((row) => [row.version, row.source, row.authorUserIds])).toEqual([
+        ...previousRows.map((row) => [row.version, row.source, row.authorUserIds]),
+        [recorded.version + 1, 'api', [fixture.bobId]],
+      ])
+      expect(rows[recorded.version - 1]).toMatchObject({
+        key: recorded.key,
+        contentHash: recorded.contentHash,
+        restoredFromVersion: source === 'revert' ? 1 : null,
+      })
+      expect(await objectExists(recorded.key)).toBe(true)
+      expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, recorded.key)).toBe('')
+    }
+  )
+
+  it('keeps an empty storage object still referenced by a retained file after the first write', async () => {
+    const fixture = await seedFile('')
+    const [original] = await db
+      .select()
+      .from(workspaceFiles)
+      .where(eq(workspaceFiles.id, fixture.fileId))
+    await db.insert(workspaceFiles).values({ ...original, id: generateId(), deletedAt: new Date() })
+    await db.transaction((tx) =>
+      handoffFileCreatorsInTx(tx, eq(workspaceFiles.id, fixture.fileId), fixture.bobId)
+    )
+    await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.bobId,
+      Buffer.from('first'),
+      undefined,
+      { version: { source: 'api', authorUserId: fixture.bobId } }
+    )
+    expect(await objectExists(fixture.firstKey)).toBe(true)
+    expect((await versionRows(fixture.fileId)).map((row) => row.version)).toEqual([1])
+  })
+
+  it.each([
+    { first: 'handoff', content: 'original' },
+    { first: 'write', content: 'original' },
+    { first: 'handoff', content: '' },
+    { first: 'write', content: '' },
+  ] as const)(
+    'serializes full member revocation with a size-changing content write ($first first, original=$content)',
+    async ({ first, content }) => {
+      const fixture = await seedFile(content)
+      await db
+        .update(workspaceFiles)
+        .set({ userId: fixture.bobId })
+        .where(eq(workspaceFiles.id, fixture.fileId))
+      const [usageBefore] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      const operations = {
+        handoff: () =>
+          db.transaction((tx) =>
+            revokeWorkspaceAccessTx(tx, {
+              workspaceId: fixture.workspaceId,
+              userId: fixture.bobId,
+            }).then((result) => expect(result.revoked).toBe(true))
+          ),
+        write: () =>
+          updateWorkspaceFileContent(
+            fixture.workspaceId,
+            fixture.fileId,
+            fixture.aliceId,
+            Buffer.from('second'),
+            undefined,
+            { version: { source: 'api', authorUserId: fixture.aliceId } }
+          ),
+      }
+      await runContendedFileOperations(fixture.fileId, [
+        operations[first],
+        operations[first === 'handoff' ? 'write' : 'handoff'],
+      ])
+      const rows = await versionRows(fixture.fileId)
+      expect(rows.map((row) => [row.version, row.authorUserIds])).toEqual(
+        content
+          ? [
+              [1, [fixture.bobId]],
+              [2, [fixture.aliceId]],
+            ]
+          : [[1, [fixture.aliceId]]]
+      )
+      expect(rows.filter((row) => row.supersededAt === null)).toHaveLength(1)
+      expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, rows[0].key)).toBe(
+        content || 'second'
+      )
+      expect((await getWorkspaceFile(fixture.workspaceId, fixture.fileId))?.uploadedBy).toBe(
+        fixture.aliceId
+      )
+      const [usageAfter] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      expect(usageAfter.bytes).toBe(
+        usageBefore.bytes + Buffer.byteLength('second') - Buffer.byteLength(content)
+      )
+      expect(
+        await db
+          .select({ id: permissions.id })
+          .from(permissions)
+          .where(
+            and(
+              eq(permissions.entityId, fixture.workspaceId),
+              eq(permissions.entityType, 'workspace'),
+              eq(permissions.userId, fixture.bobId)
+            )
+          )
+      ).toEqual([])
+      const current = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+      if (!current) throw new Error('file missing')
+      expect((await fetchWorkspaceFileBuffer(current, { maxBytes: 1024 })).toString()).toBe(
+        'second'
+      )
+    }
+  )
+
+  it.each(['handoff', 'purge'] as const)(
+    'serializes member revocation with archive rollback (%s first)',
+    async (first) => {
+      const fixture = await seedFile('original')
+      await db
+        .update(workspaceFiles)
+        .set({ userId: fixture.bobId })
+        .where(eq(workspaceFiles.id, fixture.fileId))
+      const file = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+      if (!file) throw new Error('file missing')
+      const [usageBefore] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      const operations = {
+        handoff: () =>
+          db.transaction(async (tx) => {
+            const result = await revokeWorkspaceAccessTx(tx, {
+              workspaceId: fixture.workspaceId,
+              userId: fixture.bobId,
+            })
+            expect(result.revoked).toBe(true)
+          }),
+        purge: async () => {
+          const purged = await purgeCreatedWorkspaceFile({
+            workspaceId: fixture.workspaceId,
+            fileId: fixture.fileId,
+            key: file.key,
+            expectedName: file.name,
+            expectedFolderId: file.folderId ?? null,
+            expectedUpdatedAt: file.updatedAt,
+          })
+          expect(purged).toBe(first === 'purge')
+        },
+      }
+      await runContendedFileOperations(fixture.fileId, [
+        operations[first],
+        operations[first === 'handoff' ? 'purge' : 'handoff'],
+      ])
+      const retained = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+      const [usageAfter] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      expect(usageAfter.bytes).toBe(
+        usageBefore.bytes - (first === 'purge' ? Buffer.byteLength('original') : 0)
+      )
+      expect(await objectExists(fixture.firstKey)).toBe(first === 'handoff')
+      if (first === 'purge') {
+        expect(retained).toBeNull()
+        expect(await versionRows(fixture.fileId)).toEqual([])
+      } else {
+        expect(retained?.uploadedBy).toBe(fixture.aliceId)
+        expect((await versionRows(fixture.fileId)).map((row) => row.authorUserIds)).toEqual([
+          [fixture.bobId],
+        ])
+      }
+    }
+  )
+
+  it.each(['handoff', 'cleanup'] as const)(
+    'serializes member revocation with retention cleanup (%s first)',
+    async (first) => {
+      const fixture = await seedFile('original')
+      await db
+        .update(workspaceFiles)
+        .set({ userId: fixture.bobId, deletedAt: new Date(0) })
+        .where(eq(workspaceFiles.id, fixture.fileId))
+      const [usageBefore] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      const operations = {
+        handoff: () =>
+          db.transaction(async (tx) => {
+            const result = await revokeWorkspaceAccessTx(tx, {
+              workspaceId: fixture.workspaceId,
+              userId: fixture.bobId,
+            })
+            expect(result.revoked).toBe(true)
+          }),
+        cleanup: () =>
+          runCleanupSoftDeletes({
+            workspaceIds: [fixture.workspaceId],
+            plan: 'free',
+            retentionHours: 1,
+            label: 'file-handoff-lock-regression',
+          }),
+      }
+      await runContendedFileOperations(fixture.fileId, [
+        operations[first],
+        operations[first === 'handoff' ? 'cleanup' : 'handoff'],
+      ])
+      expect(
+        await db
+          .select({ id: workspaceFiles.id })
+          .from(workspaceFiles)
+          .where(eq(workspaceFiles.id, fixture.fileId))
+      ).toEqual([])
+      expect(await versionRows(fixture.fileId)).toEqual([])
+      const [usageAfter] = await db
+        .select({ bytes: workspace.storageUsedBytes })
+        .from(workspace)
+        .where(eq(workspace.id, fixture.workspaceId))
+      expect(usageAfter.bytes).toBe(usageBefore.bytes - Buffer.byteLength('original'))
+    }
+  )
+
   it.each([
     ['named', 'download'],
     ['anonymous', 'download'],
diff --git a/apps/sim/lib/uploads/contexts/workspace/creator-handoff.ts b/apps/sim/lib/uploads/contexts/workspace/creator-handoff.ts
new file mode 100644
index 00000000000..5725ac5ac32
--- /dev/null
+++ b/apps/sim/lib/uploads/contexts/workspace/creator-handoff.ts
@@ -0,0 +1,52 @@
+import { workspaceFiles } from '@sim/db/schema'
+import { asc, inArray, type SQL } from 'drizzle-orm'
+import type { DbTransaction } from '@/lib/db/types'
+import { snapshotWorkspaceFileSecretProvenanceInTx } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
+import {
+  isVersionHeadCurrent,
+  loadWorkspaceFileVersionHead,
+  materializeWorkspaceFileVersionInTx,
+} from '@/lib/uploads/contexts/workspace/workspace-file-versions'
+
+/** Freezes implicit history under the writer lock before replacing a shared file's live creator. */
+export async function handoffFileCreatorsInTx(
+  tx: DbTransaction,
+  condition: SQL | undefined,
+  successorId: string
+) {
+  if (!condition) throw new Error('File creator handoff requires a scope')
+  const files = await tx
+    .select()
+    .from(workspaceFiles)
+    .where(condition)
+    .orderBy(asc(workspaceFiles.id))
+    .for('update')
+  for (const file of files) {
+    if (file.context !== 'workspace' || !file.workspaceId) continue
+    const head = await loadWorkspaceFileVersionHead(file.id, tx)
+    if (isVersionHeadCurrent(head, file)) continue
+    const provenance = await snapshotWorkspaceFileSecretProvenanceInTx(
+      tx,
+      file.id,
+      file.contentUpdatedAt,
+      file.secretProvenanceVersion
+    )
+    await materializeWorkspaceFileVersionInTx(
+      tx,
+      { ...file, workspaceId: file.workspaceId },
+      head,
+      provenance,
+      new Date()
+    )
+  }
+  if (!files.length) return
+  await tx
+    .update(workspaceFiles)
+    .set({ userId: successorId, updatedAt: new Date() })
+    .where(
+      inArray(
+        workspaceFiles.id,
+        files.map((file) => file.id)
+      )
+    )
+}
diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts
index 57161c2058c..dfcb8e5f82e 100644
--- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts
+++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts
@@ -36,6 +36,7 @@ import {
 import {
   decrementStorageUsageForBillingContextInTx,
   incrementStorageUsageForBillingContextInTx,
+  lockWorkspaceStorageForMutationInTx,
   maybeNotifyStorageLimitForBillingContext,
   resolveStorageBillingContext,
 } from '@/lib/billing/storage'
@@ -2039,6 +2040,7 @@ export async function updateWorkspaceFileContent(
     }
     try {
       finalized = await db.transaction(async (tx) => {
+        await lockWorkspaceStorageForMutationInTx(tx, workspaceId)
         const [currentFile] = await tx
           .select()
           .from(workspaceFiles)
@@ -2427,6 +2429,7 @@ export async function purgeCreatedWorkspaceFile(params: {
     isNull(workspaceFiles.deletedAt)
   )
   const cleanupEventIds = await db.transaction(async (tx) => {
+    await lockWorkspaceStorageForMutationInTx(tx, params.workspaceId)
     const [lockedFile] = await tx
       .select({
         id: workspaceFiles.id,
diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-versions.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-versions.ts
index 6e8b0dbc743..24bf2959a3e 100644
--- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-versions.ts
+++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-versions.ts
@@ -190,13 +190,63 @@ interface RecordedWorkspaceFileVersion {
   releasedKeys: string[]
 }
 
+/** Materializes current bytes under the file lock before content or creator attribution changes. */
+export async function materializeWorkspaceFileVersionInTx(
+  tx: DbTransaction,
+  previous: WorkspaceFileRow & { workspaceId: string },
+  head: WorkspaceFileVersionSummaryRow | undefined,
+  provenance: WorkspaceFileSecretProvenanceSnapshot,
+  now: Date
+): Promise<WorkspaceFileVersionSummaryRow> {
+  if (isVersionHeadCurrent(head, previous) && head) return head
+  if (head && head.supersededAt === null) await supersedeVersionInTx(tx, head.id, now)
+  const original = !head && isOriginalUploadContent(previous)
+  const [materialized] = await tx
+    .insert(workspaceFileVersion)
+    .values({
+      id: generateId(),
+      fileId: previous.id,
+      workspaceId: previous.workspaceId,
+      version: head ? head.version + 1 : INITIAL_WORKSPACE_FILE_VERSION,
+      ...contentColumns(previous, provenance),
+      contentHash: null,
+      source: original ? 'upload' : 'unknown',
+      authorUserIds: original && previous.userId ? [previous.userId] : [],
+      createdAt: previous.contentUpdatedAt,
+      updatedAt: previous.contentUpdatedAt,
+    })
+    .returning(versionSummaryColumns)
+  return materialized
+}
+
+/** Identifies an implicit empty first version frozen by handoff, never an explicit write or restore. */
+function isMaterializedEmptyInitialVersion(
+  head: WorkspaceFileVersionSummaryRow | undefined,
+  file: WorkspaceFileRow
+): head is WorkspaceFileVersionSummaryRow {
+  return (
+    head !== undefined &&
+    head.fileId === file.id &&
+    head.version === INITIAL_WORKSPACE_FILE_VERSION &&
+    isVersionHeadCurrent(head, file) &&
+    getWorkspaceFileSize(file) === 0 &&
+    head.sizeBytes === 0 &&
+    head.contentHash === null &&
+    (head.source === 'upload' || head.source === 'unknown') &&
+    head.restoredFromVersion === null &&
+    head.createdAt.getTime() === file.contentUpdatedAt.getTime() &&
+    head.updatedAt.getTime() === file.contentUpdatedAt.getTime()
+  )
+}
+
 /**
  * Records a committed content write in the file's history. Runs inside the content-write
  * transaction, under the file row's lock, which serializes version numbering and coalescing
  * decisions per file.
  *
  * An empty file with no history is a shell whose content arrives in this write (a create followed
- * by its first content), so the shell is not kept as a version of its own.
+ * by its first content), so the shell is not kept as a version of its own. A handoff may have
+ * materialized that implicit shell to freeze attribution; its first write still replaces version 1.
  */
 export async function recordWorkspaceFileVersionInTx(
   tx: DbTransaction,
@@ -210,28 +260,40 @@ export async function recordWorkspaceFileVersionInTx(
     return { version: 1, releasedKeys: [previous.key] }
   }
 
+  if (isMaterializedEmptyInitialVersion(head, previous)) {
+    await tx
+      .update(workspaceFileVersion)
+      .set({
+        ...contentColumns(next, params.nextProvenance),
+        contentHash: params.contentHash,
+        source: write.source,
+        authorUserIds: write.authorUserId ? [write.authorUserId] : [],
+        restoredFromVersion: write.restoredFromVersion ?? null,
+        createdAt: now,
+        updatedAt: now,
+      })
+      .where(eq(workspaceFileVersion.id, head.id))
+    const [references] = await tx.execute<{ referenced: boolean }>(sql`
+      SELECT EXISTS(SELECT 1 FROM ${workspaceFiles} WHERE ${workspaceFiles.key} = ${previous.key})
+        OR EXISTS(SELECT 1 FROM ${workspaceFileVersion} WHERE ${workspaceFileVersion.key} = ${previous.key}) AS referenced
+    `)
+    return {
+      version: INITIAL_WORKSPACE_FILE_VERSION,
+      releasedKeys: references.referenced ? [] : [previous.key],
+    }
+  }
+
   if (!head || !isVersionHeadCurrent(head, previous)) {
     if (!params.previousProvenance) {
       throw new Error('Outgoing workspace file content needs a provenance snapshot to be versioned')
     }
-    if (head && head.supersededAt === null) await supersedeVersionInTx(tx, head.id, now)
-    const original = !head && isOriginalUploadContent(previous)
-    const [materialized] = await tx
-      .insert(workspaceFileVersion)
-      .values({
-        id: generateId(),
-        fileId: previous.id,
-        workspaceId: params.workspaceId,
-        version: head ? head.version + 1 : INITIAL_WORKSPACE_FILE_VERSION,
-        ...contentColumns(previous, params.previousProvenance),
-        contentHash: null,
-        source: original ? 'upload' : 'unknown',
-        authorUserIds: original ? [previous.userId] : [],
-        createdAt: previous.contentUpdatedAt,
-        updatedAt: previous.contentUpdatedAt,
-      })
-      .returning(versionSummaryColumns)
-    head = materialized
+    head = await materializeWorkspaceFileVersionInTx(
+      tx,
+      { ...previous, workspaceId: params.workspaceId },
+      head,
+      params.previousProvenance,
+      now
+    )
   }
 
   const nextColumns = { ...contentColumns(next, params.nextProvenance), updatedAt: now }
diff --git a/apps/sim/lib/users/account-deletion.ts b/apps/sim/lib/users/account-deletion.ts
index f748152778f..277a7a43679 100644
--- a/apps/sim/lib/users/account-deletion.ts
+++ b/apps/sim/lib/users/account-deletion.ts
@@ -1,5 +1,6 @@
 import { db } from '@sim/db'
 import {
+  apiKey,
   dataDrains,
   document,
   knowledgeBase,
@@ -9,6 +10,7 @@ import {
   tableRunDispatches,
   uploadSession,
   user,
+  workflowMcpServer,
   workspaceFile,
   workspaceFiles,
   workspaceFileVersion,
@@ -24,8 +26,16 @@ import type {
   AccountDeletionResource,
 } from '@/lib/api/contracts/user'
 import { getHighestPriorityPersonalSubscription } from '@/lib/billing/core/plan'
-import { isSoleOwnerOfPaidOrganization } from '@/lib/billing/organizations/membership'
+import {
+  acquireOrganizationMutationLock,
+  isSoleOwnerOfPaidOrganization,
+} from '@/lib/billing/organizations/membership'
+import {
+  listSharedResourceOrganizationIdsForUser,
+  reassignOrganizationSharedResourcesTx,
+} from '@/lib/billing/organizations/resource-handoff'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
+import type { DbOrTx } from '@/lib/db/types'
 import {
   getProjectAccountDeletionBlockers,
   prepareProjectsForAccountDeletion,
@@ -37,6 +47,10 @@ import {
 } from '@/lib/table/rows/executions'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
+import {
+  listSharedResourceWorkspaceIdsForUser,
+  reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx,
+} from '@/lib/workspaces/resource-handoff'
 import {
   reassignBilledAccountForUser,
   reassignOwnedWorkspacesForUser,
@@ -209,25 +223,10 @@ export interface AccountDeletionFacts {
 }
 
 /**
- * Turns the gathered facts into the full picture of an account deletion: what it
- * removes, what it hands off, and every reason it would be refused.
- *
- * The governing rule is that an account is erased only once it stands alone.
- * Nearly every table that points at `user.id` does so with `ON DELETE CASCADE`,
- * and those cascades do not distinguish a workflow in the account's own workspace
- * from a knowledge base it happened to create inside somebody else's — both would
- * go. Rather than chase that blast radius across every creator column (and
- * silently lose whichever one is added next), deletion refuses while the account
- * is still entangled and names the existing action that untangles it: leave the
- * workspace, leave the organization, cancel the plan. Each of those already hands
- * the account's content to a surviving member on its own well-tested path.
- *
- * What remains is provably private, so a workspace falls into exactly one bucket:
- *  - **delete** — nobody else can reach it, so it is erased with the account.
- *  - **transfer** — the account only pays for it or is recorded as its owner
- *    while holding no access to it, so moving that anchor to a real admin
- *    changes nothing anyone can see.
- *  - **blocked** — anything else.
+ * Classifies current memberships and workspace ownership/billing anchors. Shared memberships
+ * must be left before erasure; private workspaces are deleted, and orphaned anchors transfer
+ * to a surviving administrator. Execution dependencies are checked separately, and the final
+ * transaction repairs shared creator references from departures before this policy existed.
  */
 export function classifyAccountDeletion(facts: AccountDeletionFacts): AccountDeletionPlan {
   const blockers: AccountDeletionBlocker[] = []
@@ -307,6 +306,55 @@ function formatResourceNames(resources: AccountDeletionResource[]): string {
   return formatNames(resources.map((resource) => resource.name))
 }
 
+/** Public MCP actors and live shared keys cannot acquire a new identity through attribution handoff. */
+async function sharedExecutionDeletionBlockers(
+  executor: DbOrTx,
+  userId: string,
+  doomedWorkspaceIds: string[],
+  hold = false
+): Promise<AccountDeletionBlocker[]> {
+  const doomed = new Set(doomedWorkspaceIds)
+  const serverQuery = executor
+    .select({
+      id: workflowMcpServer.id,
+      name: workflowMcpServer.name,
+      workspaceId: workflowMcpServer.workspaceId,
+      isPublic: workflowMcpServer.isPublic,
+    })
+    .from(workflowMcpServer)
+    .where(eq(workflowMcpServer.createdBy, userId))
+  const servers = await (hold ? serverQuery.for('update') : serverQuery)
+  const keyQuery = executor
+    .select({
+      id: apiKey.id,
+      name: apiKey.name,
+      workspaceId: apiKey.workspaceId,
+      expiresAt: apiKey.expiresAt,
+    })
+    .from(apiKey)
+    .where(and(eq(apiKey.userId, userId), eq(apiKey.type, 'workspace')))
+  const keys = await (hold ? keyQuery.for('update') : keyQuery)
+  const retainedServers = servers.filter((row) => row.isPublic && !doomed.has(row.workspaceId))
+  const liveKeys = keys.filter(
+    (row) =>
+      row.workspaceId &&
+      !doomed.has(row.workspaceId) &&
+      (!row.expiresAt || row.expiresAt > new Date())
+  )
+  const blockers: AccountDeletionBlocker[] = []
+  if (retainedServers.length)
+    blockers.push({
+      code: 'shared_workspace',
+      message: `Public MCP servers ${formatResourceNames(retainedServers)} still use your account to execute. Ask a workspace admin to delete them or make them private before deleting your account. Restore an archived workspace first if needed.`,
+    })
+  if (liveKeys.length)
+    blockers.push({
+      code: 'shared_workspace',
+      message: `Workspace API keys ${formatResourceNames(liveKeys)} still depend on your account. Ask a workspace admin to replace and revoke them in Settings before deleting your account.`,
+    })
+  return blockers
+}
+
 /** Gathers the facts above and classifies them. */
 export async function getAccountDeletionPlan(userId: string): Promise<AccountDeletionPlan> {
   const [workspaces, organizationNames, paidOrgCheck, personalSubscription, drains] =
@@ -344,6 +392,13 @@ export async function getAccountDeletionPlan(userId: string): Promise<AccountDel
   plan.blockers.push(
     ...projectBlockers.map((message) => ({ code: 'project_lifecycle' as const, message }))
   )
+  plan.blockers.push(
+    ...(await sharedExecutionDeletionBlockers(
+      db,
+      userId,
+      plan.workspacesToDelete.map((row) => row.id)
+    ))
+  )
   return plan
 }
 
@@ -677,7 +732,7 @@ async function announceCancelledTableWork(
  * The ordering inside step 2 is load-bearing too: Postgres evaluates the
  * `NO ACTION` check on `workspace.billed_account_user_id` *before* the `owner_id`
  * cascade that would have removed the very same workspace, so a workspace the
- * account bills for must be handed over or gone before the `user` row is touched.
+ * account bills for must be handed over or gone before the `user` row is deleted.
  *
  * The plan is recomputed here rather than accepted from the caller: a preview is
  * a display, never an authorization.
@@ -694,6 +749,15 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
   let cancelledMarkers: CancelledCellMarker[] = []
 
   await db.transaction(async (tx) => {
+    /** Blocks new user FK references through retention discovery, run cancellation, and deletion. */
+    await tx.select({ id: user.id }).from(user).where(eq(user.id, userId)).for('update')
+    const executionBlockers = await sharedExecutionDeletionBlockers(
+      tx,
+      userId,
+      doomedWorkspaceIds,
+      true
+    )
+    if (executionBlockers.length) throw new AccountDeletionBlockedError(executionBlockers)
     await prepareProjectsForAccountDeletion(tx, userId, doomedWorkspaceIds)
     if (doomedWorkspaceIds.length > 0) {
       /**
@@ -753,28 +817,28 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
       ])
     }
 
-    /**
-     * Take the departing account's own row before cancelling anything it
-     * governs.
-     *
-     * Both cancels below are `WHERE capability_governed_user_id = userId`, so
-     * they only stop work that already exists. A dispatcher that read its
-     * status as active a moment earlier goes on to pre-stamp cells, and that
-     * insert's foreign key needs a `FOR KEY SHARE` on this very row — which
-     * `FOR UPDATE` conflicts with. Taking it first therefore splits every
-     * concurrent stamp cleanly in two: one that committed before us, which the
-     * marker cancel below then sees, and one that blocks until we commit and is
-     * refused by the (now absent) foreign key. Without the barrier a stamp
-     * landing between the marker cancel and the `user` delete is nulled by
-     * `ON DELETE SET NULL` and drained by a sibling worker as actorless — the
-     * ungated run this whole passage exists to prevent.
-     *
-     * The lock is held for the rest of the transaction, so a dispatcher holding
-     * a marker row we are about to cancel can deadlock with us; Postgres aborts
-     * one side and the deletion is retried by the person, which is the right
-     * trade against a silently ungated run.
-     */
-    await tx.select({ id: user.id }).from(user).where(eq(user.id, userId)).for('update')
+    const sharedOrganizationIds = await listSharedResourceOrganizationIdsForUser(tx, userId)
+    for (const organizationId of sharedOrganizationIds) {
+      await acquireOrganizationMutationLock(tx, organizationId)
+      await reassignOrganizationSharedResourcesTx(tx, organizationId, userId)
+    }
+
+    const sharedWorkspaceIds = await listSharedResourceWorkspaceIdsForUser(tx, userId)
+    const { unresolved: resourceUnresolved } =
+      await reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
+        tx,
+        workspaceIds: sharedWorkspaceIds,
+        departingUserId: userId,
+      })
+    if (resourceUnresolved.length > 0) {
+      throw new AccountDeletionBlockedError([
+        {
+          code: 'shared_workspace',
+          message:
+            'Shared resources still reference your account. Ask a workspace admin to assign an active billing account before deleting your account. Nothing was changed.',
+        },
+      ])
+    }
 
     /**
      * Cancel every table run this account still governs before the `user` row
diff --git a/apps/sim/lib/workspaces/access/shared-resource-handoff.integration.ts b/apps/sim/lib/workspaces/access/shared-resource-handoff.integration.ts
new file mode 100644
index 00000000000..dfd16c78219
--- /dev/null
+++ b/apps/sim/lib/workspaces/access/shared-resource-handoff.integration.ts
@@ -0,0 +1,937 @@
+import { db } from '@sim/db'
+import {
+  account,
+  apiKey,
+  chat,
+  copilotChats,
+  document,
+  environment,
+  folder,
+  knowledgeBase,
+  member,
+  organization,
+  permissionGroup,
+  permissions,
+  ssoProvider,
+  user,
+  userTableDefinitions,
+  userTableRows,
+  workflow,
+  workflowMcpServer,
+  workspace,
+  workspaceFile,
+  workspaceFiles,
+  workspaceFileVersion,
+} from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { createDeferred } from '@sim/testing/helpers/deferred'
+import { generateId } from '@sim/utils/id'
+import { and, eq, inArray, sql } from 'drizzle-orm'
+import { afterAll, describe, expect, it } from 'vitest'
+import { removeUserFromOrganization } from '@/lib/billing/organizations/membership'
+import { deleteUserAccount, getAccountDeletionPlan } from '@/lib/users/account-deletion'
+import { revokeWorkspaceAccessTx } from '@/lib/workspaces/access/workspace-access'
+
+readTestDatabaseUrl()
+
+const userIds: string[] = []
+const workspaceIds: string[] = []
+const organizationIds: string[] = []
+
+async function seedResources(archived: boolean, joined = true) {
+  const ownerId = generateId()
+  const departingId = generateId()
+  const workspaceId = generateId()
+  userIds.push(ownerId, departingId)
+  workspaceIds.push(workspaceId)
+  await db.insert(user).values(
+    [ownerId, departingId].map((id) => ({
+      id,
+      name: 'Lifecycle fixture',
+      email: `${id}@example.test`,
+      emailVerified: true,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+    }))
+  )
+  await db.insert(workspace).values({
+    id: workspaceId,
+    name: 'Lifecycle fixture',
+    ownerId,
+    billedAccountUserId: ownerId,
+    archivedAt: archived ? new Date() : null,
+  })
+  await db.insert(permissions).values(
+    (joined ? [ownerId, departingId] : [ownerId]).map((userId) => ({
+      id: generateId(),
+      userId,
+      entityType: 'workspace',
+      entityId: workspaceId,
+      permissionType: 'admin' as const,
+    }))
+  )
+  const workflowId = generateId()
+  const chatId = generateId()
+  const folderId = generateId()
+  const tableId = generateId()
+  const rowId = generateId()
+  const kbId = generateId()
+  const documentId = generateId()
+  const fileId = generateId()
+  const legacyFileId = generateId()
+  const versionId = generateId()
+  const archivedAt = archived ? new Date() : null
+  await db.insert(workflow).values({
+    id: workflowId,
+    userId: departingId,
+    workspaceId,
+    name: 'Retained workflow',
+    lastSynced: new Date(),
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    archivedAt,
+  })
+  await db.insert(chat).values({
+    id: chatId,
+    workflowId,
+    userId: departingId,
+    identifier: chatId,
+    title: 'Retained deployment',
+    archivedAt,
+  })
+  await db.insert(folder).values({
+    id: folderId,
+    resourceType: 'file',
+    name: 'Retained folder',
+    userId: departingId,
+    workspaceId,
+    deletedAt: archivedAt,
+  })
+  await db.insert(userTableDefinitions).values({
+    id: tableId,
+    workspaceId,
+    name: 'Retained table',
+    schema: { columns: [] },
+    createdBy: departingId,
+    archivedAt,
+  })
+  await db.insert(userTableRows).values({
+    id: rowId,
+    tableId,
+    workspaceId,
+    data: { retained: true },
+    createdBy: departingId,
+  })
+  await db.insert(knowledgeBase).values({
+    id: kbId,
+    workspaceId,
+    name: 'Retained knowledge',
+    userId: departingId,
+    deletedAt: archivedAt,
+  })
+  await db.insert(document).values({
+    id: documentId,
+    knowledgeBaseId: kbId,
+    filename: 'retained.txt',
+    fileUrl: 'data:text/plain,retained',
+    fileSize: 8,
+    mimeType: 'text/plain',
+    uploadedBy: departingId,
+  })
+  await db.insert(workspaceFiles).values({
+    id: fileId,
+    key: `workspace/${workspaceId}/${fileId}.txt`,
+    workspaceId,
+    folderId,
+    userId: departingId,
+    context: 'workspace',
+    originalName: 'retained.txt',
+    contentType: 'text/plain',
+    sizeBytes: 8,
+    deletedAt: archivedAt,
+  })
+  await db.insert(workspaceFileVersion).values({
+    id: versionId,
+    fileId,
+    workspaceId,
+    version: 1,
+    key: `workspace/${workspaceId}/${versionId}.txt`,
+    sizeBytes: 8,
+    contentType: 'text/plain',
+    source: 'upload',
+    authorUserIds: [departingId],
+  })
+  await db.insert(workspaceFile).values({
+    id: legacyFileId,
+    workspaceId,
+    uploadedBy: departingId,
+    name: 'legacy.txt',
+    key: `workspace/${workspaceId}/${legacyFileId}.txt`,
+    size: 8,
+    type: 'text/plain',
+    deletedAt: archivedAt,
+  })
+  return {
+    ownerId,
+    departingId,
+    workspaceId,
+    workflowId,
+    chatId,
+    folderId,
+    tableId,
+    rowId,
+    kbId,
+    documentId,
+    fileId,
+    legacyFileId,
+    versionId,
+    archivedAt,
+  }
+}
+
+async function assertTransferred(fixture: Awaited<ReturnType<typeof seedResources>>) {
+  const { ownerId } = fixture
+  expect
+    .soft(await db.select().from(workflow).where(eq(workflow.id, fixture.workflowId)))
+    .toMatchObject([{ userId: ownerId, archivedAt: fixture.archivedAt }])
+  expect
+    .soft(await db.select().from(chat).where(eq(chat.id, fixture.chatId)))
+    .toMatchObject([{ userId: ownerId, archivedAt: fixture.archivedAt }])
+  expect
+    .soft(await db.select().from(folder).where(eq(folder.id, fixture.folderId)))
+    .toMatchObject([{ userId: ownerId, deletedAt: fixture.archivedAt }])
+  expect
+    .soft(
+      await db
+        .select()
+        .from(userTableDefinitions)
+        .where(eq(userTableDefinitions.id, fixture.tableId))
+    )
+    .toMatchObject([{ createdBy: ownerId, archivedAt: fixture.archivedAt }])
+  expect
+    .soft(await db.select().from(knowledgeBase).where(eq(knowledgeBase.id, fixture.kbId)))
+    .toMatchObject([{ userId: ownerId, deletedAt: fixture.archivedAt }])
+  expect
+    .soft(await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, fixture.fileId)))
+    .toMatchObject([
+      { userId: ownerId, folderId: fixture.folderId, sizeBytes: 8, deletedAt: fixture.archivedAt },
+    ])
+  expect
+    .soft(await db.select().from(workspaceFile).where(eq(workspaceFile.id, fixture.legacyFileId)))
+    .toMatchObject([{ uploadedBy: ownerId, deletedAt: fixture.archivedAt }])
+}
+
+async function assertRetained(fixture: Awaited<ReturnType<typeof seedResources>>) {
+  const { departingId, workspaceId } = fixture
+  await assertTransferred(fixture)
+  expect.soft(await db.select().from(user).where(eq(user.id, departingId))).toHaveLength(0)
+  expect
+    .soft(await db.select().from(workspace).where(eq(workspace.id, workspaceId)))
+    .toHaveLength(1)
+  expect
+    .soft(await db.select().from(userTableRows).where(eq(userTableRows.id, fixture.rowId)))
+    .toMatchObject([{ data: { retained: true }, createdBy: null }])
+  expect
+    .soft(await db.select().from(document).where(eq(document.id, fixture.documentId)))
+    .toMatchObject([{ uploadedBy: null, fileSize: 8 }])
+  expect
+    .soft(
+      await db
+        .select()
+        .from(workspaceFileVersion)
+        .where(eq(workspaceFileVersion.id, fixture.versionId))
+    )
+    .toMatchObject([{ authorUserIds: [departingId], version: 1 }])
+  expect
+    .soft(
+      await db
+        .select()
+        .from(permissions)
+        .where(and(eq(permissions.entityId, workspaceId), eq(permissions.userId, departingId)))
+    )
+    .toHaveLength(0)
+}
+
+afterAll(async () => {
+  if (workspaceIds.length) await db.delete(workspace).where(inArray(workspace.id, workspaceIds))
+  if (organizationIds.length)
+    await db.delete(organization).where(inArray(organization.id, organizationIds))
+  if (userIds.length) await db.delete(user).where(inArray(user.id, userIds))
+})
+
+describe('shared resource retention across workspace departure and account erasure', () => {
+  it.each([false, true])(
+    'retains shared roots and their children after removal (archived=%s)',
+    async (archived) => {
+      const fixture = await seedResources(archived)
+      const result = await db.transaction((tx) =>
+        revokeWorkspaceAccessTx(tx, {
+          workspaceId: fixture.workspaceId,
+          userId: fixture.departingId,
+        })
+      )
+      expect(result.revoked).toBe(true)
+      await assertTransferred(fixture)
+      await deleteUserAccount(fixture.departingId)
+      await assertRetained(fixture)
+    }
+  )
+
+  it('repairs references left by departures before the handoff fix', async () => {
+    const fixture = await seedResources(true, false)
+    await deleteUserAccount(fixture.departingId)
+    await assertRetained(fixture)
+  })
+
+  it('retains a shared root committed while account deletion waits for its creator', async () => {
+    const fixture = await seedResources(false, false)
+    const tableId = generateId()
+    const ready = createDeferred<number>()
+    const release = createDeferred<void>()
+    const writer = db.transaction(async (tx) => {
+      await tx.insert(userTableDefinitions).values({
+        id: tableId,
+        workspaceId: fixture.workspaceId,
+        name: 'Concurrent table',
+        schema: { columns: [] },
+        createdBy: fixture.departingId,
+      })
+      const [connection] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+      ready.resolve(connection.pid)
+      await release.promise
+    })
+    const writerPid = await ready.promise
+    const deletion = deleteUserAccount(fixture.departingId)
+    try {
+      await expect
+        .poll(
+          async () => {
+            const rows = await db.execute(
+              sql`SELECT 1 FROM pg_stat_activity WHERE ${writerPid} = ANY(pg_blocking_pids(pid))`
+            )
+            return rows.length
+          },
+          { timeout: 5000 }
+        )
+        .toBeGreaterThan(0)
+    } finally {
+      release.resolve()
+      await writer
+      await deletion
+    }
+    expect(
+      await db.select().from(userTableDefinitions).where(eq(userTableDefinitions.id, tableId))
+    ).toMatchObject([{ createdBy: fixture.ownerId }])
+    await assertRetained(fixture)
+  })
+
+  it('rejects a late ownership reference after account deletion holds the user barrier', async () => {
+    const fixture = await seedResources(false, false)
+    const serverId = generateId()
+    await db.insert(workflowMcpServer).values({
+      id: serverId,
+      workspaceId: fixture.workspaceId,
+      createdBy: fixture.departingId,
+      name: 'Concurrent private endpoint',
+      isPublic: false,
+    })
+    const ready = createDeferred<number>()
+    const release = createDeferred<void>()
+    const blocker = db.transaction(async (tx) => {
+      await tx
+        .select()
+        .from(workflowMcpServer)
+        .where(eq(workflowMcpServer.id, serverId))
+        .for('update')
+      const [connection] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+      ready.resolve(connection.pid)
+      await release.promise
+    })
+    const blockerPid = await ready.promise
+    const deletion = deleteUserAccount(fixture.departingId)
+    let lateWrite: Promise<unknown> | undefined
+    try {
+      await expect
+        .poll(
+          async () => {
+            const rows = await db.execute(
+              sql`SELECT 1 FROM pg_stat_activity WHERE ${blockerPid} = ANY(pg_blocking_pids(pid))`
+            )
+            return rows.length
+          },
+          { timeout: 5000 }
+        )
+        .toBeGreaterThan(0)
+      const tableId = generateId()
+      lateWrite = db
+        .insert(userTableDefinitions)
+        .values({
+          id: tableId,
+          workspaceId: fixture.workspaceId,
+          name: 'Too late table',
+          schema: { columns: [] },
+          createdBy: fixture.departingId,
+        })
+        .then(
+          () => ({ inserted: true }),
+          (error: unknown) => ({ error })
+        )
+      await expect
+        .poll(
+          async () => {
+            const rows = await db.execute(sql`
+          SELECT 1 FROM pg_stat_activity blocked
+          WHERE EXISTS (SELECT 1 FROM pg_stat_activity deleting
+            WHERE ${blockerPid} = ANY(pg_blocking_pids(deleting.pid))
+            AND deleting.pid = ANY(pg_blocking_pids(blocked.pid)))
+        `)
+            return rows.length
+          },
+          { timeout: 5000 }
+        )
+        .toBeGreaterThan(0)
+    } finally {
+      release.resolve()
+      await blocker
+      await deletion
+    }
+    expect(await lateWrite).toMatchObject({
+      error: expect.objectContaining({ cause: expect.objectContaining({ code: '23503' }) }),
+    })
+    await assertRetained(fixture)
+  })
+
+  it.each(['payer', 'grant'] as const)(
+    'revalidates a concurrent successor %s change before revoking access',
+    async (kind) => {
+      const fixture = await seedResources(false)
+      const ready = createDeferred<number>()
+      const release = createDeferred<void>()
+      const change = db.transaction(async (tx) => {
+        if (kind === 'payer') {
+          await tx
+            .update(workspace)
+            .set({ billedAccountUserId: fixture.departingId })
+            .where(eq(workspace.id, fixture.workspaceId))
+        } else {
+          await tx
+            .delete(permissions)
+            .where(
+              and(
+                eq(permissions.entityId, fixture.workspaceId),
+                eq(permissions.userId, fixture.ownerId)
+              )
+            )
+        }
+        const [connection] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+        ready.resolve(connection.pid)
+        await release.promise
+      })
+      const changerPid = await ready.promise
+      const removal = db.transaction((tx) =>
+        revokeWorkspaceAccessTx(tx, {
+          workspaceId: fixture.workspaceId,
+          userId: fixture.departingId,
+        })
+      )
+      try {
+        await expect
+          .poll(
+            async () => {
+              const rows = await db.execute(
+                sql`SELECT 1 FROM pg_stat_activity WHERE ${changerPid} = ANY(pg_blocking_pids(pid))`
+              )
+              return rows.length
+            },
+            { timeout: 5000 }
+          )
+          .toBeGreaterThan(0)
+      } finally {
+        release.resolve()
+        await change
+      }
+      expect((await removal).revoked).toBe(false)
+      expect(
+        await db
+          .select()
+          .from(userTableDefinitions)
+          .where(eq(userTableDefinitions.id, fixture.tableId))
+      ).toMatchObject([{ createdBy: fixture.departingId }])
+      expect(
+        await db
+          .select()
+          .from(permissions)
+          .where(
+            and(
+              eq(permissions.entityId, fixture.workspaceId),
+              eq(permissions.userId, fixture.departingId)
+            )
+          )
+      ).toHaveLength(1)
+    }
+  )
+
+  it('keeps private attachments and personal credentials private during departure and erases them with the user', async () => {
+    const fixture = await seedResources(false)
+    const privateChatId = generateId()
+    const privateFileId = generateId()
+    const credentialId = generateId()
+    await db
+      .insert(copilotChats)
+      .values({ id: privateChatId, userId: fixture.departingId, workspaceId: fixture.workspaceId })
+    await db.insert(workspaceFiles).values({
+      id: privateFileId,
+      key: `mothership/${privateFileId}`,
+      userId: fixture.departingId,
+      workspaceId: fixture.workspaceId,
+      chatId: privateChatId,
+      context: 'mothership',
+      originalName: 'private.txt',
+      contentType: 'text/plain',
+      sizeBytes: 8,
+    })
+    await db.insert(account).values({
+      id: credentialId,
+      accountId: credentialId,
+      providerId: 'fixture',
+      userId: fixture.departingId,
+      accessToken: 'synthetic-private-token',
+      createdAt: new Date(),
+      updatedAt: new Date(),
+    })
+    await db.insert(environment).values({
+      id: fixture.departingId,
+      userId: fixture.departingId,
+      variables: { PRIVATE: 'synthetic' },
+    })
+    await db.transaction((tx) =>
+      revokeWorkspaceAccessTx(tx, { workspaceId: fixture.workspaceId, userId: fixture.departingId })
+    )
+    expect(
+      await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, privateFileId))
+    ).toMatchObject([{ userId: fixture.departingId, chatId: privateChatId }])
+    expect(await db.select().from(account).where(eq(account.id, credentialId))).toMatchObject([
+      { userId: fixture.departingId, accessToken: 'synthetic-private-token' },
+    ])
+    await deleteUserAccount(fixture.departingId)
+    expect(
+      await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, privateFileId))
+    ).toHaveLength(0)
+    expect(
+      await db.select().from(copilotChats).where(eq(copilotChats.id, privateChatId))
+    ).toHaveLength(0)
+    expect(await db.select().from(account).where(eq(account.id, credentialId))).toHaveLength(0)
+    expect(
+      await db.select().from(environment).where(eq(environment.id, fixture.departingId))
+    ).toHaveLength(0)
+    await assertRetained(fixture)
+  })
+
+  it('refuses organization-only retention without an owner successor', async () => {
+    const fixture = await seedResources(false, false)
+    const organizationId = generateId()
+    organizationIds.push(organizationId)
+    await db.insert(organization).values({
+      id: organizationId,
+      name: 'No successor',
+      slug: organizationId,
+      createdAt: new Date(),
+    })
+    const kbId = generateId()
+    await db.insert(knowledgeBase).values({
+      id: kbId,
+      name: 'Stranded knowledge',
+      userId: fixture.departingId,
+      organizationId,
+      isSearchIndex: true,
+    })
+    await expect(deleteUserAccount(fixture.departingId)).rejects.toThrow(
+      'active organization owner'
+    )
+    expect(await db.select().from(user).where(eq(user.id, fixture.departingId))).toHaveLength(1)
+    expect(await db.select().from(knowledgeBase).where(eq(knowledgeBase.id, kbId))).toMatchObject([
+      { userId: fixture.departingId },
+    ])
+  })
+
+  it('refuses an inaccessible successor without changing ownership or access', async () => {
+    const fixture = await seedResources(false)
+    await db
+      .delete(permissions)
+      .where(
+        and(eq(permissions.entityId, fixture.workspaceId), eq(permissions.userId, fixture.ownerId))
+      )
+    const result = await db.transaction((tx) =>
+      revokeWorkspaceAccessTx(tx, {
+        workspaceId: fixture.workspaceId,
+        userId: fixture.departingId,
+      })
+    )
+    expect(result.revoked).toBe(false)
+    expect(
+      await db
+        .select()
+        .from(userTableDefinitions)
+        .where(eq(userTableDefinitions.id, fixture.tableId))
+    ).toMatchObject([{ createdBy: fixture.departingId }])
+    expect(
+      await db
+        .select()
+        .from(permissions)
+        .where(
+          and(
+            eq(permissions.entityId, fixture.workspaceId),
+            eq(permissions.userId, fixture.departingId)
+          )
+        )
+    ).toHaveLength(1)
+    await db.delete(permissions).where(eq(permissions.userId, fixture.departingId))
+    await expect(deleteUserAccount(fixture.departingId)).rejects.toThrow('active billing account')
+    expect(await db.select().from(user).where(eq(user.id, fixture.departingId))).toHaveLength(1)
+  })
+
+  it('rolls back handoff and access revocation together when the transaction fails', async () => {
+    const fixture = await seedResources(false)
+    await expect(
+      db.transaction(async (tx) => {
+        expect(
+          (
+            await revokeWorkspaceAccessTx(tx, {
+              workspaceId: fixture.workspaceId,
+              userId: fixture.departingId,
+            })
+          ).revoked
+        ).toBe(true)
+        await tx.execute(sql`SELECT 1 / 0`)
+      })
+    ).rejects.toThrow()
+    expect(
+      await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, fixture.fileId))
+    ).toMatchObject([{ userId: fixture.departingId }])
+    expect(
+      await db
+        .select()
+        .from(permissions)
+        .where(
+          and(
+            eq(permissions.entityId, fixture.workspaceId),
+            eq(permissions.userId, fixture.departingId)
+          )
+        )
+    ).toHaveLength(1)
+  })
+
+  it.each([false, true])(
+    'retains public MCP execution identity and blocks erasure (deleted=%s)',
+    async (deleted) => {
+      const fixture = await seedResources(false)
+      const serverId = generateId()
+      await db.insert(workflowMcpServer).values({
+        id: serverId,
+        workspaceId: fixture.workspaceId,
+        createdBy: fixture.departingId,
+        name: 'Retained public endpoint',
+        isPublic: true,
+        deletedAt: deleted ? new Date() : null,
+      })
+      await db.transaction((tx) =>
+        revokeWorkspaceAccessTx(tx, {
+          workspaceId: fixture.workspaceId,
+          userId: fixture.departingId,
+        })
+      )
+      expect(
+        await db.select().from(workflowMcpServer).where(eq(workflowMcpServer.id, serverId))
+      ).toMatchObject([{ createdBy: fixture.departingId, isPublic: true }])
+      expect((await getAccountDeletionPlan(fixture.departingId)).blockers).toMatchObject([
+        { message: expect.stringContaining('Public MCP servers') },
+      ])
+      await expect(deleteUserAccount(fixture.departingId)).rejects.toThrow('Public MCP servers')
+      expect(await db.select().from(user).where(eq(user.id, fixture.departingId))).toHaveLength(1)
+      await db.delete(workflowMcpServer).where(eq(workflowMcpServer.id, serverId))
+      await deleteUserAccount(fixture.departingId)
+      await assertRetained(fixture)
+    }
+  )
+
+  it.each(['public server', 'workspace key'] as const)(
+    'rechecks a %s dependency activated after the deletion preview',
+    async (kind) => {
+      const fixture = await seedResources(false, false)
+      const dependencyId = generateId()
+      if (kind === 'public server') {
+        await db.insert(workflowMcpServer).values({
+          id: dependencyId,
+          workspaceId: fixture.workspaceId,
+          createdBy: fixture.departingId,
+          name: 'Concurrent endpoint',
+          isPublic: false,
+        })
+      } else {
+        await db.insert(apiKey).values({
+          id: dependencyId,
+          workspaceId: fixture.workspaceId,
+          userId: fixture.departingId,
+          name: 'Concurrent key',
+          key: generateId(),
+          type: 'workspace',
+          expiresAt: new Date(Date.now() - 60_000),
+        })
+      }
+      const ready = createDeferred<number>()
+      const activate = createDeferred<void>()
+      const change = db.transaction(async (tx) => {
+        await tx.select().from(user).where(eq(user.id, fixture.departingId)).for('update')
+        const [connection] = await tx.execute<{ pid: number }>(sql`SELECT pg_backend_pid() AS pid`)
+        ready.resolve(connection.pid)
+        await activate.promise
+        if (kind === 'public server') {
+          await tx
+            .update(workflowMcpServer)
+            .set({ isPublic: true })
+            .where(eq(workflowMcpServer.id, dependencyId))
+        } else {
+          await tx.update(apiKey).set({ expiresAt: null }).where(eq(apiKey.id, dependencyId))
+        }
+      })
+      const changerPid = await ready.promise
+      const deletion = deleteUserAccount(fixture.departingId).then(
+        () => ({ deleted: true }),
+        (error: unknown) => ({ error })
+      )
+      try {
+        await expect
+          .poll(
+            async () => {
+              const rows = await db.execute(
+                sql`SELECT 1 FROM pg_stat_activity WHERE ${changerPid} = ANY(pg_blocking_pids(pid))`
+              )
+              return rows.length
+            },
+            { timeout: 5000 }
+          )
+          .toBeGreaterThan(0)
+      } finally {
+        activate.resolve()
+        await change
+      }
+      expect(await deletion).toMatchObject({ error: expect.objectContaining({ code: 'conflict' }) })
+      expect(await db.select().from(user).where(eq(user.id, fixture.departingId))).toHaveLength(1)
+      if (kind === 'public server') {
+        expect(
+          await db.select().from(workflowMcpServer).where(eq(workflowMcpServer.id, dependencyId))
+        ).toMatchObject([{ createdBy: fixture.departingId, isPublic: true }])
+      } else {
+        expect(await db.select().from(apiKey).where(eq(apiKey.id, dependencyId))).toMatchObject([
+          { userId: fixture.departingId, expiresAt: null },
+        ])
+      }
+    }
+  )
+
+  it('transfers private MCP metadata without rewriting public server actors', async () => {
+    const fixture = await seedResources(false, false)
+    const serverId = generateId()
+    await db.insert(workflowMcpServer).values({
+      id: serverId,
+      workspaceId: fixture.workspaceId,
+      createdBy: fixture.departingId,
+      name: 'Retained authenticated endpoint',
+      isPublic: false,
+    })
+    await deleteUserAccount(fixture.departingId)
+    expect(
+      await db.select().from(workflowMcpServer).where(eq(workflowMcpServer.id, serverId))
+    ).toMatchObject([{ createdBy: fixture.ownerId, isPublic: false }])
+    await assertRetained(fixture)
+  })
+
+  it.each(['permanent', 'future', 'expired', 'revoked'] as const)(
+    'blocks only live workspace keys (%s)',
+    async (state) => {
+      const fixture = await seedResources(false, false)
+      const keyId = generateId()
+      await db.insert(apiKey).values({
+        id: keyId,
+        userId: fixture.departingId,
+        createdBy: fixture.departingId,
+        workspaceId: fixture.workspaceId,
+        type: 'workspace',
+        name: 'Shared key',
+        key: generateId(),
+        expiresAt:
+          state === 'permanent'
+            ? null
+            : new Date(Date.now() + (state === 'expired' ? -60_000 : 60_000)),
+      })
+      if (state === 'revoked') await db.delete(apiKey).where(eq(apiKey.id, keyId))
+      if (state === 'permanent' || state === 'future') {
+        await expect(deleteUserAccount(fixture.departingId)).rejects.toThrow('Workspace API keys')
+        expect(await db.select().from(apiKey).where(eq(apiKey.id, keyId))).toMatchObject([
+          { userId: fixture.departingId },
+        ])
+        await db.delete(apiKey).where(eq(apiKey.id, keyId))
+      }
+      await deleteUserAccount(fixture.departingId)
+      await assertRetained(fixture)
+    }
+  )
+
+  it.each([true, false])(
+    'retains organization resources and upload history with no workspace membership (currentMember=%s)',
+    async (currentMember) => {
+      const fixture = await seedResources(false, false)
+      const organizationId = generateId()
+      const memberId = generateId()
+      const kbId = generateId()
+      const orgFileId = generateId()
+      const implicitFileId = generateId()
+      const groupId = generateId()
+      const providerId = generateId()
+      organizationIds.push(organizationId)
+      await db.insert(organization).values({
+        id: organizationId,
+        name: 'Shared organization',
+        slug: organizationId,
+        createdAt: new Date(),
+      })
+      await db
+        .update(workspace)
+        .set({ organizationId })
+        .where(eq(workspace.id, fixture.workspaceId))
+      await db.insert(member).values({
+        id: generateId(),
+        organizationId,
+        userId: fixture.ownerId,
+        role: 'owner',
+        createdAt: new Date(),
+      })
+      if (currentMember)
+        await db.insert(member).values({
+          id: memberId,
+          organizationId,
+          userId: fixture.departingId,
+          role: 'member',
+          createdAt: new Date(),
+        })
+      await db.insert(knowledgeBase).values({
+        id: kbId,
+        organizationId,
+        userId: fixture.departingId,
+        isSearchIndex: true,
+        name: 'Organization index',
+        deletedAt: new Date(),
+      })
+      await db.insert(workspaceFiles).values({
+        id: orgFileId,
+        key: `knowledge-base/${organizationId}/${orgFileId}.txt`,
+        organizationId,
+        userId: fixture.departingId,
+        context: 'knowledge-base',
+        originalName: 'organization-index.txt',
+        contentType: 'text/plain',
+        sizeBytes: 8,
+        deletedAt: new Date(),
+      })
+
+      await db.insert(workspaceFiles).values({
+        id: implicitFileId,
+        key: `workspace/${fixture.workspaceId}/${implicitFileId}.txt`,
+        workspaceId: fixture.workspaceId,
+        userId: fixture.departingId,
+        context: 'workspace',
+        originalName: 'implicit-upload.txt',
+        contentType: 'text/plain',
+        sizeBytes: 8,
+      })
+      const assertUploadHistory = async () => {
+        expect(
+          await db
+            .select()
+            .from(workspaceFileVersion)
+            .where(eq(workspaceFileVersion.fileId, implicitFileId))
+        ).toMatchObject([{ version: 1, source: 'upload', authorUserIds: [fixture.departingId] }])
+        expect(
+          await db
+            .select({ userId: workspaceFiles.userId })
+            .from(workspaceFiles)
+            .where(eq(workspaceFiles.id, implicitFileId))
+        ).toEqual([{ userId: fixture.ownerId }])
+        expect(
+          await db
+            .select()
+            .from(workspaceFileVersion)
+            .where(eq(workspaceFileVersion.fileId, orgFileId))
+        ).toEqual([])
+      }
+
+      await db.insert(permissionGroup).values({
+        id: groupId,
+        organizationId,
+        name: 'Retained policy',
+        createdBy: fixture.departingId,
+        config: { tools: ['restricted'] },
+      })
+      await db.insert(ssoProvider).values({
+        id: providerId,
+        providerId,
+        organizationId,
+        userId: fixture.departingId,
+        domain: `${providerId}.example.test`,
+        issuer: `https://${providerId}.example.test`,
+      })
+      if (currentMember)
+        expect(
+          (
+            await removeUserFromOrganization({
+              organizationId,
+              userId: fixture.departingId,
+              memberId,
+              skipBillingLogic: true,
+              onError: 'throw',
+            })
+          ).success
+        ).toBe(true)
+      if (currentMember) {
+        await assertUploadHistory()
+        expect(
+          await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, orgFileId))
+        ).toMatchObject([
+          { userId: fixture.ownerId, organizationId, workspaceId: null, sizeBytes: 8 },
+        ])
+        expect(
+          await db.select().from(knowledgeBase).where(eq(knowledgeBase.id, kbId))
+        ).toMatchObject([{ userId: fixture.ownerId }])
+        expect(
+          await db.select().from(permissionGroup).where(eq(permissionGroup.id, groupId))
+        ).toMatchObject([{ createdBy: fixture.ownerId }])
+        expect(
+          await db.select().from(ssoProvider).where(eq(ssoProvider.id, providerId))
+        ).toMatchObject([{ userId: fixture.ownerId }])
+      }
+      await deleteUserAccount(fixture.departingId)
+      await assertUploadHistory()
+      expect(
+        await db.select().from(workspaceFiles).where(eq(workspaceFiles.id, orgFileId))
+      ).toMatchObject([
+        {
+          userId: fixture.ownerId,
+          organizationId,
+          workspaceId: null,
+          sizeBytes: 8,
+          deletedAt: expect.any(Date),
+        },
+      ])
+      expect(await db.select().from(knowledgeBase).where(eq(knowledgeBase.id, kbId))).toMatchObject(
+        [{ userId: fixture.ownerId }]
+      )
+      expect(
+        await db.select().from(permissionGroup).where(eq(permissionGroup.id, groupId))
+      ).toMatchObject([{ createdBy: fixture.ownerId, config: { tools: ['restricted'] } }])
+      expect(
+        await db.select().from(ssoProvider).where(eq(ssoProvider.id, providerId))
+      ).toMatchObject([{ userId: fixture.ownerId, providerId, organizationId }])
+      await assertRetained(fixture)
+    }
+  )
+})
diff --git a/apps/sim/lib/workspaces/access/workspace-access.ts b/apps/sim/lib/workspaces/access/workspace-access.ts
index deffbea554a..0c5eb78d10f 100644
--- a/apps/sim/lib/workspaces/access/workspace-access.ts
+++ b/apps/sim/lib/workspaces/access/workspace-access.ts
@@ -1,12 +1,12 @@
-import { permissions } from '@sim/db/schema'
+import { permissions, workspace } from '@sim/db/schema'
 import type { PermissionType } from '@sim/platform-authz/workspace'
 import { generateId } from '@sim/utils/id'
 import { and, eq } from 'drizzle-orm'
 import { revokeWorkspaceCredentialMembershipsTx } from '@/lib/credentials/access'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { removeWorkspaceSkillMembershipsTx } from '@/lib/skills/access'
+import { reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx } from '@/lib/workspaces/resource-handoff'
 import {
-  reassignWorkflowOwnershipForWorkspaceMemberRemovalTx,
   transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx,
   WorkspaceBillingAccountRemovalError,
 } from '@/lib/workspaces/utils'
@@ -117,39 +117,41 @@ export async function lowerWorkspaceAccessTx(
 export type RevokeWorkspaceAccessResult =
   /** `ownershipTransferred` is true when the departing user owned the workspace and it moved to the billed account. */
   | { revoked: true; ownershipTransferred: boolean }
-  /** Workflows whose owner could not be reassigned; the access row is left in place. */
+  /** Workspace IDs whose shared resources or ownership lack a successor; access remains unchanged. */
   | { revoked: false; reason: 'unresolved-workflows'; unresolvedWorkflows: string[] }
   /** The user owns the workspace and it has no billed account to hand it to. */
   | { revoked: false; reason: 'workspace-owner-without-successor' }
+  /** The requested membership no longer exists; a replacement grant must remain untouched. */
+  | { revoked: false; reason: 'membership-changed' }
 
-/**
- * Removes a user's access to one workspace and everything that hangs off it.
- *
- * Ownership moves first, in the same order the members route uses: the workspace
- * itself to its billed account when the departing user owns it, then every
- * workflow they own to a remaining member. Either can fail, and a failure is a
- * refusal rather than a partial removal — deleting the access row would orphan
- * what could not be moved.
- */
+/** Reassigns shared resources and workspace ownership before revoking access in the same transaction. */
 export async function revokeWorkspaceAccessTx(
-  tx: DbOrTx,
-  params: { workspaceId: string; userId: string }
+  tx: DbTransaction,
+  params: { workspaceId: string; userId: string; expectedPermissionId?: string }
 ): Promise<RevokeWorkspaceAccessResult> {
-  let ownershipTransferred: boolean
-  try {
-    ownershipTransferred = await transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx({
-      tx,
-      workspaceId: params.workspaceId,
-      departingUserId: params.userId,
-    })
-  } catch (error) {
-    if (error instanceof WorkspaceBillingAccountRemovalError) {
-      return { revoked: false, reason: 'workspace-owner-without-successor' }
-    }
-    throw error
+  if (params.expectedPermissionId !== undefined) {
+    /** Match the handoff's workspace-before-grant lock order before validating the selected row. */
+    await tx
+      .select({ id: workspace.id })
+      .from(workspace)
+      .where(eq(workspace.id, params.workspaceId))
+      .for('no key update')
+    const [selectedGrant] = await tx
+      .select({ id: permissions.id })
+      .from(permissions)
+      .where(
+        and(
+          eq(permissions.id, params.expectedPermissionId),
+          eq(permissions.userId, params.userId),
+          eq(permissions.entityType, 'workspace'),
+          eq(permissions.entityId, params.workspaceId)
+        )
+      )
+      .for('update')
+    if (!selectedGrant) return { revoked: false, reason: 'membership-changed' }
   }
 
-  const reassignment = await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
+  const reassignment = await reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
     tx,
     workspaceIds: [params.workspaceId],
     departingUserId: params.userId,
@@ -162,13 +164,30 @@ export async function revokeWorkspaceAccessTx(
     }
   }
 
+  let ownershipTransferred: boolean
+  try {
+    ownershipTransferred = await transferWorkspaceOwnershipToBilledAccountForMemberRemovalTx({
+      tx,
+      workspaceId: params.workspaceId,
+      departingUserId: params.userId,
+    })
+  } catch (error) {
+    if (error instanceof WorkspaceBillingAccountRemovalError) {
+      return { revoked: false, reason: 'workspace-owner-without-successor' }
+    }
+    throw error
+  }
+
   await tx
     .delete(permissions)
     .where(
       and(
         eq(permissions.userId, params.userId),
         eq(permissions.entityType, 'workspace'),
-        eq(permissions.entityId, params.workspaceId)
+        eq(permissions.entityId, params.workspaceId),
+        params.expectedPermissionId !== undefined
+          ? eq(permissions.id, params.expectedPermissionId)
+          : undefined
       )
     )
 
diff --git a/apps/sim/lib/workspaces/resource-handoff.ts b/apps/sim/lib/workspaces/resource-handoff.ts
new file mode 100644
index 00000000000..cadec67e460
--- /dev/null
+++ b/apps/sim/lib/workspaces/resource-handoff.ts
@@ -0,0 +1,190 @@
+import {
+  chat,
+  folder,
+  knowledgeBase,
+  member,
+  permissions,
+  user,
+  userTableDefinitions,
+  workflow,
+  workflowMcpServer,
+  workspace,
+  workspaceFile,
+  workspaceFiles,
+} from '@sim/db/schema'
+import { ORG_ADMIN_ROLES } from '@sim/platform-authz/workspace'
+import { and, asc, eq, inArray, isNull, sql } from 'drizzle-orm'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
+import { handoffFileCreatorsInTx } from '@/lib/uploads/contexts/workspace/creator-handoff'
+import { reassignWorkflowOwnershipForWorkspaceMemberRemovalTx } from '@/lib/workspaces/utils'
+
+/** Private chat attachments keep their user's lifecycle even when stored under a workspace key. */
+const SHARED_FILE_CONTEXTS = ['workspace', 'knowledge-base', 'execution', 'chat']
+
+/** Finds surviving containers from ownership references, including departures predating handoff. */
+export async function listSharedResourceWorkspaceIdsForUser(executor: DbOrTx, userId: string) {
+  const rows = await executor.execute<{ workspaceId: string }>(sql`
+    SELECT DISTINCT "workspaceId" FROM (
+      SELECT ${workflow.workspaceId} AS "workspaceId" FROM ${workflow} WHERE ${workflow.userId} = ${userId}
+      UNION ALL SELECT ${folder.workspaceId} FROM ${folder} WHERE ${folder.userId} = ${userId}
+      UNION ALL SELECT ${userTableDefinitions.workspaceId} FROM ${userTableDefinitions} WHERE ${userTableDefinitions.createdBy} = ${userId}
+      UNION ALL SELECT ${knowledgeBase.workspaceId} FROM ${knowledgeBase} WHERE ${knowledgeBase.userId} = ${userId}
+      UNION ALL SELECT ${workspaceFiles.workspaceId} FROM ${workspaceFiles}
+        WHERE ${workspaceFiles.userId} = ${userId} AND ${workspaceFiles.context} IN ('workspace', 'knowledge-base', 'execution', 'chat') AND ${workspaceFiles.chatId} IS NULL
+      UNION ALL SELECT ${workspaceFile.workspaceId} FROM ${workspaceFile} WHERE ${workspaceFile.uploadedBy} = ${userId}
+      UNION ALL SELECT ${workflowMcpServer.workspaceId} FROM ${workflowMcpServer}
+        WHERE ${workflowMcpServer.createdBy} = ${userId} AND NOT ${workflowMcpServer.isPublic}
+      UNION ALL SELECT ${workflow.workspaceId} FROM ${chat} JOIN ${workflow} ON ${workflow.id} = ${chat.workflowId} WHERE ${chat.userId} = ${userId}
+    ) owned WHERE "workspaceId" IS NOT NULL
+    ORDER BY "workspaceId"
+  `)
+  return rows.map((row) => row.workspaceId)
+}
+
+/** Holds a real successor and their current grant through the handoff transaction. */
+async function holdWorkspaceSuccessor(
+  tx: DbTransaction,
+  row: { id: string; organizationId: string | null; billedAccountUserId: string },
+  departingUserId: string
+): Promise<boolean> {
+  const successor = row.billedAccountUserId
+  if (!successor || successor === departingUserId) return false
+  const [account] = await tx
+    .select({ id: user.id })
+    .from(user)
+    .where(eq(user.id, successor))
+    .for('key share')
+  if (!account) return false
+  const [grant] = await tx
+    .select({ id: permissions.id })
+    .from(permissions)
+    .where(
+      and(
+        eq(permissions.entityType, 'workspace'),
+        eq(permissions.entityId, row.id),
+        eq(permissions.userId, successor)
+      )
+    )
+    .for('share')
+  if (grant) return true
+  if (!row.organizationId) return false
+  const [membership] = await tx
+    .select({ id: member.id })
+    .from(member)
+    .where(
+      and(
+        eq(member.organizationId, row.organizationId),
+        eq(member.userId, successor),
+        inArray(member.role, ORG_ADMIN_ROLES)
+      )
+    )
+    .for('share')
+  return Boolean(membership)
+}
+
+/**
+ * Moves shared lifecycle references before access is revoked. No active-row filter: archived
+ * resources and their children must survive the eventual user cascade too. Successor validation
+ * finishes before any write, so a refusal cannot leave a partially transferred batch.
+ */
+export async function reassignSharedResourceOwnershipForWorkspaceMemberRemovalTx({
+  tx,
+  workspaceIds,
+  departingUserId,
+}: {
+  tx: DbTransaction
+  workspaceIds: string[]
+  departingUserId: string
+}): Promise<{ unresolved: string[] }> {
+  const ids = [...new Set(workspaceIds)].sort()
+  if (!ids.length) return { unresolved: [] }
+  const rows = await tx
+    .select({
+      id: workspace.id,
+      ownerId: workspace.ownerId,
+      organizationId: workspace.organizationId,
+      billedAccountUserId: workspace.billedAccountUserId,
+    })
+    .from(workspace)
+    .where(inArray(workspace.id, ids))
+    .orderBy(asc(workspace.id))
+    .for('no key update')
+
+  const owned = new Set(await listSharedResourceWorkspaceIdsForUser(tx, departingUserId))
+  const unresolved: string[] = []
+  for (const row of rows) {
+    if (
+      (owned.has(row.id) || row.ownerId === departingUserId) &&
+      !(await holdWorkspaceSuccessor(tx, row, departingUserId))
+    )
+      unresolved.push(row.id)
+  }
+  if (unresolved.length) return { unresolved }
+
+  for (const row of rows) {
+    if (!owned.has(row.id)) continue
+    const userId = row.billedAccountUserId
+    const updatedAt = new Date()
+    await reassignWorkflowOwnershipForWorkspaceMemberRemovalTx({
+      tx,
+      workspaceIds: [row.id],
+      departingUserId,
+    })
+    await tx
+      .update(folder)
+      .set({ userId, updatedAt })
+      .where(and(eq(folder.workspaceId, row.id), eq(folder.userId, departingUserId)))
+    await tx
+      .update(userTableDefinitions)
+      .set({ createdBy: userId, updatedAt })
+      .where(
+        and(
+          eq(userTableDefinitions.workspaceId, row.id),
+          eq(userTableDefinitions.createdBy, departingUserId)
+        )
+      )
+    await tx
+      .update(knowledgeBase)
+      .set({ userId, updatedAt })
+      .where(and(eq(knowledgeBase.workspaceId, row.id), eq(knowledgeBase.userId, departingUserId)))
+    await handoffFileCreatorsInTx(
+      tx,
+      and(
+        eq(workspaceFiles.workspaceId, row.id),
+        eq(workspaceFiles.userId, departingUserId),
+        inArray(workspaceFiles.context, SHARED_FILE_CONTEXTS),
+        isNull(workspaceFiles.chatId)
+      ),
+      userId
+    )
+    await tx
+      .update(workspaceFile)
+      .set({ uploadedBy: userId })
+      .where(
+        and(eq(workspaceFile.workspaceId, row.id), eq(workspaceFile.uploadedBy, departingUserId))
+      )
+    await tx
+      .update(chat)
+      .set({ userId, updatedAt })
+      .where(
+        and(
+          eq(chat.userId, departingUserId),
+          inArray(
+            chat.workflowId,
+            tx.select({ id: workflow.id }).from(workflow).where(eq(workflow.workspaceId, row.id))
+          )
+        )
+      )
+    await tx
+      .update(workflowMcpServer)
+      .set({ createdBy: userId, updatedAt })
+      .where(
+        and(
+          eq(workflowMcpServer.workspaceId, row.id),
+          eq(workflowMcpServer.createdBy, departingUserId),
+          eq(workflowMcpServer.isPublic, false)
+        )
+      )
+  }
+  return { unresolved: [] }
+}
diff --git a/apps/sim/scripts/test-shared-resource-lifecycle-e2e.ts b/apps/sim/scripts/test-shared-resource-lifecycle-e2e.ts
new file mode 100644
index 00000000000..ab21a2f9c80
--- /dev/null
+++ b/apps/sim/scripts/test-shared-resource-lifecycle-e2e.ts
@@ -0,0 +1,495 @@
+import assert from 'node:assert/strict'
+import { execFileSync } from 'node:child_process'
+import { readFile, writeFile } from 'node:fs/promises'
+import { assertDisposableTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { createLogger } from '@sim/logger'
+import { sha256Hex } from '@sim/security/hash'
+import { getErrorMessage } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
+import { generateId, generateShortId } from '@sim/utils/id'
+import { isRecordLike } from '@sim/utils/object'
+import { truncate } from '@sim/utils/string'
+import { makeSignature } from 'better-auth/crypto'
+import postgres from 'postgres'
+import { getKnowledgeBaseContract } from '@/lib/api/contracts/knowledge'
+import { getTableContract } from '@/lib/api/contracts/tables'
+import { v2ListFileVersionsContract } from '@/lib/api/contracts/v2/file-versions'
+import { readWorkspaceFileContract } from '@/lib/api/contracts/workspace-files'
+
+/** Real HTTP lifecycle exercise; SQL seeds identities, access fixtures, and a provider-independent document. */
+const logger = createLogger('SharedResourceLifecycleE2E')
+function required(name: string): string {
+  const value = process.env[name]
+  assert(value, `${name} must be explicitly supplied`)
+  return value
+}
+const baseUrl = new URL(required('LIFECYCLE_E2E_BASE_URL'))
+assert(['localhost', '127.0.0.1', '[::1]'].includes(baseUrl.hostname))
+assert.equal(baseUrl.protocol, 'http:')
+const databaseUrl = assertDisposableTestDatabaseUrl(required('LIFECYCLE_E2E_DATABASE_URL'))
+const authSecret = required('LIFECYCLE_E2E_AUTH_SECRET')
+const adminKey = required('LIFECYCLE_E2E_ADMIN_KEY')
+const reportPath = required('LIFECYCLE_E2E_REPORT_PATH')
+const sql = postgres(databaseUrl.toString(), { max: 2 })
+const ownerKey = `sk-sim-fixture-${generateId()}`
+const ownerId = generateId()
+const departingId = generateId()
+const workspaceId = generateId()
+const cookies = new Map<string, string>()
+const checks: { name: string; status: string; durationMs: number; error?: string }[] = []
+const requests: { method: string; path: string; status: number }[] = []
+const startedAt = new Date().toISOString()
+let fileId = ''
+let implicitFileId = ''
+let folderId = ''
+let tableId = ''
+let kbId = ''
+const documentId = generateId()
+
+function record(value: unknown): Record<string, unknown> {
+  assert(isRecordLike(value), 'Expected an object')
+  return value
+}
+function id(value: unknown): string {
+  assert.equal(typeof value, 'string')
+  return value as string
+}
+async function request(
+  userId: string,
+  path: string,
+  method = 'GET',
+  body?: unknown,
+  expected = 200
+) {
+  const response = await fetch(new URL(path, baseUrl), {
+    method,
+    headers: {
+      Cookie: `better-auth.session_token=${cookies.get(userId) ?? ''}`,
+      Origin: baseUrl.origin,
+      'Content-Type': 'application/json',
+      ...(path.startsWith('/api/v2/') ? { 'X-API-Key': ownerKey } : {}),
+      ...(path.startsWith('/api/v1/admin/') ? { 'x-admin-key': adminKey } : {}),
+    },
+    body: body === undefined ? undefined : JSON.stringify(body),
+    signal: AbortSignal.timeout(120_000),
+    redirect: 'error',
+  })
+  requests.push({ method, path, status: response.status })
+  const text = await response.text()
+  assert.equal(
+    response.status,
+    expected,
+    `${method} ${path}: ${response.status} ${truncate(text, 500)}`
+  )
+  return text ? record(JSON.parse(text)) : {}
+}
+async function readVersionBytes(version: number, expected: string, targetFileId = fileId) {
+  const path = `/api/v2/files/${targetFileId}/versions/${version}/content?workspaceId=${workspaceId}`
+  const response = await fetch(new URL(path, baseUrl), {
+    headers: { 'X-API-Key': ownerKey },
+    signal: AbortSignal.timeout(120_000),
+  })
+  requests.push({ method: 'GET', path, status: response.status })
+  assert.equal(response.status, 200, await response.clone().text())
+  assert.equal(await response.text(), expected)
+}
+
+async function check(name: string, run: () => Promise<void>) {
+  const start = performance.now()
+  try {
+    await run()
+    checks.push({ name, status: 'passed', durationMs: performance.now() - start })
+    logger.info(name)
+  } catch (error) {
+    checks.push({
+      name,
+      status: 'failed',
+      durationMs: performance.now() - start,
+      error: getErrorMessage(error),
+    })
+    throw error
+  }
+}
+
+try {
+  for (const userId of [ownerId, departingId]) {
+    const now = new Date()
+    const token = generateShortId()
+    await sql`INSERT INTO "user" (id, name, email, normalized_email, email_verified, created_at, updated_at)
+      VALUES (${userId}, 'Lifecycle E2E', ${`${userId}@example.test`}, ${`${userId}@example.test`}, true, ${now}, ${now})`
+    await sql`INSERT INTO user_stats (id, user_id) VALUES (${generateId()}, ${userId})`
+    await sql`INSERT INTO session (id, token, user_id, expires_at, created_at, updated_at)
+      VALUES (${generateId()}, ${token}, ${userId}, ${new Date(Date.now() + 86_400_000)}, ${now}, ${now})`
+    cookies.set(userId, encodeURIComponent(`${token}.${await makeSignature(token, authSecret)}`))
+  }
+  await sql`INSERT INTO api_key (id, user_id, name, key, key_hash, type) VALUES (${generateId()}, ${ownerId}, 'Lifecycle fixture', ${ownerKey}, ${sha256Hex(ownerKey)}, 'personal')`
+  await sql`INSERT INTO workspace (id, name, owner_id, billed_account_user_id)
+    VALUES (${workspaceId}, 'Lifecycle HTTP fixture', ${ownerId}, ${ownerId})`
+  for (const userId of [ownerId, departingId]) {
+    await sql`INSERT INTO permissions (id, user_id, entity_type, entity_id, permission_type)
+      VALUES (${generateId()}, ${userId}, 'workspace', ${workspaceId}, 'admin')`
+  }
+  await check('Create shared roots and children through HTTP', async () => {
+    folderId = id(
+      record(
+        (
+          await request(departingId, `/api/workspaces/${workspaceId}/files/folders`, 'POST', {
+            name: 'Retained folder',
+          })
+        ).folder
+      ).id
+    )
+    tableId = id(
+      record(
+        record(
+          (
+            await request(departingId, '/api/table', 'POST', {
+              workspaceId,
+              name: 'Retained_table',
+              schema: { columns: [{ name: 'value', type: 'string' }] },
+            })
+          ).data
+        ).table
+      ).id
+    )
+    await request(departingId, `/api/table/${tableId}/rows`, 'POST', {
+      workspaceId,
+      data: { value: 'survives' },
+    })
+    kbId = id(
+      record(
+        (
+          await request(departingId, '/api/knowledge', 'POST', {
+            workspaceId,
+            name: 'Retained knowledge',
+          })
+        ).data
+      ).id
+    )
+    await sql`INSERT INTO document (id, knowledge_base_id, filename, file_url, file_size, mime_type, uploaded_by)
+      VALUES (${documentId}, ${kbId}, 'fixture.txt', 'data:text/plain,survives', 8, 'text/plain', ${departingId})`
+    fileId = id(
+      record(
+        (
+          await request(
+            departingId,
+            `/api/workspaces/${workspaceId}/files`,
+            'POST',
+            { name: 'retained.txt', folderId, content: 'version one' },
+            201
+          )
+        ).file
+      ).id
+    )
+    await request(departingId, `/api/workspaces/${workspaceId}/files/${fileId}/content`, 'PUT', {
+      content: 'version two',
+    })
+  })
+  implicitFileId = id(
+    record(
+      (
+        await request(
+          departingId,
+          `/api/workspaces/${workspaceId}/files`,
+          'POST',
+          {
+            name: 'implicit-upload.txt',
+            folderId,
+            content: 'original upload',
+          },
+          201
+        )
+      ).file
+    ).id
+  )
+  const [implicitBefore] =
+    await sql`SELECT key, size_bytes, content_updated_at, uploaded_at, secret_provenance_version FROM workspace_files WHERE id = ${implicitFileId}`
+  assert.equal(
+    (await sql`SELECT id FROM workspace_file_version WHERE file_id = ${implicitFileId}`).length,
+    0
+  )
+  const beforeStorage =
+    await sql`SELECT billed_account_user_id, storage_used_bytes FROM workspace WHERE id = ${workspaceId}`
+  const versionsBefore =
+    await sql`SELECT id, author_user_ids FROM workspace_file_version WHERE file_id = ${fileId} ORDER BY version`
+  await check(
+    'Both admin removal endpoints reject an owner transfer to a payer without existing access',
+    async () => {
+      await sql`UPDATE workspace SET owner_id = ${departingId} WHERE id = ${workspaceId}`
+      await sql`DELETE FROM permissions WHERE entity_id = ${workspaceId} AND user_id = ${ownerId}`
+      const [grant] =
+        await sql`SELECT id FROM permissions WHERE entity_id = ${workspaceId} AND user_id = ${departingId}`
+      for (const path of [
+        `/api/v1/admin/workspaces/${workspaceId}/members?userId=${departingId}`,
+        `/api/v1/admin/workspaces/${workspaceId}/members/${grant.id}`,
+      ]) {
+        await request(ownerId, path, 'DELETE', undefined, 400)
+        assert.equal(
+          (await sql`SELECT owner_id FROM workspace WHERE id = ${workspaceId}`)[0].owner_id,
+          departingId
+        )
+        assert.equal(
+          (
+            await sql`SELECT id FROM permissions WHERE entity_id = ${workspaceId} AND user_id = ${ownerId}`
+          ).length,
+          0
+        )
+        assert.equal(
+          (await sql`SELECT user_id FROM workspace_files WHERE id = ${fileId}`)[0].user_id,
+          departingId
+        )
+        assert.equal((await sql`SELECT id FROM permissions WHERE id = ${grant.id}`).length, 1)
+      }
+      await sql`UPDATE workspace SET owner_id = ${ownerId} WHERE id = ${workspaceId}`
+      await sql`INSERT INTO permissions (id, user_id, entity_type, entity_id, permission_type) VALUES (${generateId()}, ${ownerId}, 'workspace', ${workspaceId}, 'admin')`
+    }
+  )
+  await check(
+    'A stale membership-ID removal preserves a replacement grant and creator references',
+    async () => {
+      const [original] =
+        await sql`SELECT id FROM permissions WHERE entity_id = ${workspaceId} AND user_id = ${departingId}`
+      const replacementId = generateId()
+      let pending: Promise<{ error: unknown } | null> | undefined
+      try {
+        await sql.begin(async (tx) => {
+          const [connection] = await tx`SELECT pg_backend_pid() AS pid`
+          await tx`SELECT id FROM workspace WHERE id = ${workspaceId} FOR NO KEY UPDATE`
+          pending = request(
+            ownerId,
+            `/api/v1/admin/workspaces/${workspaceId}/members/${original.id}`,
+            'DELETE',
+            undefined,
+            404
+          ).then(
+            () => null,
+            (error: unknown) => ({ error })
+          )
+          const deadline = Date.now() + 10_000
+          let waiting = false
+          while (Date.now() < deadline) {
+            const blocked = await tx`SELECT pid FROM pg_stat_activity
+            WHERE datname = current_database() AND ${connection.pid} = ANY(pg_blocking_pids(pid))`
+            if (blocked.length) {
+              waiting = true
+              break
+            }
+            await sleep(25)
+          }
+          assert(waiting, 'Removal must reach the workspace lock after reading the original grant')
+          await tx`DELETE FROM permissions WHERE id = ${original.id}`
+          await tx`INSERT INTO permissions (id, user_id, entity_type, entity_id, permission_type)
+          VALUES (${replacementId}, ${departingId}, 'workspace', ${workspaceId}, 'admin')`
+        })
+        const outcome = await pending
+        assert.deepEqual(
+          {
+            grants: Array.from(
+              await sql`SELECT id FROM permissions WHERE entity_id = ${workspaceId} AND user_id = ${departingId}`,
+              (row) => ({ id: row.id })
+            ),
+            creator: (await sql`SELECT user_id FROM workspace_files WHERE id = ${fileId}`)[0]
+              .user_id,
+          },
+          { grants: [{ id: replacementId }], creator: departingId }
+        )
+        if (outcome) throw outcome.error
+        await request(departingId, `/api/workspaces/${workspaceId}/files/${fileId}`)
+      } finally {
+        await pending
+      }
+    }
+  )
+  await check('Remove member atomically and deny subsequent read and create requests', async () => {
+    await request(ownerId, `/api/workspaces/members/${departingId}`, 'DELETE', { workspaceId })
+    assert.equal(
+      (await sql`SELECT created_by FROM user_table_definitions WHERE id = ${tableId}`)[0]
+        .created_by,
+      ownerId
+    )
+    assert.equal(
+      (await sql`SELECT user_id FROM knowledge_base WHERE id = ${kbId}`)[0].user_id,
+      ownerId
+    )
+    assert.equal(
+      (await sql`SELECT user_id FROM workspace_files WHERE id = ${fileId}`)[0].user_id,
+      ownerId
+    )
+    for (const path of [
+      `/api/table/${tableId}?workspaceId=${workspaceId}`,
+      `/api/knowledge/${kbId}`,
+      `/api/workspaces/${workspaceId}/files/${fileId}`,
+    ]) {
+      await request(departingId, path, 'GET', undefined, 404)
+    }
+    await request(
+      departingId,
+      '/api/table',
+      'POST',
+      { workspaceId, name: 'Denied', schema: { columns: [{ name: 'value', type: 'string' }] } },
+      403
+    )
+  })
+  await check(
+    'Membership removal preserves implicit upload authorship before and after a survivor write',
+    async () => {
+      const path = `/api/v2/files/${implicitFileId}/versions?workspaceId=${workspaceId}&sortOrder=asc`
+      const history = v2ListFileVersionsContract.response.schema.parse(await request(ownerId, path))
+      assert.deepEqual(
+        history.data.map((version) => [
+          version.version,
+          version.source,
+          version.authors.map((author) => author.id),
+        ]),
+        [[1, 'upload', [departingId]]]
+      )
+      const [implicitAfter] =
+        await sql`SELECT key, size_bytes, content_updated_at, uploaded_at, secret_provenance_version FROM workspace_files WHERE id = ${implicitFileId}`
+      assert.deepEqual(implicitAfter, implicitBefore)
+      assert.equal(
+        (await sql`SELECT user_id FROM workspace_files WHERE id = ${implicitFileId}`)[0].user_id,
+        ownerId
+      )
+      assert.deepEqual(
+        await sql`SELECT billed_account_user_id, storage_used_bytes FROM workspace WHERE id = ${workspaceId}`,
+        beforeStorage
+      )
+      await readVersionBytes(1, 'original upload', implicitFileId)
+      await request(
+        ownerId,
+        `/api/workspaces/${workspaceId}/files/${implicitFileId}/content`,
+        'PUT',
+        { content: 'survivor writes' }
+      )
+      const afterWrite = v2ListFileVersionsContract.response.schema.parse(
+        await request(ownerId, path)
+      )
+      assert.deepEqual(
+        afterWrite.data.map((version) => [
+          version.version,
+          version.authors.map((author) => author.id),
+        ]),
+        [
+          [1, [departingId]],
+          [2, [ownerId]],
+        ]
+      )
+      await readVersionBytes(1, 'original upload', implicitFileId)
+      await readVersionBytes(2, 'survivor writes', implicitFileId)
+    }
+  )
+  await check('Delete departed account through authenticated HTTP', async () => {
+    await request(departingId, '/api/users/me/deletion', 'POST', {
+      confirmEmail: `${departingId}@example.test`,
+    })
+    assert.equal((await sql`SELECT id FROM "user" WHERE id = ${departingId}`).length, 0)
+  })
+  await check('Survivor reads existing non-null contracts and edits all shared roots', async () => {
+    const table = getTableContract.response.schema.parse(
+      await request(ownerId, `/api/table/${tableId}?workspaceId=${workspaceId}`)
+    )
+    assert.equal(table.data.table.createdBy, ownerId)
+    const kb = getKnowledgeBaseContract.response.schema.parse(
+      await request(ownerId, `/api/knowledge/${kbId}`)
+    )
+    assert.equal(kb.data.userId, ownerId)
+    const file = readWorkspaceFileContract.response.schema.parse(
+      await request(ownerId, `/api/workspaces/${workspaceId}/files/${fileId}`)
+    )
+    assert.equal(file.file.uploadedBy, ownerId)
+    await readVersionBytes(1, 'version one')
+    await readVersionBytes(2, 'version two')
+    await request(ownerId, `/api/table/${tableId}`, 'PATCH', {
+      workspaceId,
+      name: 'Retained_table_edited',
+    })
+    const rows = await request(ownerId, `/api/table/${tableId}/rows?workspaceId=${workspaceId}`)
+    const retainedRows = record(rows.data).rows
+    assert(Array.isArray(retainedRows))
+    assert.deepEqual(record(retainedRows[0]).data, { value: 'survives' })
+    await request(ownerId, `/api/knowledge/${kbId}`, 'PUT', { name: 'Retained knowledge edited' })
+    await request(ownerId, `/api/workspaces/${workspaceId}/files/folders/${folderId}`, 'PATCH', {
+      name: 'Retained folder edited',
+    })
+    await request(ownerId, `/api/workspaces/${workspaceId}/files/${fileId}`, 'PATCH', {
+      name: 'retained-edited.txt',
+    })
+    assert.deepEqual(
+      await sql`SELECT billed_account_user_id, storage_used_bytes FROM workspace WHERE id = ${workspaceId}`,
+      beforeStorage
+    )
+    assert.deepEqual(
+      await sql`SELECT id, author_user_ids FROM workspace_file_version WHERE file_id = ${fileId} ORDER BY version`,
+      versionsBefore
+    )
+    assert.equal((await sql`SELECT id FROM document WHERE id = ${documentId}`).length, 1)
+    await request(ownerId, `/api/workspaces/${workspaceId}/files/${fileId}/content`, 'PUT', {
+      content: 'survivor version',
+    })
+  })
+} catch (error) {
+  logger.error('Lifecycle E2E failed', { error })
+  process.exitCode = 1
+} finally {
+  try {
+    for (const cleanupFileId of [fileId, implicitFileId].filter(Boolean)) {
+      const versions =
+        await sql`SELECT version FROM workspace_file_version WHERE file_id = ${cleanupFileId} ORDER BY version`
+      for (const version of versions.slice(0, -1)) {
+        const path = `/api/v2/files/${cleanupFileId}/versions/${version.version}?workspaceId=${workspaceId}`
+        const response = await fetch(new URL(path, baseUrl), {
+          method: 'DELETE',
+          headers: { 'X-API-Key': ownerKey },
+          signal: AbortSignal.timeout(120_000),
+        })
+        assert.equal(response.status, 200, await response.text())
+      }
+      const [file] = await sql`SELECT key FROM workspace_files WHERE id = ${cleanupFileId}`
+      if (file)
+        await request(ownerId, '/api/files/delete', 'POST', {
+          filePath: `/api/files/serve/${file.key}`,
+          context: 'workspace',
+        })
+    }
+  } catch (error) {
+    checks.push({
+      name: 'Stored-object cleanup',
+      status: 'failed',
+      durationMs: 0,
+      error: getErrorMessage(error),
+    })
+    process.exitCode = 1
+  } finally {
+    try {
+      await sql`DELETE FROM workspace WHERE id = ${workspaceId}`
+      await sql`DELETE FROM "user" WHERE id IN (${ownerId}, ${departingId})`
+    } finally {
+      await sql.end()
+      const untrackedHashes: Record<string, string> = {}
+      for (const path of execFileSync('git', ['ls-files', '--others', '--exclude-standard'], {
+        encoding: 'utf8',
+      })
+        .trim()
+        .split('\n')
+        .filter(Boolean)) {
+        untrackedHashes[path] = sha256Hex(await readFile(path))
+      }
+      await writeFile(
+        reportPath,
+        JSON.stringify(
+          {
+            startedAt,
+            finishedAt: new Date().toISOString(),
+            sourceSha: execFileSync('git', ['rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(),
+            trackedDiffSha256: sha256Hex(execFileSync('git', ['diff', 'HEAD'])),
+            untrackedHashes,
+            checks,
+            requests,
+          },
+          null,
+          2
+        )
+      )
+    }
+  }
+}
diff --git a/packages/testing/src/mocks/billing-storage.mock.ts b/packages/testing/src/mocks/billing-storage.mock.ts
index 9b1de2043a4..334280ec593 100644
--- a/packages/testing/src/mocks/billing-storage.mock.ts
+++ b/packages/testing/src/mocks/billing-storage.mock.ts
@@ -38,6 +38,7 @@ export const billingStorageMockFns = {
   mockGetUserStorageLimit: vi.fn(),
   mockGetUserStorageUsage: vi.fn(),
   mockApplyStorageUsageDeltasInTx: vi.fn(),
+  mockLockWorkspaceStorageForMutationInTx: vi.fn(),
   mockCheckAndIncrementStorageUsageInTx: vi.fn(),
   mockDecrementStorageUsageForBillingContextInTx: vi.fn(),
   mockIncrementAdmittedStorageUsageForBillingContextInTx: vi.fn(),
@@ -64,6 +65,8 @@ export const billingStorageMock = {
   getUserStorageLimit: billingStorageMockFns.mockGetUserStorageLimit,
   getUserStorageUsage: billingStorageMockFns.mockGetUserStorageUsage,
   applyStorageUsageDeltasInTx: billingStorageMockFns.mockApplyStorageUsageDeltasInTx,
+  lockWorkspaceStorageForMutationInTx:
+    billingStorageMockFns.mockLockWorkspaceStorageForMutationInTx,
   checkAndIncrementStorageUsageInTx: billingStorageMockFns.mockCheckAndIncrementStorageUsageInTx,
   decrementStorageUsageForBillingContextInTx:
     billingStorageMockFns.mockDecrementStorageUsageForBillingContextInTx,