feat(BA-5797): add effective permissions resolver for entities

Add resolve_effective_permissions across the full stack
(data types → DB source → repository → service/action) to answer
"what operations can user X perform on entities [A, B, C]?"

Uses a single batched query with the existing scope chain CTE,
eliminating N+1 by processing all entity IDs in one DB roundtrip.

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

Author: fregataa

changes/11236.feature.md

@@ -0,0 +1 @@
+Add an effective permissions resolver API that returns all permitted operations a user can perform on given entities by traversing the RBAC scope chain.

src/ai/backend/manager/data/permission/role.py

@@ -94,6 +94,28 @@ class ScopeChainPermissionCheckInput:
     permission_entity_type: EntityType | None
 
 
+@dataclass(frozen=True)
+class EffectivePermissionsInput:
+    """Input for resolving effective permissions per entity for a given user.
+
+    Given a user, an element type, and a list of entity IDs, returns the
+    set of permitted operations per entity by traversing the scope chain
+    and evaluating all role/permission assignments.
+    """
+
+    user_id: uuid.UUID
+    target_element_type: RBACElementType
+    target_entity_ids: list[str]
+    permission_entity_type: EntityType | None = None
+
+
+@dataclass(frozen=True)
+class EffectivePermissionsResult:
+    """Mapping from entity ID to the set of operations the user is authorized to perform."""
+
+    permissions: dict[str, set[OperationType]]
+
+
 @dataclass(frozen=True)
 class BulkPermissionCheckInput:
     user_id: uuid.UUID

src/ai/backend/manager/repositories/permission_controller/db_source/db_source.py

@@ -40,6 +40,8 @@
     BulkRoleRevocationFailure,
     BulkRoleRevocationResultData,
     BulkUserRoleRevocationInput,
+    EffectivePermissionsInput,
+    EffectivePermissionsResult,
     ProjectRoleCount,
     RoleListResult,
     RolePermissionsUpdateInput,
@@ -1088,6 +1090,88 @@ async def check_bulk_permission_with_scope_chain(
         )
         return {eid: eid in granted for eid in data.target_entity_ids}
 
+    async def resolve_effective_permissions(
+        self,
+        data: EffectivePermissionsInput,
+    ) -> EffectivePermissionsResult:
+        """Resolve the effective permissions for a user across multiple entities.
+
+        Uses a single batched query that traverses the scope chain (AUTO edges)
+        and self-scope permissions to collect all operations the user can perform
+        on each entity.
+
+        Returns a mapping from entity ID to the set of permitted operations.
+        """
+        if not data.target_entity_ids:
+            return EffectivePermissionsResult(permissions={})
+
+        association_entity_type = data.target_element_type.to_entity_type()
+        permission_entity_type = data.permission_entity_type or association_entity_type
+        target_scope_type = data.target_element_type.to_scope_type()
+
+        perm = PermissionRow.__table__
+        user_roles = UserRoleRow.__table__
+        roles = RoleRow.__table__
+
+        scope_chain_cte = self._build_scope_chain_cte(
+            association_entity_type, data.target_entity_ids
+        )
+        scope_chain_query = (
+            sa.select(
+                scope_chain_cte.c.entity_id,
+                perm.c.operation,
+            )
+            .select_from(
+                scope_chain_cte.join(
+                    perm,
+                    sa.and_(
+                        perm.c.scope_type == scope_chain_cte.c.scope_type,
+                        perm.c.scope_id == scope_chain_cte.c.scope_id,
+                    ),
+                )
+                .join(roles, roles.c.id == perm.c.role_id)
+                .join(user_roles, user_roles.c.role_id == roles.c.id)
+            )
+            .where(
+                sa.and_(
+                    user_roles.c.user_id == data.user_id,
+                    roles.c.status == RoleStatus.ACTIVE,
+                    perm.c.entity_type == permission_entity_type,
+                )
+            )
+        )
+
+        self_scope_query = (
+            sa.select(
+                perm.c.scope_id.label("entity_id"),
+                perm.c.operation,
+            )
+            .select_from(
+                perm.join(roles, roles.c.id == perm.c.role_id).join(
+                    user_roles, user_roles.c.role_id == roles.c.id
+                )
+            )
+            .where(
+                sa.and_(
+                    user_roles.c.user_id == data.user_id,
+                    roles.c.status == RoleStatus.ACTIVE,
+                    perm.c.scope_type == target_scope_type,
+                    perm.c.scope_id.in_(data.target_entity_ids),
+                    perm.c.entity_type == permission_entity_type,
+                )
+            )
+        )
+
+        combined_query = sa.union_all(scope_chain_query, self_scope_query)
+
+        permissions: dict[str, set[OperationType]] = {eid: set() for eid in data.target_entity_ids}
+        async with self._db.begin_readonly_session_read_committed() as db_session:
+            result = await db_session.execute(combined_query)
+            for row in result:
+                permissions[row.entity_id].add(OperationType(row.operation))
+
+        return EffectivePermissionsResult(permissions=permissions)
+
     async def bulk_assign_role(
         self, bulk_creator: BulkCreator[UserRoleRow]
     ) -> BulkCreatorResultWithFailures[UserRoleRow]:

src/ai/backend/manager/repositories/permission_controller/repository.py

@@ -31,6 +31,8 @@
     BulkRoleAssignmentResultData,
     BulkRoleRevocationResultData,
     BulkUserRoleRevocationInput,
+    EffectivePermissionsInput,
+    EffectivePermissionsResult,
     RoleData,
     RoleDetailData,
     RoleListResult,
@@ -363,6 +365,18 @@ async def check_bulk_permission_with_scope_chain(
         """
         return await self._db_source.check_bulk_permission_with_scope_chain(data)
 
+    @permission_controller_repository_resilience.apply()
+    async def resolve_effective_permissions(
+        self,
+        data: EffectivePermissionsInput,
+    ) -> EffectivePermissionsResult:
+        """Resolve the set of permitted operations per entity for a given user.
+
+        For each target entity, traverses the scope chain (AUTO edges) and
+        self-scope permissions to collect all operations the user can perform.
+        """
+        return await self._db_source.resolve_effective_permissions(data)
+
     # -- role invitation --
 
     @permission_controller_repository_resilience.apply()

src/ai/backend/manager/services/permission_contoller/actions/__init__.py

@@ -7,6 +7,10 @@
 from .get_permission_matrix import GetPermissionMatrixAction, GetPermissionMatrixActionResult
 from .get_role_detail import GetRoleDetailAction, GetRoleDetailActionResult
 from .purge_role import PurgeRoleAction, PurgeRoleActionResult
+from .resolve_effective_permissions import (
+    ResolveEffectivePermissionsAction,
+    ResolveEffectivePermissionsActionResult,
+)
 from .revoke_role import RevokeRoleAction, RevokeRoleActionResult
 from .search_permissions import (
     SearchPermissionsAction,
@@ -47,6 +51,8 @@
     "GetRoleDetailActionResult",
     "PurgeRoleAction",
     "PurgeRoleActionResult",
+    "ResolveEffectivePermissionsAction",
+    "ResolveEffectivePermissionsActionResult",
     "RevokeRoleAction",
     "RevokeRoleActionResult",
     "SearchRolesAction",

src/ai/backend/manager/services/permission_contoller/actions/resolve_effective_permissions.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import override
+from uuid import UUID
+
+from ai.backend.common.data.permission.types import EntityType, OperationType, RBACElementType
+from ai.backend.manager.actions.action import BaseActionResult
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.services.permission_contoller.actions.base import RoleAction
+
+
+@dataclass
+class ResolveEffectivePermissionsAction(RoleAction):
+    """Action to resolve effective permissions per entity for a given user.
+
+    Given a user ID, an element type, and a list of entity IDs, returns the
+    set of permitted operations per entity by traversing the scope chain and
+    evaluating all role/permission assignments.
+    """
+
+    user_id: UUID
+    target_element_type: RBACElementType
+    target_entity_ids: list[str]
+    permission_entity_type: EntityType | None = None
+
+    @override
+    def entity_id(self) -> str | None:
+        return str(self.user_id)
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.GET
+
+
+@dataclass
+class ResolveEffectivePermissionsActionResult(BaseActionResult):
+    """Result containing the effective permissions per entity."""
+
+    permissions: dict[str, set[OperationType]] = field(default_factory=dict)
+
+    @override
+    def entity_id(self) -> str | None:
+        return None

src/ai/backend/manager/services/permission_contoller/service.py

@@ -17,6 +17,7 @@
     RejectRoleInvitationAction,
 )
 from ai.backend.manager.data.permission.role import (
+    EffectivePermissionsInput,
     UserRoleRevocationData,
 )
 from ai.backend.manager.data.role_invitation.types import RoleInvitationData
@@ -74,6 +75,10 @@
     PurgeRoleAction,
     PurgeRoleActionResult,
 )
+from ai.backend.manager.services.permission_contoller.actions.resolve_effective_permissions import (
+    ResolveEffectivePermissionsAction,
+    ResolveEffectivePermissionsActionResult,
+)
 from ai.backend.manager.services.permission_contoller.actions.revoke_role import (
     RevokeRoleAction,
     RevokeRoleActionResult,
@@ -360,6 +365,24 @@ async def get_permission_matrix(
             actions[action_cls.action_name()] = perm
         return GetPermissionMatrixActionResult(matrix=result)
 
+    async def resolve_effective_permissions(
+        self, action: ResolveEffectivePermissionsAction
+    ) -> ResolveEffectivePermissionsActionResult:
+        """Resolve the set of permitted operations per entity for a given user.
+
+        Traverses the scope chain and evaluates all role/permission assignments
+        to return all operations the user is authorized to perform on each entity.
+        """
+        result = await self._repository.resolve_effective_permissions(
+            EffectivePermissionsInput(
+                user_id=action.user_id,
+                target_element_type=action.target_element_type,
+                target_entity_ids=action.target_entity_ids,
+                permission_entity_type=action.permission_entity_type,
+            )
+        )
+        return ResolveEffectivePermissionsActionResult(permissions=result.permissions)
+
     async def create_role_invitation_by_email(
         self, action: CreateRoleInvitationByEmailAction
     ) -> CreateRoleInvitationResult: