feat(BA-5681): auto-sync user-scope entries on role assign/unassign

[fregataa]

Summary

• Auto-sync user-scope membership entries in association_scopes_entities when roles are assigned or revoked
• On assign: INSERT user-scope entries for all scopes bound to the role (ON CONFLICT DO NOTHING)
• On revoke: DELETE user-scope entries only when no other role covers the same scope (correlated NOT EXISTS check)
• Scope-type-agnostic — works for project, domain, and any future scope types
• Integrated into assign_role(), revoke_role(), bulk_assign_role(), and bulk_revoke_role()

Test plan

• Unit tests: assign creates user-scope entries (single scope, multi-scope, no-scope, idempotent)
• Unit tests: revoke removes entries (only role, overlapping roles, multi-scope partial, no-scope)
• Existing role assignment tests still pass

Resolves BA-5681

[Copilot]

Pull request overview

Adds automatic synchronization of scope→user membership edges in association_scopes_entities when roles are assigned or revoked, so RBAC scope membership stays consistent with role assignments.

Changes:

• Add sync_user_scope_on_assign() / sync_user_scope_on_revoke() helpers to maintain scope→user edges based on role-scope bindings.
• Integrate the sync logic into single and bulk role assign/revoke flows.
• Add unit tests validating scope entry creation/removal behavior for assign/revoke scenarios.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

File	Description
src/ai/backend/manager/repositories/permission_controller/db_source/db_source.py	Adds scope-sync helpers and wires them into assign/revoke (including bulk) operations.
tests/unit/manager/repositories/permission_controller/test_user_scope_sync.py	New unit tests covering user-scope sync behavior for role assign/revoke cases.

Comments suppressed due to low confidence (1)

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

• The new user-scope sync behavior is integrated into bulk_assign_role() / bulk_revoke_role(), but the added unit tests only cover the single-user assign_role() / revoke_role() paths. Adding coverage for the bulk paths would help prevent regressions (e.g., partial-success batches, idempotency, and ensuring scope entries are synced for each successful row).

    async def bulk_assign_role(
        self, bulk_creator: BulkCreator[UserRoleRow]
    ) -> BulkCreatorResultWithFailures[UserRoleRow]:
        async with self._db.begin_session() as db_session:
            result = await execute_bulk_creator_partial(db_session, bulk_creator)
            for row in result.successes:
                await sync_user_scope_on_assign(
                    db_session, user_id=row.user_id, role_id=row.role_id
                )
            return result

    async def bulk_revoke_role(
        self, data: BulkUserRoleRevocationInput
    ) -> BulkRoleRevocationResultData:
        successes: list[UserRoleRevocationData] = []
        failures: list[BulkRoleRevocationFailure] = []

        async with self._db.begin_session() as db_session:
            for user_id in data.user_ids:
                try:
                    async with db_session.begin_nested():
                        stmt = (
                            sa.select(UserRoleRow)
                            .where(UserRoleRow.user_id == user_id)
                            .where(UserRoleRow.role_id == data.role_id)
                        )
                        user_role_row = await db_session.scalar(stmt)
                        if user_role_row is None:
                            raise RoleNotAssigned(
                                f"Role {data.role_id} is not assigned to user {user_id}."
                            )
                        user_role_id = user_role_row.id
                        await db_session.delete(user_role_row)
                        await db_session.flush()
                        await sync_user_scope_on_revoke(
                            db_session, user_id=user_id, role_id=data.role_id
                        )
                        successes.append(
                            UserRoleRevocationData(
                                user_role_id=user_role_id,
                                user_id=user_id,
                                role_id=data.role_id,
                            )
                        )
                except Exception as e:
                    log.warning(
                        "Failed to revoke role {} from user {}: {}",
                        data.role_id,
                        user_id,
                        str(e),
                    )
                    failures.append(BulkRoleRevocationFailure(user_id=user_id, message=str(e)))

        return BulkRoleRevocationResultData(successes=successes, failures=failures)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.