refactor(BA-5795): unify delete actions into RBAC-enforced DeleteVFolderV2Action

Remove the separate delete_v2_rbac path and convert the existing
delete_v2 action to use SingleEntityActionProcessor with RBAC
validators, following the same pattern applied to purge_v2 in BA-5765.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: fregataa

src/ai/backend/manager/api/adapters/vfolder.py

@@ -94,12 +94,10 @@
     combine_conditions_or,
     negate_conditions,
 )
-from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.repositories.vfolder.types import (
     ProjectVFolderSearchScope,
     UserVFolderSearchScope,
 )
-from ai.backend.manager.repositories.vfolder.updaters import VFolderTrashUpdaterSpec
 from ai.backend.manager.services.deployment.actions.create_deployment import CreateDeploymentAction
 from ai.backend.manager.services.vfolder.actions.admin_search_vfolders import (
     AdminSearchVFoldersAction,
@@ -133,9 +131,6 @@
     DeleteVFolderV2Action,
     PurgeVFolderV2Action,
 )
-from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
-    DeleteVFolderV2RBACAction,
-)
 
 from .base import BaseAdapter
 
@@ -393,9 +388,8 @@ async def get(self, vfolder_id: UUID) -> VFolderNode:
 
     async def delete(self, vfolder_id: UUID) -> DeleteVFolderPayload:
         """Soft-delete a vfolder (move to trash). RBAC enforced."""
-        updater = Updater(spec=VFolderTrashUpdaterSpec(), pk_value=vfolder_id)
-        action = DeleteVFolderV2RBACAction(vfolder_id=vfolder_id, updater=updater)
-        await self._processors.vfolder.delete_v2_rbac.wait_for_complete(action)
+        action = DeleteVFolderV2Action(vfolder_id=vfolder_id)
+        await self._processors.vfolder.delete_v2.wait_for_complete(action)
         return DeleteVFolderPayload(id=vfolder_id)
 
     async def restore(self, vfolder_id: UUID) -> RestoreVFolderPayload:
@@ -497,11 +491,8 @@ async def deploy(
 
     async def bulk_delete(self, input: BulkDeleteVFoldersInput) -> BulkDeleteVFoldersPayload:
         """Soft-delete multiple vfolders."""
-        me = current_user()
-        if me is None:
-            raise UnreachableError("User context is not available")
         for vfolder_id in input.ids:
-            action = DeleteVFolderV2Action(user_id=me.user_id, vfolder_id=vfolder_id)
+            action = DeleteVFolderV2Action(vfolder_id=vfolder_id)
             await self._processors.vfolder.delete_v2.wait_for_complete(action)
         return BulkDeleteVFoldersPayload(deleted_count=len(input.ids))
 

src/ai/backend/manager/services/vfolder/actions/vfolder_v2.py

@@ -1,4 +1,4 @@
-"""V2 vfolder actions — user_id based, no keypair_resource_policy."""
+"""V2 vfolder actions — RBAC-enforced via SingleEntityActionProcessor."""
 
 import uuid
 from dataclasses import dataclass
@@ -13,35 +13,31 @@
 from ai.backend.manager.actions.action.scope import BaseScopeAction
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.permission.types import RBACElementRef
+from ai.backend.manager.services.vfolder.actions.base import (
+    VFolderSingleEntityAction,
+    VFolderSingleEntityActionResult,
+)
 
 
 @dataclass
-class DeleteVFolderV2Action(BaseScopeAction):
-    user_id: uuid.UUID
+class DeleteVFolderV2Action(VFolderSingleEntityAction):
+    """Soft-delete a vfolder by ID with RBAC enforcement."""
+
     vfolder_id: uuid.UUID
 
     @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.VFOLDER
+    def entity_id(self) -> str | None:
+        return str(self.vfolder_id)
 
     @override
     @classmethod
     def operation_type(cls) -> ActionOperationType:
         return ActionOperationType.DELETE
 
     @override
-    def entity_id(self) -> str | None:
+    def target_entity_id(self) -> str:
         return str(self.vfolder_id)
 
-    @override
-    def scope_type(self) -> ScopeType:
-        return ScopeType.USER
-
-    @override
-    def scope_id(self) -> str:
-        return str(self.user_id)
-
     @override
     def target_element(self) -> RBACElementRef:
         return RBACElementRef(
@@ -51,13 +47,17 @@ def target_element(self) -> RBACElementRef:
 
 
 @dataclass
-class DeleteVFolderV2ActionResult(BaseActionResult):
+class DeleteVFolderV2ActionResult(VFolderSingleEntityActionResult):
     vfolder_id: uuid.UUID
 
     @override
     def entity_id(self) -> str | None:
         return str(self.vfolder_id)
 
+    @override
+    def target_entity_id(self) -> str:
+        return str(self.vfolder_id)
+
 
 @dataclass
 class PurgeVFolderV2Action(BaseScopeAction):

src/ai/backend/manager/services/vfolder/actions/vfolder_v2_rbac.py

@@ -98,10 +98,6 @@
     PurgeVFolderV2Action,
     PurgeVFolderV2ActionResult,
 )
-from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
-    DeleteVFolderV2RBACAction,
-    DeleteVFolderV2RBACActionResult,
-)
 from ai.backend.manager.services.vfolder.services.vfolder import VFolderService
 
 
@@ -164,12 +160,9 @@ class VFolderProcessors(AbstractProcessorPackage):
     create_upload_session_v2: ActionProcessor[
         CreateUploadSessionV2Action, CreateUploadSessionV2ActionResult
     ]
-    delete_v2: ActionProcessor[DeleteVFolderV2Action, DeleteVFolderV2ActionResult]
+    delete_v2: SingleEntityActionProcessor[DeleteVFolderV2Action, DeleteVFolderV2ActionResult]
     purge_v2: ActionProcessor[PurgeVFolderV2Action, PurgeVFolderV2ActionResult]
     clone_v2: ActionProcessor[CloneVFolderV2Action, CloneVFolderV2ActionResult]
-    delete_v2_rbac: SingleEntityActionProcessor[
-        DeleteVFolderV2RBACAction, DeleteVFolderV2RBACActionResult
-    ]
 
     def __init__(
         self,
@@ -261,12 +254,11 @@ def __init__(
         self.create_upload_session_v2 = ActionProcessor(
             service.create_upload_session_v2, action_monitors
         )
-        self.delete_v2 = ActionProcessor(service.delete_v2, action_monitors)
+        self.delete_v2 = SingleEntityActionProcessor(
+            service.delete_v2, action_monitors, validators=single_entity_rbac_validators
+        )
         self.purge_v2 = ActionProcessor(service.purge_v2, action_monitors)
         self.clone_v2 = ActionProcessor(service.clone_v2, action_monitors)
-        self.delete_v2_rbac = SingleEntityActionProcessor(
-            service.delete_v2_rbac, action_monitors, validators=single_entity_rbac_validators
-        )
 
     @override
     def supported_actions(self) -> list[ActionSpec]:
@@ -306,5 +298,4 @@ def supported_actions(self) -> list[ActionSpec]:
             DeleteVFolderV2Action.spec(),
             PurgeVFolderV2Action.spec(),
             CloneVFolderV2Action.spec(),
-            DeleteVFolderV2RBACAction.spec(),
         ]

src/ai/backend/manager/services/vfolder/processors/vfolder.py

@@ -16,8 +16,10 @@
 
 from ai.backend.common.bgtask.bgtask import BackgroundTaskManager
 from ai.backend.common.clients.valkey_client.valkey_stat.client import ValkeyStatClient
+from ai.backend.common.contexts.user import current_user
 from ai.backend.common.defs import VFOLDER_GROUP_PERMISSION_MODE
 from ai.backend.common.etcd import AsyncEtcd
+from ai.backend.common.exception import UnreachableError
 from ai.backend.common.types import (
     QuotaScopeID,
     QuotaScopeType,
@@ -162,10 +164,6 @@
     PurgeVFolderV2Action,
     PurgeVFolderV2ActionResult,
 )
-from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
-    DeleteVFolderV2RBACAction,
-    DeleteVFolderV2RBACActionResult,
-)
 from ai.backend.manager.services.vfolder.types import (
     VFolderBaseInfo,
     VFolderOwnershipInfo,
@@ -1630,19 +1628,17 @@ async def get_v2(self, action: GetVFolderV2Action) -> GetVFolderV2ActionResult:
         return GetVFolderV2ActionResult(vfolder=vfolder_data)
 
     async def delete_v2(self, action: DeleteVFolderV2Action) -> DeleteVFolderV2ActionResult:
-        """Delete (trash) a vfolder (v2). Resolves policy internally from user_id."""
-        user = await self._user_repository.get_user_by_uuid(action.user_id)
-        if not user.domain_name:
-            raise VFolderInvalidParameter("User has no domain assigned")
-        vfolder_data = await self._vfolder_repository.get_by_id_validated(
-            action.vfolder_id, user.id, user.domain_name
-        )
+        """Soft-delete a vfolder by ID. RBAC enforced at processor level."""
+        me = current_user()
+        if me is None:
+            raise UnreachableError("User context is not available")
+        vfolder_data = await self._vfolder_repository.get_by_id(action.vfolder_id)
 
-        # Host permission check — resolved from user_id
+        # Host permission check — resolved from current user context
         await self._vfolder_repository.ensure_host_permission_allowed_by_user(
             vfolder_data.host,
             permission=VFolderHostPermission.DELETE,
-            user_uuid=action.user_id,
+            user_uuid=me.user_id,
         )
 
         await self._vfolder_repository.move_vfolders_to_trash([vfolder_data.id])
@@ -1668,13 +1664,6 @@ async def purge_v2(self, action: PurgeVFolderV2Action) -> PurgeVFolderV2ActionRe
         await self._remove_vfolder_from_storage(vfolder_data)
         return PurgeVFolderV2ActionResult(vfolder_id=action.vfolder_id)
 
-    async def delete_v2_rbac(
-        self, action: DeleteVFolderV2RBACAction
-    ) -> DeleteVFolderV2RBACActionResult:
-        """Soft-delete a vfolder by ID. RBAC enforced at processor level."""
-        await self._vfolder_repository.trash_vfolder(action.updater)
-        return DeleteVFolderV2RBACActionResult(vfolder_id=action.vfolder_id)
-
     async def clone_v2(self, action: CloneVFolderV2Action) -> CloneVFolderV2ActionResult:
         """Clone a vfolder (v2). Resolves policy internally from user_id."""
         allowed_vfolder_types = (