diff --git a/apps/sim/app/api/workspaces/[id]/route.test.ts b/apps/sim/app/api/workspaces/[id]/route.test.ts
new file mode 100644
index 00000000000..2774b0c8638
--- /dev/null
+++ b/apps/sim/app/api/workspaces/[id]/route.test.ts
@@ -0,0 +1,134 @@
+/**
+ * @vitest-environment node
+ */
+import {
+  auditMock,
+  auditMockFns,
+  authMockFns,
+  createMockRequest,
+  permissionsMock,
+  permissionsMockFns,
+  posthogServerMock,
+  posthogServerMockFns,
+} from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockArchiveWorkspace } = vi.hoisted(() => ({
+  mockArchiveWorkspace: vi.fn(),
+}))
+
+vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock)
+
+vi.mock('@/lib/workspaces/lifecycle', () => ({
+  archiveWorkspace: mockArchiveWorkspace,
+}))
+
+vi.mock('@sim/audit', () => auditMock)
+
+vi.mock('@/lib/posthog/server', () => posthogServerMock)
+
+import { DELETE } from '@/app/api/workspaces/[id]/route'
+
+function callDelete(workspaceId = 'workspace-1') {
+  const request = createMockRequest('DELETE', {})
+  return DELETE(request, { params: Promise.resolve({ id: workspaceId }) })
+}
+
+describe('DELETE /api/workspaces/[id]', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    authMockFns.mockGetSession.mockResolvedValue({
+      user: { id: 'user-admin', name: 'Admin', email: 'admin@example.com' },
+    })
+    permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('admin')
+  })
+
+  it('returns 401 when the caller is unauthenticated', async () => {
+    authMockFns.mockGetSession.mockResolvedValue(null)
+
+    const response = await callDelete()
+
+    expect(response.status).toBe(401)
+    expect(mockArchiveWorkspace).not.toHaveBeenCalled()
+  })
+
+  it('returns 403 when the caller lacks admin permission', async () => {
+    permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('write')
+
+    const response = await callDelete()
+
+    expect(response.status).toBe(403)
+    expect(mockArchiveWorkspace).not.toHaveBeenCalled()
+  })
+
+  it('returns 404 when the workspace does not exist', async () => {
+    mockArchiveWorkspace.mockResolvedValue({ archived: false })
+
+    const response = await callDelete()
+
+    expect(response.status).toBe(404)
+    expect(auditMockFns.mockRecordAudit).not.toHaveBeenCalled()
+  })
+
+  it('archives the workspace, records an audit entry, and captures the event on success', async () => {
+    mockArchiveWorkspace.mockResolvedValue({
+      archived: true,
+      workspaceName: 'Test Workspace',
+    })
+
+    const response = await callDelete('workspace-1')
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(body).toEqual({ success: true })
+    expect(mockArchiveWorkspace).toHaveBeenCalledWith('workspace-1', {
+      requestId: 'workspace-workspace-1',
+      provisionFallbackForStrandedMembers: true,
+      actorId: 'user-admin',
+      actorName: 'Admin',
+      actorEmail: 'admin@example.com',
+    })
+    expect(auditMockFns.mockRecordAudit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workspaceId: 'workspace-1',
+        actorId: 'user-admin',
+        resourceName: 'Test Workspace',
+      })
+    )
+    expect(posthogServerMockFns.mockCaptureServerEvent).toHaveBeenCalledWith(
+      'user-admin',
+      'workspace_deleted',
+      expect.objectContaining({ workspace_id: 'workspace-1' }),
+      expect.objectContaining({ groups: { workspace: 'workspace-1' } })
+    )
+  })
+
+  it('succeeds and records which members were auto-provisioned a replacement workspace', async () => {
+    mockArchiveWorkspace.mockResolvedValue({
+      archived: true,
+      workspaceName: 'Test Workspace',
+      provisionedWorkspaceUserIds: ['user-victim'],
+    })
+
+    const response = await callDelete('workspace-1')
+    const body = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(body).toEqual({ success: true })
+    expect(auditMockFns.mockRecordAudit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        metadata: expect.objectContaining({
+          provisionedWorkspaceUserIds: ['user-victim'],
+        }),
+      })
+    )
+  })
+
+  it('returns 500 when archival throws unexpectedly', async () => {
+    mockArchiveWorkspace.mockRejectedValue(new Error('db exploded'))
+
+    const response = await callDelete()
+
+    expect(response.status).toBe(500)
+  })
+})
diff --git a/apps/sim/app/api/workspaces/[id]/route.ts b/apps/sim/app/api/workspaces/[id]/route.ts
index 1b215792d94..e84c3d3f855 100644
--- a/apps/sim/app/api/workspaces/[id]/route.ts
+++ b/apps/sim/app/api/workspaces/[id]/route.ts
@@ -12,7 +12,7 @@ import { archiveWorkspace } from '@/lib/workspaces/lifecycle'
 const logger = createLogger('WorkspaceByIdAPI')
 
 import { db } from '@sim/db'
-import { permissions, workspace } from '@sim/db/schema'
+import { workspace } from '@sim/db/schema'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import {
   getEffectiveWorkspacePermission,
@@ -239,30 +239,6 @@ export const DELETE = withRouteHandler(
     }
 
     try {
-      const [[workspaceRecord], totalWorkspaces] = await Promise.all([
-        db
-          .select({ name: workspace.name })
-          .from(workspace)
-          .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt)))
-          .limit(1),
-        db
-          .select({ id: permissions.entityId })
-          .from(permissions)
-          .innerJoin(workspace, eq(permissions.entityId, workspace.id))
-          .where(
-            and(
-              eq(permissions.userId, session.user.id),
-              eq(permissions.entityType, 'workspace'),
-              isNull(workspace.archivedAt)
-            )
-          ),
-      ])
-
-      /** Counts all workspace memberships (any role), not just admin — prevents the user from reaching a zero-workspace state. */
-      if (totalWorkspaces.length <= 1) {
-        return NextResponse.json({ error: 'Cannot delete the only workspace' }, { status: 400 })
-      }
-
       logger.info(`Deleting workspace ${workspaceId} for user ${session.user.id}`)
 
       const workspaceWorkflows = await db
@@ -274,9 +250,13 @@ export const DELETE = withRouteHandler(
 
       const archiveResult = await archiveWorkspace(workspaceId, {
         requestId: `workspace-${workspaceId}`,
+        provisionFallbackForStrandedMembers: true,
+        actorId: session.user.id,
+        actorName: session.user.name,
+        actorEmail: session.user.email,
       })
 
-      if (!archiveResult.archived && !workspaceRecord) {
+      if (!archiveResult.archived) {
         return NextResponse.json({ error: 'Workspace not found' }, { status: 404 })
       }
 
@@ -288,13 +268,16 @@ export const DELETE = withRouteHandler(
         action: AuditAction.WORKSPACE_DELETED,
         resourceType: AuditResourceType.WORKSPACE,
         resourceId: workspaceId,
-        resourceName: workspaceRecord?.name,
-        description: `Archived workspace "${workspaceRecord?.name || workspaceId}"`,
+        resourceName: archiveResult.workspaceName,
+        description: `Archived workspace "${archiveResult.workspaceName || workspaceId}"`,
         metadata: {
           affected: {
             workflows: workflowIds.length,
           },
           archived: archiveResult.archived,
+          ...(archiveResult.provisionedWorkspaceUserIds?.length && {
+            provisionedWorkspaceUserIds: archiveResult.provisionedWorkspaceUserIds,
+          }),
         },
         request,
       })
diff --git a/apps/sim/app/api/workspaces/route.ts b/apps/sim/app/api/workspaces/route.ts
index 8d35dc7429b..e8cc520609c 100644
--- a/apps/sim/app/api/workspaces/route.ts
+++ b/apps/sim/app/api/workspaces/route.ts
@@ -1,8 +1,7 @@
 import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
-import { permissions, settings, type WorkspaceMode, workflow, workspace } from '@sim/db/schema'
+import { settings, type WorkspaceMode, workflow } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { generateId } from '@sim/utils/id'
 import { and, eq, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { listWorkspacesQuerySchema } from '@/lib/api/contracts'
@@ -10,12 +9,9 @@ import { createWorkspaceContract } from '@/lib/api/contracts/workspaces'
 import { parseRequest } from '@/lib/api/server'
 import { getSession } from '@/lib/auth'
 import type { PlanCategory } from '@/lib/billing/plan-helpers'
-import { PlatformEvents } from '@/lib/core/telemetry'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { captureServerEvent } from '@/lib/posthog/server'
-import { buildDefaultWorkflowArtifacts } from '@/lib/workflows/defaults'
-import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils'
-import { getRandomWorkspaceColor } from '@/lib/workspaces/colors'
+import { type CreateWorkspaceRecordParams, createWorkspaceRecord } from '@/lib/workspaces/create'
 import {
   CONTACT_OWNER_TO_UPGRADE_REASON,
   evaluateWorkspaceInvitePolicy,
@@ -264,15 +260,7 @@ async function createDefaultWorkspace(
   })
 }
 
-interface CreateWorkspaceParams {
-  userId: string
-  name: string
-  skipDefaultWorkflow?: boolean
-  explicitColor?: string
-  organizationId: string | null
-  workspaceMode: WorkspaceMode
-  billedAccountUserId: string
-}
+type CreateWorkspaceParams = Omit<CreateWorkspaceRecordParams, 'executor'>
 
 async function createWorkspace({
   userId,
@@ -283,96 +271,15 @@ async function createWorkspace({
   workspaceMode,
   billedAccountUserId,
 }: CreateWorkspaceParams) {
-  const workspaceId = generateId()
-  const workflowId = generateId()
-  const now = new Date()
-  const color = explicitColor || getRandomWorkspaceColor()
-
-  try {
-    await db.transaction(async (tx) => {
-      await tx.insert(workspace).values({
-        id: workspaceId,
-        name,
-        color,
-        ownerId: userId,
-        organizationId,
-        workspaceMode,
-        billedAccountUserId,
-        allowPersonalApiKeys: true,
-        createdAt: now,
-        updatedAt: now,
-      })
-
-      const permissionRows = [
-        {
-          id: generateId(),
-          entityType: 'workspace' as const,
-          entityId: workspaceId,
-          userId,
-          permissionType: 'admin' as const,
-          createdAt: now,
-          updatedAt: now,
-        },
-      ]
-
-      if (
-        workspaceMode === WORKSPACE_MODE.ORGANIZATION &&
-        billedAccountUserId &&
-        billedAccountUserId !== userId
-      ) {
-        permissionRows.push({
-          id: generateId(),
-          entityType: 'workspace' as const,
-          entityId: workspaceId,
-          userId: billedAccountUserId,
-          permissionType: 'admin' as const,
-          createdAt: now,
-          updatedAt: now,
-        })
-      }
-
-      await tx.insert(permissions).values(permissionRows)
-
-      if (!skipDefaultWorkflow) {
-        await tx.insert(workflow).values({
-          id: workflowId,
-          userId,
-          workspaceId,
-          folderId: null,
-          name: 'default-agent',
-          description: 'Your first workflow - start building here!',
-          lastSynced: now,
-          createdAt: now,
-          updatedAt: now,
-          isDeployed: false,
-          runCount: 0,
-          variables: {},
-        })
-
-        const { workflowState } = buildDefaultWorkflowArtifacts()
-        await saveWorkflowToNormalizedTables(workflowId, workflowState, tx)
-      }
-
-      logger.info(
-        skipDefaultWorkflow
-          ? `Created ${workspaceMode} workspace ${workspaceId} for user ${userId}`
-          : `Created ${workspaceMode} workspace ${workspaceId} with initial workflow ${workflowId} for user ${userId}`
-      )
-    })
-  } catch (error) {
-    logger.error(`Failed to create workspace ${workspaceId}:`, error)
-    throw error
-  }
-
-  try {
-    PlatformEvents.workspaceCreated({
-      workspaceId,
-      userId,
-      name,
-    })
-  } catch {
-    // Telemetry should not fail the operation
-  }
+  const record = await createWorkspaceRecord({
+    userId,
+    name,
+    skipDefaultWorkflow,
+    explicitColor,
+    organizationId,
+    workspaceMode,
+    billedAccountUserId,
+  })
 
   const invitePolicy = await getWorkspaceInvitePolicy({
     organizationId,
@@ -389,16 +296,7 @@ async function createWorkspace({
       : CONTACT_OWNER_TO_UPGRADE_REASON
 
   return {
-    id: workspaceId,
-    name,
-    color,
-    ownerId: userId,
-    organizationId,
-    workspaceMode,
-    billedAccountUserId,
-    allowPersonalApiKeys: true,
-    createdAt: now,
-    updatedAt: now,
+    ...record,
     role: 'owner',
     permissions: 'admin',
     inviteMembersEnabled: invitePolicy.allowed,
diff --git a/apps/sim/lib/auth/ban.ts b/apps/sim/lib/auth/ban.ts
index 8ca733d2994..621c675e422 100644
--- a/apps/sim/lib/auth/ban.ts
+++ b/apps/sim/lib/auth/ban.ts
@@ -1,6 +1,7 @@
 import { db, user } from '@sim/db'
 import { inArray, sql } from 'drizzle-orm'
 import { getAccessControlConfig, isEmailBlockedByAccessControl } from '@/lib/auth/access-control'
+import type { DbOrTx } from '@/lib/db/types'
 
 /**
  * True when a ban is currently in effect. Mirrors better-auth admin-plugin
@@ -34,14 +35,22 @@ export async function isEmailBlocked(email: string | null | undefined): Promise<
  * active account ban, or an email/domain in the appconfig blocked lists.
  * One user query plus the cached access-control fetch. Throws on db
  * failure — callers must fail closed.
+ *
+ * Accepts an optional executor so callers already inside a transaction (e.g. a
+ * workspace-archival safety check under `serializable` isolation) can run this
+ * against `tx` instead of borrowing a second pooled connection while the first
+ * sits idle-in-transaction.
  */
-export async function getActivelyBannedUserIds(userIds: string[]): Promise<string[]> {
+export async function getActivelyBannedUserIds(
+  userIds: string[],
+  executor: DbOrTx = db
+): Promise<string[]> {
   const ids = [...new Set(userIds.filter(Boolean))]
   if (ids.length === 0) return []
 
   const [accessControl, rows] = await Promise.all([
     getAccessControlConfig(),
-    db
+    executor
       .select({ id: user.id, email: user.email, banned: user.banned, banExpires: user.banExpires })
       .from(user)
       .where(inArray(user.id, ids)),
diff --git a/apps/sim/lib/workflows/lifecycle.test.ts b/apps/sim/lib/workflows/lifecycle.test.ts
index db4af82822e..c63a08f590f 100644
--- a/apps/sim/lib/workflows/lifecycle.test.ts
+++ b/apps/sim/lib/workflows/lifecycle.test.ts
@@ -10,14 +10,21 @@ import {
 } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockSelect, mockTransaction, mockCleanupExternalWebhook, mockWorkflowDeleted } = vi.hoisted(
-  () => ({
-    mockSelect: vi.fn(),
-    mockTransaction: vi.fn(),
-    mockCleanupExternalWebhook: vi.fn(),
-    mockWorkflowDeleted: vi.fn(),
-  })
-)
+const {
+  mockSelect,
+  mockTransaction,
+  mockDelete,
+  mockCleanupExternalWebhook,
+  mockWorkflowDeleted,
+  mockArchiveWorkspace,
+} = vi.hoisted(() => ({
+  mockSelect: vi.fn(),
+  mockTransaction: vi.fn(),
+  mockDelete: vi.fn(),
+  mockCleanupExternalWebhook: vi.fn(),
+  mockWorkflowDeleted: vi.fn(),
+  mockArchiveWorkspace: vi.fn(),
+}))
 
 const mockGetWorkflowById = workflowsUtilsMockFns.mockGetWorkflowById
 
@@ -25,9 +32,14 @@ vi.mock('@sim/db', () => ({
   db: {
     select: mockSelect,
     transaction: mockTransaction,
+    delete: mockDelete,
   },
 }))
 
+vi.mock('@/lib/workspaces/lifecycle', () => ({
+  archiveWorkspace: mockArchiveWorkspace,
+}))
+
 vi.mock('@/lib/workflows/utils', () => workflowsUtilsMock)
 
 vi.mock('@/lib/webhooks/provider-subscriptions', () => ({
@@ -46,7 +58,7 @@ vi.mock('@/lib/core/telemetry', () => ({
   },
 }))
 
-import { archiveWorkflow } from '@/lib/workflows/lifecycle'
+import { archiveWorkflow, disableUserResources } from '@/lib/workflows/lifecycle'
 
 function createSelectChain<T>(result: T) {
   const chain = {
@@ -129,3 +141,28 @@ describe('workflow lifecycle', () => {
     expect(fetch).not.toHaveBeenCalled()
   })
 })
+
+describe('disableUserResources', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('archives every owned workspace with fallback provisioning opted in, so a non-banned co-member is not stranded', async () => {
+    mockSelect.mockReturnValue(createSelectChain([{ id: 'workspace-1' }, { id: 'workspace-2' }]))
+    mockDelete.mockReturnValue({ where: vi.fn().mockResolvedValue([]) })
+    mockArchiveWorkspace.mockResolvedValue({ archived: true, workspaceName: 'Workspace' })
+
+    await disableUserResources('user-banned')
+
+    expect(mockArchiveWorkspace).toHaveBeenCalledTimes(2)
+    expect(mockArchiveWorkspace).toHaveBeenCalledWith(
+      'workspace-1',
+      expect.objectContaining({ provisionFallbackForStrandedMembers: true })
+    )
+    expect(mockArchiveWorkspace).toHaveBeenCalledWith(
+      'workspace-2',
+      expect.objectContaining({ provisionFallbackForStrandedMembers: true })
+    )
+    expect(mockDelete).toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts
index 82040e2750e..0ccd8c942b8 100644
--- a/apps/sim/lib/workflows/lifecycle.ts
+++ b/apps/sim/lib/workflows/lifecycle.ts
@@ -346,7 +346,11 @@ export async function archiveWorkflowsByIdsInWorkspace(
 /**
  * Disables all resources owned by a banned user by archiving every workspace
  * they own (cascading to workflows, chats, KBs, tables, files, etc.)
- * and deleting their personal API keys.
+ * and deleting their personal API keys. Still opts into stranded-member
+ * fallback provisioning: the banned owner is excluded from receiving a
+ * fallback (`findMembersStrandedByArchival` filters actively banned users),
+ * but a non-banned co-member of an owned workspace can still be stranded by
+ * the ban and must get the same safety net as an interactive deletion.
  */
 export async function disableUserResources(userId: string): Promise<void> {
   const requestId = generateRequestId()
@@ -360,7 +364,9 @@ export async function disableUserResources(userId: string): Promise<void> {
     .where(and(eq(workspace.ownerId, userId), isNull(workspace.archivedAt)))
 
   await Promise.all([
-    ...ownedWorkspaces.map((w) => archiveWorkspace(w.id, { requestId })),
+    ...ownedWorkspaces.map((w) =>
+      archiveWorkspace(w.id, { requestId, provisionFallbackForStrandedMembers: true })
+    ),
     db.delete(apiKey).where(eq(apiKey.userId, userId)),
   ])
 
diff --git a/apps/sim/lib/workspaces/create.test.ts b/apps/sim/lib/workspaces/create.test.ts
new file mode 100644
index 00000000000..e7cd6ce9352
--- /dev/null
+++ b/apps/sim/lib/workspaces/create.test.ts
@@ -0,0 +1,162 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockTransaction, mockSaveWorkflowToNormalizedTables, mockWorkspaceCreatedEvent } =
+  vi.hoisted(() => ({
+    mockTransaction: vi.fn(),
+    mockSaveWorkflowToNormalizedTables: vi.fn(),
+    mockWorkspaceCreatedEvent: vi.fn(),
+  }))
+
+vi.mock('@sim/db', () => ({
+  db: {
+    transaction: mockTransaction,
+  },
+}))
+
+vi.mock('@/lib/core/telemetry', () => ({
+  PlatformEvents: { workspaceCreated: mockWorkspaceCreatedEvent },
+}))
+
+vi.mock('@/lib/workflows/defaults', () => ({
+  buildDefaultWorkflowArtifacts: () => ({ workflowState: { blocks: {}, edges: [] } }),
+}))
+
+vi.mock('@/lib/workflows/persistence/utils', () => ({
+  saveWorkflowToNormalizedTables: (...args: unknown[]) =>
+    mockSaveWorkflowToNormalizedTables(...args),
+}))
+
+vi.mock('@/lib/workspaces/colors', () => ({
+  getRandomWorkspaceColor: () => '#123456',
+}))
+
+import { createWorkspaceRecord } from './create'
+
+function createInsertOnlyTx() {
+  return {
+    insert: vi.fn().mockImplementation(() => ({
+      values: vi.fn().mockResolvedValue([]),
+    })),
+  }
+}
+
+describe('createWorkspaceRecord', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('opens its own transaction when no executor is provided', async () => {
+    const tx = createInsertOnlyTx()
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const record = await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'My Workspace',
+      organizationId: null,
+      workspaceMode: 'personal',
+      billedAccountUserId: 'user-1',
+    })
+
+    expect(mockTransaction).toHaveBeenCalledTimes(1)
+    expect(record.name).toBe('My Workspace')
+    expect(record.ownerId).toBe('user-1')
+    expect(record.workspaceMode).toBe('personal')
+    expect(tx.insert).toHaveBeenCalledTimes(3)
+    expect(mockSaveWorkflowToNormalizedTables).toHaveBeenCalledTimes(1)
+  })
+
+  it('runs directly against a provided executor instead of opening a nested transaction', async () => {
+    const tx = createInsertOnlyTx()
+
+    await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'My Workspace',
+      organizationId: null,
+      workspaceMode: 'personal',
+      billedAccountUserId: 'user-1',
+      executor: tx as never,
+    })
+
+    expect(mockTransaction).not.toHaveBeenCalled()
+    expect(tx.insert).toHaveBeenCalledTimes(3)
+  })
+
+  it('fires the workspaceCreated event once its own transaction commits', async () => {
+    const tx = createInsertOnlyTx()
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const record = await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'My Workspace',
+      organizationId: null,
+      workspaceMode: 'personal',
+      billedAccountUserId: 'user-1',
+    })
+
+    expect(mockWorkspaceCreatedEvent).toHaveBeenCalledWith({
+      workspaceId: record.id,
+      userId: 'user-1',
+      name: 'My Workspace',
+    })
+  })
+
+  it('does not fire the workspaceCreated event when given a caller-owned executor', async () => {
+    const tx = createInsertOnlyTx()
+
+    await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'My Workspace',
+      organizationId: null,
+      workspaceMode: 'personal',
+      billedAccountUserId: 'user-1',
+      executor: tx as never,
+    })
+
+    expect(mockWorkspaceCreatedEvent).not.toHaveBeenCalled()
+  })
+
+  it('skips the default workflow insert when skipDefaultWorkflow is set', async () => {
+    const tx = createInsertOnlyTx()
+
+    await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'My Workspace',
+      organizationId: null,
+      workspaceMode: 'personal',
+      billedAccountUserId: 'user-1',
+      skipDefaultWorkflow: true,
+      executor: tx as never,
+    })
+
+    expect(tx.insert).toHaveBeenCalledTimes(2)
+    expect(mockSaveWorkflowToNormalizedTables).not.toHaveBeenCalled()
+  })
+
+  it('adds a second admin permission row for the billed account when it differs from the owner in org mode', async () => {
+    const tx = createInsertOnlyTx()
+
+    await createWorkspaceRecord({
+      userId: 'user-1',
+      name: 'Org Workspace',
+      organizationId: 'org-1',
+      workspaceMode: 'organization',
+      billedAccountUserId: 'user-billing',
+      executor: tx as never,
+    })
+
+    const permissionValuesCall = tx.insert.mock.results[1].value.values as ReturnType<typeof vi.fn>
+    const insertedPermissionRows = permissionValuesCall.mock.calls[0][0]
+    expect(insertedPermissionRows).toHaveLength(2)
+    expect(insertedPermissionRows.map((row: { userId: string }) => row.userId)).toEqual([
+      'user-1',
+      'user-billing',
+    ])
+  })
+})
diff --git a/apps/sim/lib/workspaces/create.ts b/apps/sim/lib/workspaces/create.ts
new file mode 100644
index 00000000000..893c961b82e
--- /dev/null
+++ b/apps/sim/lib/workspaces/create.ts
@@ -0,0 +1,170 @@
+import { db } from '@sim/db'
+import { permissions, type WorkspaceMode, workflow, workspace } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { generateId } from '@sim/utils/id'
+import { PlatformEvents } from '@/lib/core/telemetry'
+import type { DbOrTx } from '@/lib/db/types'
+import { buildDefaultWorkflowArtifacts } from '@/lib/workflows/defaults'
+import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils'
+import { getRandomWorkspaceColor } from '@/lib/workspaces/colors'
+import { WORKSPACE_MODE } from '@/lib/workspaces/policy'
+
+const logger = createLogger('WorkspaceCreate')
+
+export interface CreateWorkspaceRecordParams {
+  userId: string
+  name: string
+  skipDefaultWorkflow?: boolean
+  explicitColor?: string
+  organizationId: string | null
+  workspaceMode: WorkspaceMode
+  billedAccountUserId: string
+  /**
+   * Runs the insert against an existing transaction instead of opening a new one — for callers
+   * that need workspace creation to be atomic with other writes (e.g. archiving a workspace that
+   * would otherwise strand a member). Defaults to opening its own transaction against `db`.
+   */
+  executor?: DbOrTx
+}
+
+export interface CreatedWorkspaceRecord {
+  id: string
+  name: string
+  color: string
+  ownerId: string
+  organizationId: string | null
+  workspaceMode: WorkspaceMode
+  billedAccountUserId: string
+  allowPersonalApiKeys: boolean
+  createdAt: Date
+  updatedAt: Date
+}
+
+/**
+ * Core workspace-creation write: inserts the workspace, its owner admin permission row, and
+ * (unless skipped) a default starter workflow. Shared by the `POST /api/workspaces` route and
+ * the workspace-archival safety net that auto-provisions a replacement workspace for a member
+ * who would otherwise be left with zero workspaces.
+ *
+ * Fires the `workspaceCreated` telemetry event itself only when it manages its own transaction
+ * (no `executor` passed). Callers that pass `executor` are joining an outer transaction that can
+ * still roll back after this returns, so they own firing that event once their transaction commits.
+ */
+export async function createWorkspaceRecord({
+  userId,
+  name,
+  skipDefaultWorkflow = false,
+  explicitColor,
+  organizationId,
+  workspaceMode,
+  billedAccountUserId,
+  executor,
+}: CreateWorkspaceRecordParams): Promise<CreatedWorkspaceRecord> {
+  const workspaceId = generateId()
+  const workflowId = generateId()
+  const now = new Date()
+  const color = explicitColor || getRandomWorkspaceColor()
+
+  const run = async (tx: DbOrTx) => {
+    await tx.insert(workspace).values({
+      id: workspaceId,
+      name,
+      color,
+      ownerId: userId,
+      organizationId,
+      workspaceMode,
+      billedAccountUserId,
+      allowPersonalApiKeys: true,
+      createdAt: now,
+      updatedAt: now,
+    })
+
+    const permissionRows = [
+      {
+        id: generateId(),
+        entityType: 'workspace' as const,
+        entityId: workspaceId,
+        userId,
+        permissionType: 'admin' as const,
+        createdAt: now,
+        updatedAt: now,
+      },
+    ]
+
+    if (
+      workspaceMode === WORKSPACE_MODE.ORGANIZATION &&
+      billedAccountUserId &&
+      billedAccountUserId !== userId
+    ) {
+      permissionRows.push({
+        id: generateId(),
+        entityType: 'workspace' as const,
+        entityId: workspaceId,
+        userId: billedAccountUserId,
+        permissionType: 'admin' as const,
+        createdAt: now,
+        updatedAt: now,
+      })
+    }
+
+    await tx.insert(permissions).values(permissionRows)
+
+    if (!skipDefaultWorkflow) {
+      await tx.insert(workflow).values({
+        id: workflowId,
+        userId,
+        workspaceId,
+        folderId: null,
+        name: 'default-agent',
+        description: 'Your first workflow - start building here!',
+        lastSynced: now,
+        createdAt: now,
+        updatedAt: now,
+        isDeployed: false,
+        runCount: 0,
+        variables: {},
+      })
+
+      const { workflowState } = buildDefaultWorkflowArtifacts()
+      await saveWorkflowToNormalizedTables(workflowId, workflowState, tx)
+    }
+
+    logger.info(
+      skipDefaultWorkflow
+        ? `Created ${workspaceMode} workspace ${workspaceId} for user ${userId}`
+        : `Created ${workspaceMode} workspace ${workspaceId} with initial workflow ${workflowId} for user ${userId}`
+    )
+  }
+
+  try {
+    if (executor) {
+      await run(executor)
+    } else {
+      await db.transaction(run)
+    }
+  } catch (error) {
+    logger.error(`Failed to create workspace ${workspaceId}:`, error)
+    throw error
+  }
+
+  if (!executor) {
+    try {
+      PlatformEvents.workspaceCreated({ workspaceId, userId, name })
+    } catch {
+      // Telemetry should not fail the operation
+    }
+  }
+
+  return {
+    id: workspaceId,
+    name,
+    color,
+    ownerId: userId,
+    organizationId,
+    workspaceMode,
+    billedAccountUserId,
+    allowPersonalApiKeys: true,
+    createdAt: now,
+    updatedAt: now,
+  }
+}
diff --git a/apps/sim/lib/workspaces/lifecycle.test.ts b/apps/sim/lib/workspaces/lifecycle.test.ts
index 1013b2280f4..dee447fa0d4 100644
--- a/apps/sim/lib/workspaces/lifecycle.test.ts
+++ b/apps/sim/lib/workspaces/lifecycle.test.ts
@@ -1,13 +1,25 @@
 /**
  * @vitest-environment node
  */
-import { permissionsMock, permissionsMockFns } from '@sim/testing'
+import { auditMock, auditMockFns, permissionsMock, permissionsMockFns } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockSelect, mockTransaction, mockArchiveWorkflowsForWorkspace } = vi.hoisted(() => ({
+const {
+  mockSelect,
+  mockTransaction,
+  mockArchiveWorkflowsForWorkspace,
+  mockListAccessibleWorkspaceRowsForUser,
+  mockCreateWorkspaceRecord,
+  mockGetActivelyBannedUserIds,
+  mockWorkspaceCreatedEvent,
+} = vi.hoisted(() => ({
   mockSelect: vi.fn(),
   mockTransaction: vi.fn(),
   mockArchiveWorkflowsForWorkspace: vi.fn(),
+  mockListAccessibleWorkspaceRowsForUser: vi.fn(),
+  mockCreateWorkspaceRecord: vi.fn(),
+  mockGetActivelyBannedUserIds: vi.fn(),
+  mockWorkspaceCreatedEvent: vi.fn(),
 }))
 
 const mockGetWorkspaceWithOwner = permissionsMockFns.mockGetWorkspaceWithOwner
@@ -25,6 +37,24 @@ vi.mock('@/lib/workflows/lifecycle', () => ({
 
 vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock)
 
+vi.mock('@/lib/workspaces/utils', () => ({
+  listAccessibleWorkspaceRowsForUser: mockListAccessibleWorkspaceRowsForUser,
+}))
+
+vi.mock('@/lib/workspaces/create', () => ({
+  createWorkspaceRecord: mockCreateWorkspaceRecord,
+}))
+
+vi.mock('@/lib/auth/ban', () => ({
+  getActivelyBannedUserIds: mockGetActivelyBannedUserIds,
+}))
+
+vi.mock('@/lib/core/telemetry', () => ({
+  PlatformEvents: { workspaceCreated: mockWorkspaceCreatedEvent },
+}))
+
+vi.mock('@sim/audit', () => auditMock)
+
 import { archiveWorkspace } from './lifecycle'
 
 function createUpdateChain() {
@@ -35,12 +65,54 @@ function createUpdateChain() {
   }
 }
 
+function createMembersChain(members: Array<{ userId: string }>) {
+  return {
+    from: vi.fn().mockReturnValue({
+      where: vi.fn().mockResolvedValue(members),
+    }),
+  }
+}
+
+function accessibleWorkspaceRow(workspaceId: string) {
+  return { workspace: { id: workspaceId }, permissionType: 'admin' as const }
+}
+
+function createTx(
+  members: Array<{ userId: string }>,
+  orgAdminMembers: Array<{ userId: string }> = []
+) {
+  const selectDistinct = vi.fn()
+  // Mocked in call order: explicit-permissions query, then org-admin query.
+  selectDistinct.mockReturnValueOnce(createMembersChain(members))
+  selectDistinct.mockReturnValueOnce(createMembersChain(orgAdminMembers))
+
+  return {
+    selectDistinct,
+    select: vi.fn().mockReturnValue({
+      from: vi.fn().mockReturnValue({
+        where: vi.fn().mockResolvedValue([]),
+      }),
+    }),
+    update: vi.fn().mockImplementation(() => createUpdateChain()),
+    delete: vi.fn().mockImplementation(() => ({
+      where: vi.fn().mockResolvedValue([]),
+    })),
+  }
+}
+
 describe('workspace lifecycle', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    mockSelect.mockReturnValue({
+      from: vi.fn().mockReturnValue({
+        where: vi.fn().mockResolvedValue([]),
+      }),
+    })
+    mockCreateWorkspaceRecord.mockResolvedValue({ id: 'fallback-workspace', name: 'My Workspace' })
+    mockGetActivelyBannedUserIds.mockResolvedValue([])
   })
 
-  it('archives workspace and dependent resources', async () => {
+  it('archives workspace and dependent resources under serializable isolation', async () => {
     mockGetWorkspaceWithOwner.mockResolvedValue({
       id: 'workspace-1',
       name: 'Workspace 1',
@@ -48,37 +120,369 @@ describe('workspace lifecycle', () => {
       archivedAt: null,
     })
     mockArchiveWorkflowsForWorkspace.mockResolvedValue(2)
-    mockSelect.mockReturnValue({
-      from: vi.fn().mockReturnValue({
-        where: vi.fn().mockResolvedValue([{ id: 'server-1' }]),
-      }),
+
+    const tx = createTx([])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+    })
+    expect(mockArchiveWorkflowsForWorkspace).toHaveBeenCalledWith('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+    expect(tx.update).toHaveBeenCalledTimes(8)
+    expect(tx.delete).toHaveBeenCalledTimes(1)
+    expect(mockCreateWorkspaceRecord).not.toHaveBeenCalled()
+    expect(mockTransaction).toHaveBeenCalledWith(expect.any(Function), {
+      isolationLevel: 'serializable',
+    })
+  })
+
+  it('auto-provisions a replacement workspace for a member who would be stranded, and still archives', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+
+    const tx = createTx([{ userId: 'user-victim' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+      actorId: 'admin-1',
+      actorName: 'Admin',
+      actorEmail: 'admin@example.com',
     })
 
-    const tx = {
-      select: vi.fn().mockReturnValue({
-        from: vi.fn().mockReturnValue({
-          where: vi.fn().mockResolvedValue([{ id: 'kb-1' }]),
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+      provisionedWorkspaceUserIds: ['user-victim'],
+    })
+    expect(mockListAccessibleWorkspaceRowsForUser).toHaveBeenCalledWith('user-victim', 'active', tx)
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledWith(
+      expect.objectContaining({
+        userId: 'user-victim',
+        workspaceMode: 'personal',
+        organizationId: null,
+        billedAccountUserId: 'user-victim',
+        executor: tx,
+      })
+    )
+    expect(auditMockFns.mockRecordAudit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        actorId: 'admin-1',
+        resourceId: 'fallback-workspace',
+        metadata: expect.objectContaining({
+          deletedWorkspaceId: 'workspace-1',
+          recipientUserId: 'user-victim',
         }),
-      }),
-      update: vi.fn().mockImplementation(() => createUpdateChain()),
-      delete: vi.fn().mockImplementation(() => ({
-        where: vi.fn().mockResolvedValue([]),
-      })),
-    }
+      })
+    )
+    expect(mockWorkspaceCreatedEvent).toHaveBeenCalledWith({
+      workspaceId: 'fallback-workspace',
+      userId: 'user-victim',
+      name: 'My Workspace',
+    })
+    expect(tx.update).toHaveBeenCalledTimes(8)
+    expect(tx.delete).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not record an audit entry for the fallback workspace when no actor is provided', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+
+    const tx = createTx([{ userId: 'user-victim' }])
     mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
       callback(tx)
     )
 
-    const result = await archiveWorkspace('workspace-1', { requestId: 'req-1' })
+    await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalled()
+    expect(auditMockFns.mockRecordAudit).not.toHaveBeenCalled()
+  })
+
+  it('does not record an audit entry or fire the workspaceCreated event for a fallback workspace whose transaction subsequently fails', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+
+    const tx = createTx([{ userId: 'user-victim' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) => {
+      await callback(tx)
+      throw new Error('serialization_failure')
+    })
+
+    await expect(
+      archiveWorkspace('workspace-1', {
+        requestId: 'req-1',
+        provisionFallbackForStrandedMembers: true,
+        actorId: 'admin-1',
+      })
+    ).rejects.toThrow('serialization_failure')
+
+    expect(auditMockFns.mockRecordAudit).not.toHaveBeenCalled()
+    expect(mockWorkspaceCreatedEvent).not.toHaveBeenCalled()
+  })
+
+  it('only provisions a fallback for the one member who would actually be stranded', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockImplementation(async (userId: string) =>
+      userId === 'user-victim'
+        ? [accessibleWorkspaceRow('workspace-1')]
+        : [accessibleWorkspaceRow('workspace-1'), accessibleWorkspaceRow('workspace-2')]
+    )
+
+    const tx = createTx([{ userId: 'user-victim' }, { userId: 'user-safe' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
 
     expect(result).toEqual({
       archived: true,
       workspaceName: 'Workspace 1',
+      provisionedWorkspaceUserIds: ['user-victim'],
     })
-    expect(mockArchiveWorkflowsForWorkspace).toHaveBeenCalledWith('workspace-1', {
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledTimes(1)
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'user-victim' })
+    )
+  })
+
+  it('does not strand an org admin who has no explicit permission row on another workspace', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+      accessibleWorkspaceRow('workspace-2'),
+    ])
+
+    const tx = createTx([{ userId: 'user-org-admin' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+    })
+    expect(mockCreateWorkspaceRecord).not.toHaveBeenCalled()
+    expect(tx.update).toHaveBeenCalledTimes(8)
+  })
+
+  it('provisions a fallback for an org admin who is stranded but has no explicit permission row', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      organizationId: 'org-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+
+    const tx = createTx([], [{ userId: 'user-org-admin-no-row' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+      provisionedWorkspaceUserIds: ['user-org-admin-no-row'],
+    })
+    expect(tx.selectDistinct).toHaveBeenCalledTimes(2)
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'user-org-admin-no-row' })
+    )
+  })
+
+  it('does not provision a fallback for an actively banned stranded member', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+    mockGetActivelyBannedUserIds.mockResolvedValue(['user-banned'])
+
+    const tx = createTx([{ userId: 'user-banned' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+      actorId: 'admin-1',
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+    })
+    expect(mockCreateWorkspaceRecord).not.toHaveBeenCalled()
+    expect(auditMockFns.mockRecordAudit).not.toHaveBeenCalled()
+  })
+
+  it('provisions a fallback for a non-banned stranded co-member while excluding the banned owner in the same run', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-banned',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+    ])
+    mockGetActivelyBannedUserIds.mockResolvedValue(['user-banned'])
+
+    const tx = createTx([{ userId: 'user-banned' }, { userId: 'user-cohort' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
       requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+      provisionedWorkspaceUserIds: ['user-cohort'],
+    })
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledTimes(1)
+    expect(mockCreateWorkspaceRecord).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'user-cohort' })
+    )
+  })
+
+  it('proceeds without provisioning when every member has another active workspace', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+    mockListAccessibleWorkspaceRowsForUser.mockResolvedValue([
+      accessibleWorkspaceRow('workspace-1'),
+      accessibleWorkspaceRow('workspace-2'),
+    ])
+
+    const tx = createTx([{ userId: 'user-safe-1' }, { userId: 'user-safe-2' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', {
+      requestId: 'req-1',
+      provisionFallbackForStrandedMembers: true,
+    })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
+    })
+    expect(mockCreateWorkspaceRecord).not.toHaveBeenCalled()
+    expect(tx.update).toHaveBeenCalledTimes(8)
+  })
+
+  it('never checks or provisions when provisionFallbackForStrandedMembers is not set, but still performs the archival writes (ban flow default)', async () => {
+    mockGetWorkspaceWithOwner.mockResolvedValue({
+      id: 'workspace-1',
+      name: 'Workspace 1',
+      ownerId: 'user-1',
+      archivedAt: null,
+    })
+    mockArchiveWorkflowsForWorkspace.mockResolvedValue(0)
+
+    const tx = createTx([{ userId: 'user-banned' }])
+    mockTransaction.mockImplementation(async (callback: (trx: typeof tx) => Promise<void>) =>
+      callback(tx)
+    )
+
+    const result = await archiveWorkspace('workspace-1', { requestId: 'req-1' })
+
+    expect(result).toEqual({
+      archived: true,
+      workspaceName: 'Workspace 1',
     })
-    expect(tx.update).toHaveBeenCalledTimes(10)
+    expect(tx.selectDistinct).not.toHaveBeenCalled()
+    expect(mockListAccessibleWorkspaceRowsForUser).not.toHaveBeenCalled()
+    expect(mockCreateWorkspaceRecord).not.toHaveBeenCalled()
+    expect(mockTransaction).toHaveBeenCalledWith(expect.any(Function), undefined)
+    expect(tx.update).toHaveBeenCalledTimes(8)
     expect(tx.delete).toHaveBeenCalledTimes(1)
   })
 
diff --git a/apps/sim/lib/workspaces/lifecycle.ts b/apps/sim/lib/workspaces/lifecycle.ts
index 59162b42682..1d3fd7a643b 100644
--- a/apps/sim/lib/workspaces/lifecycle.ts
+++ b/apps/sim/lib/workspaces/lifecycle.ts
@@ -1,3 +1,4 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
 import {
   apiKey,
@@ -7,6 +8,8 @@ import {
   knowledgeBase,
   knowledgeConnector,
   mcpServers,
+  member,
+  permissions,
   userTableDefinitions,
   workflowMcpServer,
   workflowSchedule,
@@ -14,22 +17,111 @@ import {
   workspaceFiles,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { ORG_ADMIN_ROLES } from '@sim/platform-authz/workspace'
 import { and, eq, inArray, isNull, sql } from 'drizzle-orm'
+import { getActivelyBannedUserIds } from '@/lib/auth/ban'
+import { PlatformEvents } from '@/lib/core/telemetry'
+import type { DbOrTx } from '@/lib/db/types'
 import { mcpPubSub } from '@/lib/mcp/pubsub'
 import { mcpService } from '@/lib/mcp/service'
 import { archiveWorkflowsForWorkspace } from '@/lib/workflows/lifecycle'
+import { createWorkspaceRecord } from '@/lib/workspaces/create'
 import { getWorkspaceWithOwner } from '@/lib/workspaces/permissions/utils'
+import { WORKSPACE_MODE } from '@/lib/workspaces/policy'
+import { listAccessibleWorkspaceRowsForUser } from '@/lib/workspaces/utils'
 
 const logger = createLogger('WorkspaceLifecycle')
 
+/** Matches the pre-existing "you have no workspaces" client-side recovery naming. */
+const FALLBACK_WORKSPACE_NAME = 'My Workspace'
+
 interface ArchiveWorkspaceOptions {
   requestId: string
+  /**
+   * Opts into auto-provisioning a replacement workspace for any member who'd otherwise be left
+   * with zero active workspaces. Off by default so archival stays a pure "delete this workspace"
+   * primitive for callers that don't need it. Safe to combine with a banned owner: actively banned
+   * users are excluded from receiving a fallback regardless of this flag (see
+   * `findMembersStrandedByArchival`), so the account-disable flow also sets this to protect any
+   * non-banned co-member of the banned owner's workspace.
+   */
+  provisionFallbackForStrandedMembers?: boolean
+  actorId?: string
+  actorName?: string | null
+  actorEmail?: string | null
+}
+
+interface ArchiveWorkspaceResult {
+  archived: boolean
+  workspaceName?: string
+  /** userIds who were auto-provisioned a replacement workspace because this deletion would
+   *  otherwise have left them with zero active workspaces. */
+  provisionedWorkspaceUserIds?: string[]
+}
+
+/**
+ * Returns the userIds who would be left with zero accessible active (non-archived) workspaces if
+ * `workspaceId` were archived. Candidates are the union of explicit workspace members AND the
+ * organization's admins/owners — an org admin can access a workspace purely through their org
+ * role with no permission row at all, so they must be checked even though they never show up as
+ * an explicit member. "Accessible" (via `listAccessibleWorkspaceRowsForUser`) already accounts for
+ * that same org-admin-derived access when deciding whether a candidate has another workspace to
+ * fall back to. Actively banned users are excluded from the result — they should never receive a
+ * new resource as a side effect of someone else's action.
+ *
+ * Must be called against the same executor used to perform the archival, under
+ * `serializable` isolation, so the check and the write are atomic with respect to a concurrent
+ * deletion of another workspace shared by the same member.
+ */
+async function findMembersStrandedByArchival(
+  executor: DbOrTx,
+  workspaceId: string,
+  organizationId: string | null
+): Promise<string[]> {
+  const explicitMembers = await executor
+    .selectDistinct({ userId: permissions.userId })
+    .from(permissions)
+    .where(and(eq(permissions.entityId, workspaceId), eq(permissions.entityType, 'workspace')))
+
+  const candidateUserIds = new Set(explicitMembers.map((row) => row.userId))
+
+  if (organizationId) {
+    const orgAdmins = await executor
+      .selectDistinct({ userId: member.userId })
+      .from(member)
+      .where(
+        and(eq(member.organizationId, organizationId), inArray(member.role, [...ORG_ADMIN_ROLES]))
+      )
+    for (const { userId } of orgAdmins) {
+      candidateUserIds.add(userId)
+    }
+  }
+
+  if (candidateUserIds.size === 0) {
+    return []
+  }
+
+  const strandedUserIds: string[] = []
+  for (const userId of candidateUserIds) {
+    const accessible = await listAccessibleWorkspaceRowsForUser(userId, 'active', executor)
+    const hasOtherWorkspace = accessible.some((row) => row.workspace.id !== workspaceId)
+    if (!hasOtherWorkspace) {
+      strandedUserIds.push(userId)
+    }
+  }
+
+  if (strandedUserIds.length === 0) {
+    return []
+  }
+
+  const bannedUserIds = new Set(await getActivelyBannedUserIds(strandedUserIds, executor))
+  return strandedUserIds.filter((userId) => !bannedUserIds.has(userId))
 }
 
 export async function archiveWorkspace(
   workspaceId: string,
   options: ArchiveWorkspaceOptions
-): Promise<{ archived: boolean; workspaceName?: string }> {
+): Promise<ArchiveWorkspaceResult> {
   const workspaceRecord = await getWorkspaceWithOwner(workspaceId, { includeArchived: true })
 
   if (!workspaceRecord) {
@@ -47,7 +139,38 @@ export async function archiveWorkspace(
     .from(workflowMcpServer)
     .where(eq(workflowMcpServer.workspaceId, workspaceId))
 
-  await db.transaction(async (tx) => {
+  // serializable: without it, two concurrent deletions sharing a sole member could each read a
+  // pre-deletion workspace count and both skip provisioning a replacement. Postgres detects this
+  // write skew under serializable isolation and aborts one transaction. Only needed when the
+  // stranded-member check actually runs.
+  const transactionConfig = options.provisionFallbackForStrandedMembers
+    ? ({ isolationLevel: 'serializable' } as const)
+    : undefined
+
+  const provisionedFallbacks = await db.transaction(async (tx) => {
+    const fallbacks: Array<{ userId: string; workspaceId: string; name: string }> = []
+
+    if (options.provisionFallbackForStrandedMembers) {
+      const strandedUserIds = await findMembersStrandedByArchival(
+        tx,
+        workspaceId,
+        workspaceRecord.organizationId
+      )
+      for (const userId of strandedUserIds) {
+        // Intentionally bypasses getWorkspaceCreationPolicy: this is a system-provisioned safety
+        // net (never blocked by "who can create a workspace" rules), not user self-service.
+        const fallbackWorkspace = await createWorkspaceRecord({
+          userId,
+          name: FALLBACK_WORKSPACE_NAME,
+          organizationId: null,
+          workspaceMode: WORKSPACE_MODE.PERSONAL,
+          billedAccountUserId: userId,
+          executor: tx,
+        })
+        fallbacks.push({ userId, workspaceId: fallbackWorkspace.id, name: fallbackWorkspace.name })
+      }
+    }
+
     await tx
       .update(knowledgeBase)
       .set({
@@ -169,11 +292,53 @@ export async function archiveWorkspace(
         updatedAt: now,
       })
       .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt)))
-  })
+
+    return fallbacks
+  }, transactionConfig)
+
+  // Recorded/fired only after the transaction commits — recordAudit and the telemetry event are
+  // fire-and-forget and don't participate in the transaction, so triggering them earlier could
+  // leave a phantom audit entry / event pointing at a fallback workspace that got rolled back
+  // (e.g. on a serialization failure). `createWorkspaceRecord` defers its own `workspaceCreated`
+  // event for exactly this reason when given an `executor` — this is where it gets fired instead.
+  for (const fallback of provisionedFallbacks) {
+    try {
+      PlatformEvents.workspaceCreated({
+        workspaceId: fallback.workspaceId,
+        userId: fallback.userId,
+        name: fallback.name,
+      })
+    } catch {
+      // Telemetry should not fail the operation
+    }
+
+    if (options.actorId) {
+      recordAudit({
+        workspaceId: fallback.workspaceId,
+        actorId: options.actorId,
+        actorName: options.actorName,
+        actorEmail: options.actorEmail,
+        action: AuditAction.WORKSPACE_CREATED,
+        resourceType: AuditResourceType.WORKSPACE,
+        resourceId: fallback.workspaceId,
+        resourceName: fallback.name,
+        description: `Auto-created replacement workspace "${fallback.name}" for a member left with no workspace after deleting "${workspaceRecord.name}"`,
+        metadata: { deletedWorkspaceId: workspaceId, recipientUserId: fallback.userId },
+      })
+    }
+  }
+
+  const provisionedWorkspaceUserIds = provisionedFallbacks.map((fallback) => fallback.userId)
 
   await archiveWorkflowsForWorkspace(workspaceId, options)
 
   logger.info(`[${options.requestId}] Archived workspace ${workspaceId}`)
+  if (provisionedWorkspaceUserIds.length > 0) {
+    logger.info(
+      `[${options.requestId}] Provisioned replacement workspaces for members stranded by archiving ${workspaceId}`,
+      { userIds: provisionedWorkspaceUserIds }
+    )
+  }
 
   await mcpService.clearCache(workspaceId).catch(() => undefined)
 
@@ -189,5 +354,6 @@ export async function archiveWorkspace(
   return {
     archived: true,
     workspaceName: workspaceRecord.name,
+    ...(provisionedWorkspaceUserIds.length > 0 && { provisionedWorkspaceUserIds }),
   }
 }
diff --git a/apps/sim/lib/workspaces/utils.ts b/apps/sim/lib/workspaces/utils.ts
index 4c65d5e5f35..e448f54d69e 100644
--- a/apps/sim/lib/workspaces/utils.ts
+++ b/apps/sim/lib/workspaces/utils.ts
@@ -51,12 +51,16 @@ export async function getWorkspaceBilledAccountUserId(workspaceId: string): Prom
  * Workspaces the user administers purely through organization owner/admin role,
  * with no explicit permission row required. Empty when the user is not an org
  * owner/admin. Implements the workspace-permission inheritance model.
+ *
+ * Accepts an optional executor so callers already inside a transaction (e.g. a
+ * workspace-archival safety check) can run this against `tx` instead of `db`.
  */
 export async function getOrgAdminWorkspaceRows(
   userId: string,
-  scope: WorkspaceScope = 'active'
+  scope: WorkspaceScope = 'active',
+  executor: DbOrTx = db
 ): Promise<Array<typeof workspaceTable.$inferSelect>> {
-  const [membership] = await db
+  const [membership] = await executor
     .select({ organizationId: member.organizationId, role: member.role })
     .from(member)
     .where(eq(member.userId, userId))
@@ -74,20 +78,24 @@ export async function getOrgAdminWorkspaceRows(
         ? and(orgFilter, sql`${workspaceTable.archivedAt} IS NOT NULL`)
         : and(orgFilter, isNull(workspaceTable.archivedAt))
 
-  return db.select().from(workspaceTable).where(where).orderBy(desc(workspaceTable.createdAt))
+  return executor.select().from(workspaceTable).where(where).orderBy(desc(workspaceTable.createdAt))
 }
 
 /**
  * Every workspace a user can access: explicit permission grants plus workspaces
  * derived from organization owner/admin role. Deduped with explicit rows first.
+ *
+ * Accepts an optional executor so callers already inside a transaction can run
+ * this against `tx` instead of `db`.
  */
 export async function listAccessibleWorkspaceRowsForUser(
   userId: string,
-  scope: WorkspaceScope = 'active'
+  scope: WorkspaceScope = 'active',
+  executor: DbOrTx = db
 ): Promise<
   Array<{ workspace: typeof workspaceTable.$inferSelect; permissionType: PermissionType }>
 > {
-  const explicit = await db
+  const explicit = await executor
     .select({ workspace: workspaceTable, permissionType: permissions.permissionType })
     .from(permissions)
     .innerJoin(workspaceTable, eq(permissions.entityId, workspaceTable.id))
@@ -108,7 +116,7 @@ export async function listAccessibleWorkspaceRowsForUser(
     )
     .orderBy(desc(workspaceTable.createdAt))
 
-  const orgRows = await getOrgAdminWorkspaceRows(userId, scope)
+  const orgRows = await getOrgAdminWorkspaceRows(userId, scope, executor)
   if (orgRows.length === 0) {
     return explicit
   }