refactor(BA-5765): unify purge actions into RBAC-enforced PurgeVFolderV2Action

Merge PurgeVFolderV2RBACAction into PurgeVFolderV2Action by switching
its base to VFolderSingleEntityAction and removing user_id field.
The service method now uses current_user() for host permission checks
instead of receiving user_id through the action. Drop the redundant
purge_v2_rbac service/processor path and vfolder_v2_rbac.py module.

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

Author: fregataa

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

@@ -412,8 +412,8 @@ async def restore(self, vfolder_id: UUID) -> RestoreVFolderPayload:
 
     async def purge(self, vfolder_id: UUID) -> PurgeVFolderPayload:
         """Permanently delete a vfolder. RBAC enforced."""
-        action = PurgeVFolderV2RBACAction(vfolder_id=vfolder_id)
-        await self._processors.vfolder.purge_v2_rbac.wait_for_complete(action)
+        action = PurgeVFolderV2Action(vfolder_id=vfolder_id)
+        await self._processors.vfolder.purge_v2.wait_for_complete(action)
         return PurgeVFolderPayload(id=vfolder_id)
 
     async def deploy(
@@ -504,11 +504,8 @@ async def bulk_delete(self, input: BulkDeleteVFoldersInput) -> BulkDeleteVFolder
 
     async def bulk_purge(self, input: BulkPurgeVFoldersInput) -> BulkPurgeVFoldersPayload:
         """Permanently purge multiple vfolders."""
-        me = current_user()
-        if me is None:
-            raise UnreachableError("User context is not available")
         for vfolder_id in input.ids:
-            action = PurgeVFolderV2Action(user_id=me.user_id, vfolder_id=vfolder_id)
+            action = PurgeVFolderV2Action(vfolder_id=vfolder_id)
             await self._processors.vfolder.purge_v2.wait_for_complete(action)
         return BulkPurgeVFoldersPayload(purged_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,6 +13,10 @@
 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
@@ -60,32 +64,24 @@ def entity_id(self) -> str | None:
 
 
 @dataclass
-class PurgeVFolderV2Action(BaseScopeAction):
-    user_id: uuid.UUID
+class PurgeVFolderV2Action(VFolderSingleEntityAction):
+    """Permanently purge 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.PURGE
 
     @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(
@@ -95,9 +91,13 @@ def target_element(self) -> RBACElementRef:
 
 
 @dataclass
-class PurgeVFolderV2ActionResult(BaseActionResult):
+class PurgeVFolderV2ActionResult(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)

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

@@ -165,7 +165,7 @@ class VFolderProcessors(AbstractProcessorPackage):
         CreateUploadSessionV2Action, CreateUploadSessionV2ActionResult
     ]
     delete_v2: ActionProcessor[DeleteVFolderV2Action, DeleteVFolderV2ActionResult]
-    purge_v2: ActionProcessor[PurgeVFolderV2Action, PurgeVFolderV2ActionResult]
+    purge_v2: SingleEntityActionProcessor[PurgeVFolderV2Action, PurgeVFolderV2ActionResult]
     clone_v2: ActionProcessor[CloneVFolderV2Action, CloneVFolderV2ActionResult]
     delete_v2_rbac: SingleEntityActionProcessor[
         DeleteVFolderV2RBACAction, DeleteVFolderV2RBACActionResult
@@ -262,7 +262,9 @@ def __init__(
             service.create_upload_session_v2, action_monitors
         )
         self.delete_v2 = ActionProcessor(service.delete_v2, action_monitors)
-        self.purge_v2 = ActionProcessor(service.purge_v2, action_monitors)
+        self.purge_v2 = SingleEntityActionProcessor(
+            service.purge_v2, action_monitors, validators=single_entity_rbac_validators
+        )
         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

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

@@ -16,6 +16,7 @@
 
 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.types import (
@@ -1649,20 +1650,17 @@ async def delete_v2(self, action: DeleteVFolderV2Action) -> DeleteVFolderV2Actio
         return DeleteVFolderV2ActionResult(vfolder_id=action.vfolder_id)
 
     async def purge_v2(self, action: PurgeVFolderV2Action) -> PurgeVFolderV2ActionResult:
-        """Purge a vfolder permanently (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
-        )
-
-        # Host permission check — resolved from user_id
-        await self._vfolder_repository.ensure_host_permission_allowed_by_user(
-            vfolder_data.host,
-            permission=VFolderHostPermission.DELETE,
-            user_uuid=action.user_id,
-        )
+        """Permanently purge a vfolder by ID. RBAC enforced at processor level."""
+        me = current_user()
+        vfolder_data = await self._vfolder_repository.get_by_id(action.vfolder_id)
+
+        # Host permission check — resolved from current user context
+        if me is not None:
+            await self._vfolder_repository.ensure_host_permission_allowed_by_user(
+                vfolder_data.host,
+                permission=VFolderHostPermission.DELETE,
+                user_uuid=me.user_id,
+            )
 
         await self._vfolder_repository.delete_vfolders_forever([action.vfolder_id])
         await self._remove_vfolder_from_storage(vfolder_data)