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

@@ -10,13 +10,6 @@
 from ai.backend.common.resilience.policies.metrics import MetricArgs, MetricPolicy
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
-from ai.backend.manager.actions.action.rbac_role_invitation import (
-    AcceptRoleInvitationAction,
-    CancelRoleInvitationAction,
-    CreateRoleInvitationByEmailAction,
-    CreateRoleInvitationResult,
-    RejectRoleInvitationAction,
-)
 from ai.backend.manager.data.permission.entity import ElementAssociationListResult, EntityListResult
 from ai.backend.manager.data.permission.id import ObjectId
 from ai.backend.manager.data.permission.permission import (
@@ -26,11 +19,12 @@
 from ai.backend.manager.data.permission.role import (
     AssignedUserListResult,
     BatchEntityPermissionCheckInput,
-    BulkPermissionCheckInput,
     BulkRoleAssignmentFailure,
     BulkRoleAssignmentResultData,
     BulkRoleRevocationResultData,
     BulkUserRoleRevocationInput,
+    EffectivePermissionsInput,
+    EffectivePermissionsResult,
     RoleData,
     RoleDetailData,
     RoleListResult,
@@ -46,7 +40,6 @@
 from ai.backend.manager.data.permission.types import (
     ScopeListResult,
 )
-from ai.backend.manager.data.role_invitation.types import RoleInvitationData
 from ai.backend.manager.models.rbac_models.permission.permission import PermissionRow
 from ai.backend.manager.models.rbac_models.role import RoleRow
 from ai.backend.manager.models.rbac_models.user_role import UserRoleRow
@@ -60,11 +53,6 @@
     PermissionSearchScope,
     ScopedRoleSearchScope,
 )
-from ai.backend.manager.repositories.role_invitation.types import (
-    InviteeSearchScope,
-    RoleInvitationSearchResult,
-    RoleInvitationSearchScope,
-)
 
 from .db_source.db_source import CreateRoleInput, PermissionDBSource
 
@@ -352,59 +340,13 @@ async def check_permission_with_scope_chain(
         return await self._db_source.check_permission_with_scope_chain(data)
 
     @permission_controller_repository_resilience.apply()
-    async def check_bulk_permission_with_scope_chain(
+    async def resolve_effective_permissions(
         self,
-        data: BulkPermissionCheckInput,
-    ) -> dict[str, bool]:
-        """Batch permission check that traverses the scope chain via AUTO edges.
+        data: EffectivePermissionsInput,
+    ) -> EffectivePermissionsResult:
+        """Resolve the set of permitted operations per entity for a given user.
 
-        Same semantics as check_permission_with_scope_chain but for multiple
-        entities of the same RBACElementType in a single query.
+        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.check_bulk_permission_with_scope_chain(data)
-
-    # -- role invitation --
-
-    @permission_controller_repository_resilience.apply()
-    async def create_invitation_by_email(
-        self,
-        action: CreateRoleInvitationByEmailAction,
-    ) -> CreateRoleInvitationResult:
-        return await self._db_source.create_invitation_by_email(action)
-
-    @permission_controller_repository_resilience.apply()
-    async def search_invitations_by_invitee(
-        self,
-        querier: BatchQuerier,
-        scope: InviteeSearchScope,
-    ) -> RoleInvitationSearchResult:
-        return await self._db_source.search_invitations_by_invitee(querier, scope)
-
-    @permission_controller_repository_resilience.apply()
-    async def search_invitations_by_role(
-        self,
-        querier: BatchQuerier,
-        scope: RoleInvitationSearchScope,
-    ) -> RoleInvitationSearchResult:
-        return await self._db_source.search_invitations_by_role(querier, scope)
-
-    @permission_controller_repository_resilience.apply()
-    async def accept_invitation(
-        self,
-        action: AcceptRoleInvitationAction,
-    ) -> RoleInvitationData:
-        return await self._db_source.accept_invitation(action)
-
-    @permission_controller_repository_resilience.apply()
-    async def reject_invitation(
-        self,
-        action: RejectRoleInvitationAction,
-    ) -> RoleInvitationData:
-        return await self._db_source.reject_invitation(action)
-
-    @permission_controller_repository_resilience.apply()
-    async def cancel_invitation(
-        self,
-        action: CancelRoleInvitationAction,
-    ) -> RoleInvitationData:
-        return await self._db_source.cancel_invitation(action)
+        return await self._db_source.resolve_effective_permissions(data)

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

@@ -9,17 +9,10 @@
     RBACActionName,
     RBACRequiredPermission,
 )
-from ai.backend.manager.actions.action.rbac_role_invitation import (
-    AcceptRoleInvitationAction,
-    CancelRoleInvitationAction,
-    CreateRoleInvitationByEmailAction,
-    CreateRoleInvitationResult,
-    RejectRoleInvitationAction,
-)
 from ai.backend.manager.data.permission.role import (
+    EffectivePermissionsInput,
     UserRoleRevocationData,
 )
-from ai.backend.manager.data.role_invitation.types import RoleInvitationData
 from ai.backend.manager.repositories.group.repository import GroupRepository
 from ai.backend.manager.repositories.permission_controller.creators import UserRoleCreatorSpec
 from ai.backend.manager.repositories.permission_controller.db_source.db_source import (
@@ -74,6 +67,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,30 +357,20 @@ async def get_permission_matrix(
             actions[action_cls.action_name()] = perm
         return GetPermissionMatrixActionResult(matrix=result)
 
-    async def create_role_invitation_by_email(
-        self, action: CreateRoleInvitationByEmailAction
-    ) -> CreateRoleInvitationResult:
-        """Create role invitations by resolving invitee emails."""
-        return await self._repository.create_invitation_by_email(action)
-
-    async def accept_role_invitation(
-        self, action: AcceptRoleInvitationAction
-    ) -> RoleInvitationData:
-        """Accept a PENDING invitation and assign the role.
+    async def resolve_effective_permissions(
+        self, action: ResolveEffectivePermissionsAction
+    ) -> ResolveEffectivePermissionsActionResult:
+        """Resolve the set of permitted operations per entity for a given user.
 
-        State transition and role assignment happen atomically
-        in a single DB session within the repository.
+        Traverses the scope chain and evaluates all role/permission assignments
+        to return all operations the user is authorized to perform on each entity.
         """
-        return await self._repository.accept_invitation(action)
-
-    async def reject_role_invitation(
-        self, action: RejectRoleInvitationAction
-    ) -> RoleInvitationData:
-        """Reject a PENDING invitation."""
-        return await self._repository.reject_invitation(action)
-
-    async def cancel_role_invitation(
-        self, action: CancelRoleInvitationAction
-    ) -> RoleInvitationData:
-        """Cancel a PENDING invitation."""
-        return await self._repository.cancel_invitation(action)
+        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)