fix: deduplicate user access lookups per request

src/backend/bisheng/user/api/user.py

@@ -178,6 +178,8 @@ async def get_info(login_user: LoginUser = Depends(LoginUser.get_login_user)):
     if not db_user:
         raise NotFoundError()
 
+    db_user_roles = await UserRoleDao.aget_user_roles(user_id)
+    role_ids = [user_role.role_id for user_role in db_user_roles]
     admin_group = await UserGroupDao.aget_user_admin_group(user_id)
     admin_group = [one.group_id for one in admin_group]
     dept_admin_depts = await DepartmentDao.aget_user_admin_departments(user_id)
@@ -185,8 +187,12 @@ async def get_info(login_user: LoginUser = Depends(LoginUser.get_login_user)):
     role, web_menu = await login_user.get_roles_web_menu(
         db_user,
         is_department_admin=is_department_admin,
+        role_ids=role_ids,
+    )
+    menu_approval_mode_workbench, menu_approval_mode_admin = await login_user.compute_menu_approval_modes(
+        db_user,
+        role_ids=role_ids,
     )
-    menu_approval_mode_workbench, menu_approval_mode_admin = await login_user.compute_menu_approval_modes(db_user)
     # Legacy union flag, kept for back-compat clients during transition.
     menu_approval_mode = menu_approval_mode_workbench or menu_approval_mode_admin
 
@@ -206,7 +212,7 @@ async def get_info(login_user: LoginUser = Depends(LoginUser.get_login_user)):
         from bisheng.utils.http_middleware import _check_is_global_super
 
         is_global_super, active = await asyncio.gather(
-            _check_is_global_super(user_id),
+            _check_is_global_super(user_id, role_ids=role_ids),
             UserTenantDao.aget_active_user_tenant(user_id),
         )
         leaf_tenant_id = active.tenant_id if active else ROOT_TENANT_ID
@@ -226,7 +232,11 @@ async def get_info(login_user: LoginUser = Depends(LoginUser.get_login_user)):
     # PRD §4.5: Child Admin manages own tenant's user groups → enable tab.
     can_manage_user_groups = bool(login_user.is_admin() or is_department_admin or is_child_admin)
 
-    entry = await LoginUser.user_entry_payload_for_read(db_user)
+    entry = await LoginUser.user_entry_payload_for_read(
+        db_user,
+        role_ids=role_ids,
+        is_department_admin=is_department_admin,
+    )
     return resp_200(
         await UserService.build_user_read(
             db_user,

src/backend/bisheng/user/domain/services/auth.py

@@ -246,16 +246,16 @@ class LoginUser(BaseModel):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
+        provided_user_role = kwargs.get("user_role")
         self.user_id = kwargs.get("user_id")
         self.user_name = kwargs.get("user_name")
-        self.user_role = kwargs.get("user_role")
+        self.user_role = provided_user_role or []
         self.group_cache = kwargs.get("group_cache", {})
         self.tenant_id = kwargs.get("tenant_id", DEFAULT_TENANT_ID)
         self.is_global_super = bool(kwargs.get("is_global_super", False))
         # token_version is validated by Pydantic Field(default=0) above.
 
-        if not self.user_role:
-            self.user_role = []
+        if provided_user_role is None:
             user_role = UserRoleDao.get_user_roles(self.user_id)
             self.user_role = [user_role.role_id for user_role in user_role]
 
@@ -568,6 +568,7 @@ class LoginUser(BaseModel):
         user_name: str,
         tenant_id: int = None,
         token_version: int = 0,
+        role_ids: Iterable[int] | None = None,
     ) -> Self:
         # Pre-resolve global-super so downstream gates (mount/unmount/
         # migrate, also serialised to /user/info & login response) read
@@ -580,16 +581,16 @@ class LoginUser(BaseModel):
         from bisheng.core.context.tenant import bypass_tenant_filter
         from bisheng.utils.http_middleware import _check_is_global_super
 
+        resolved_role_ids = list(role_ids) if role_ids is not None else None
         with bypass_tenant_filter():
-            user_roles, is_global_super = await asyncio.gather(
-                UserRoleDao.aget_user_roles(user_id),
-                _check_is_global_super(user_id),
-            )
-        role_ids = [user_role.role_id for user_role in user_roles]
+            if resolved_role_ids is None:
+                user_roles = await UserRoleDao.aget_user_roles(user_id)
+                resolved_role_ids = [user_role.role_id for user_role in user_roles]
+            is_global_super = await _check_is_global_super(user_id, role_ids=resolved_role_ids)
         login_user = cls(
             user_id=user_id,
             user_name=user_name,
-            user_role=role_ids,
+            user_role=resolved_role_ids,
             tenant_id=tenant_id or DEFAULT_TENANT_ID,
             token_version=token_version,
             is_global_super=is_global_super,
@@ -669,6 +670,7 @@ class LoginUser(BaseModel):
         user: User,
         *,
         is_department_admin: bool = False,
+        role_ids: Iterable[int] | None = None,
     ) -> (list[int] | str, list[str]):
         """Resolve role key(s) and web_menu.
 
@@ -677,18 +679,22 @@ class LoginUser(BaseModel):
         - ``system_config`` is only granted via super-admin or department-admin; it is
           stripped for other users even if legacy role_access rows exist.
         """
-        db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
+        if role_ids is None:
+            db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
+            resolved_role_ids = [user_role.role_id for user_role in db_user_role]
+        else:
+            resolved_role_ids = list(role_ids)
         role = ""
-        role_ids = []
-        for user_role in db_user_role:
-            if user_role.role_id == AdminRole:
+        non_admin_role_ids = []
+        for role_id in resolved_role_ids:
+            if role_id == AdminRole:
                 role = "admin"
             else:
-                role_ids.append(user_role.role_id)
+                non_admin_role_ids.append(role_id)
         if role != "admin":
-            role = role_ids
+            role = non_admin_role_ids
             # AC-13: union of all roles' menu permissions
-            web_menu = await RoleAccessDao.aget_role_access(role_ids, AccessType.WEB_MENU)
+            web_menu = await RoleAccessDao.aget_role_access(non_admin_role_ids, AccessType.WEB_MENU)
             web_menu = list({one.third_id for one in web_menu})
             personal_menu = await UserMenuAccessService.list_effective_menu_grants(
                 get_current_tenant_id() or DEFAULT_TENANT_ID,
@@ -706,22 +712,47 @@ class LoginUser(BaseModel):
         return role, web_menu
 
     @classmethod
-    async def effective_workbench_admin_flags(cls, user: User) -> tuple[bool, bool]:
+    async def effective_workbench_admin_flags(
+        cls,
+        user: User,
+        *,
+        role_ids: Iterable[int] | None = None,
+        is_department_admin: bool | None = None,
+    ) -> tuple[bool, bool]:
         """Effective (has_workbench, has_admin_console) after orphan strip and dept-admin merge."""
-        db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
-        if any(ur.role_id == AdminRole for ur in db_user_role):
+        if role_ids is None:
+            db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
+            resolved_role_ids = [user_role.role_id for user_role in db_user_role]
+        else:
+            resolved_role_ids = list(role_ids)
+        if AdminRole in resolved_role_ids:
             return True, True
-        is_department_admin = bool(await DepartmentDao.aget_user_admin_departments(user.user_id))
-        _, web_menu = await cls.get_roles_web_menu(user, is_department_admin=is_department_admin)
+        if is_department_admin is None:
+            is_department_admin = bool(await DepartmentDao.aget_user_admin_departments(user.user_id))
+        _, web_menu = await cls.get_roles_web_menu(
+            user,
+            is_department_admin=is_department_admin,
+            role_ids=resolved_role_ids,
+        )
         wm = set(web_menu)
         has_wb = bool(wm & _WEB_MENU_WORKBENCH_ALL)
         has_adm = bool(wm & _WEB_MENU_ADMIN_ALL)
         return has_wb, has_adm
 
     @classmethod
-    async def user_has_workbench_or_admin_effective_menu(cls, user: User) -> bool:
+    async def user_has_workbench_or_admin_effective_menu(
+        cls,
+        user: User,
+        *,
+        role_ids: Iterable[int] | None = None,
+        is_department_admin: bool | None = None,
+    ) -> bool:
         """True if effective web_menu grants workbench or admin-console area (incl. parent-only keys)."""
-        has_wb, has_adm = await cls.effective_workbench_admin_flags(user)
+        has_wb, has_adm = await cls.effective_workbench_admin_flags(
+            user,
+            role_ids=role_ids,
+            is_department_admin=is_department_admin,
+        )
         return has_wb or has_adm
 
     @classmethod
@@ -743,15 +774,31 @@ class LoginUser(BaseModel):
         return "platform"
 
     @classmethod
-    async def user_entry_payload_for_read(cls, user: User) -> dict:
-        hw, ha = await cls.effective_workbench_admin_flags(user)
+    async def user_entry_payload_for_read(
+        cls,
+        user: User,
+        *,
+        role_ids: Iterable[int] | None = None,
+        is_department_admin: bool | None = None,
+    ) -> dict:
+        if role_ids is None:
+            db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
+            resolved_role_ids = [user_role.role_id for user_role in db_user_role]
+        else:
+            resolved_role_ids = list(role_ids)
+        is_super = AdminRole in resolved_role_ids
+        if not is_super and is_department_admin is None:
+            is_department_admin = bool(await DepartmentDao.aget_user_admin_departments(user.user_id))
+        hw, ha = await cls.effective_workbench_admin_flags(
+            user,
+            role_ids=resolved_role_ids,
+            is_department_admin=is_department_admin,
+        )
         # Regular users (neither super-admin nor department admin) prefer 工作台
         # when they have workbench access — even if their role also carries
         # admin-console menus (e.g. legacy default roles) — so login doesn't drop
         # them into 管理后台. `and` short-circuits the dept-admin FGA call for supers.
-        db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
-        is_super = any(ur.role_id == AdminRole for ur in db_user_role)
-        prefer_workbench = (not is_super) and not bool(await DepartmentDao.aget_user_admin_departments(user.user_id))
+        prefer_workbench = (not is_super) and not bool(is_department_admin)
         return {
             "has_workbench": hw,
             "has_admin_console": ha,
@@ -759,7 +806,12 @@ class LoginUser(BaseModel):
         }
 
     @classmethod
-    async def compute_menu_approval_modes(cls, user: User) -> tuple[bool, bool]:
+    async def compute_menu_approval_modes(
+        cls,
+        user: User,
+        *,
+        role_ids: Iterable[int] | None = None,
+    ) -> tuple[bool, bool]:
         """Per-area "show unauthorized menus (apply)" flags: ``(workbench, admin)``.
 
         True when any assigned (non-super-admin) role enables the corresponding
@@ -768,11 +820,15 @@ class LoginUser(BaseModel):
         ``menu_approval_mode``; reads fall back to the legacy key so existing
         roles keep their current behavior until re-saved.
         """
-        db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
-        role_ids = [ur.role_id for ur in db_user_role if ur.role_id != AdminRole]
-        if not role_ids:
+        if role_ids is None:
+            db_user_role = await UserRoleDao.aget_user_roles(user.user_id)
+            resolved_role_ids = [user_role.role_id for user_role in db_user_role]
+        else:
+            resolved_role_ids = list(role_ids)
+        non_admin_role_ids = [role_id for role_id in resolved_role_ids if role_id != AdminRole]
+        if not non_admin_role_ids:
             return False, False
-        roles = await RoleDao.aget_role_by_ids(role_ids)
+        roles = await RoleDao.aget_role_by_ids(non_admin_role_ids)
 
         def _truthy(v: object) -> bool:
             return v is True or v == 1 or str(v).lower() in ("true", "1")

src/backend/bisheng/user/domain/services/user.py

@@ -1,30 +1,30 @@
 import re
 from base64 import b64decode
 from datetime import datetime
-from typing import List, Optional, TYPE_CHECKING
+from typing import TYPE_CHECKING
 from urllib.parse import unquote, urlsplit
 
 import rsa
-from fastapi import Request, Depends, UploadFile, HTTPException
+from fastapi import Depends, HTTPException, Request, UploadFile
 from loguru import logger
 
 from bisheng.common.constants.enums.telemetry import BaseTelemetryTypeEnum
 from bisheng.common.errcode.user import (
     CaptchaError,
     UserForbiddenError,
-    UserValidateError,
-    UserPasswordMaxTryError,
-    UserPasswordExpireError,
     UserNameTooLongError,
     UserNoRoleForLoginError,
     UserNoWebMenuForLoginError,
+    UserPasswordExpireError,
+    UserPasswordMaxTryError,
     UserPasswordStrengthError,
+    UserValidateError,
 )
 from bisheng.common.schemas.api import UnifiedResponseModel, resp_200
 from bisheng.common.schemas.telemetry.event_data_schema import UserLoginEventData
 from bisheng.common.services import telemetry_service
 from bisheng.common.services.config_service import settings
-from bisheng.core.cache.redis_manager import get_redis_client_sync, get_redis_client
+from bisheng.core.cache.redis_manager import get_redis_client, get_redis_client_sync
 from bisheng.core.context.tenant import (
     DEFAULT_TENANT_ID,
     bypass_tenant_filter,
@@ -38,30 +38,31 @@ from bisheng.database.constants import AdminRole, DefaultRole
 from bisheng.database.models.department import DepartmentDao
 from bisheng.database.models.user_group import UserGroupDao
 from bisheng.permission.domain.services.legacy_rbac_sync_service import LegacyRBACSyncService
-from bisheng.user.domain.models.user import User, UserDao, UserLogin, UserRead, UserCreate
+from bisheng.user.domain.models.user import User, UserCreate, UserDao, UserLogin, UserRead
 from bisheng.user.domain.models.user_role import UserRoleDao
-from bisheng.utils import md5_hash, get_request_ip, generate_uuid
+from bisheng.utils import generate_uuid, get_request_ip, md5_hash
 from bisheng.utils.constants import RSA_KEY
-from .auth import LoginUser, AuthJwt
+
+from ..const import USER_CURRENT_SESSION, USER_PASSWORD_ERROR
+from .auth import AuthJwt, LoginUser
 from .captcha import verify_captcha
-from ..const import USER_PASSWORD_ERROR, USER_CURRENT_SESSION
 
 if TYPE_CHECKING:
     from bisheng.api.v1.schemas import CreateUserReq
 
 # Allowed avatar file types and their MIME types
 ALLOWED_AVATAR_TYPES = {
-    'image/jpeg': '.jpg',
-    'image/png': '.png',
-    'image/webp': '.webp',
-    'image/gif': '.gif',
+    "image/jpeg": ".jpg",
+    "image/png": ".png",
+    "image/webp": ".webp",
+    "image/gif": ".gif",
 }
 MAX_AVATAR_SIZE = 10 * 1024 * 1024  # 10MB
-AVATAR_OBJECT_PREFIX = 'avatar/'
+AVATAR_OBJECT_PREFIX = "avatar/"
 
 # Password strength rule, kept in sync with the frontend PWD_RULE:
 # at least 8 chars, with a lowercase, an uppercase, a digit and a symbol.
-PASSWORD_STRENGTH_PATTERN = re.compile(r'^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[\W_]).{8,128}$')
+PASSWORD_STRENGTH_PATTERN = re.compile(r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[\W_]).{8,128}$")
 
 
 class UserService:
@@ -70,19 +71,22 @@ class UserService:
         """禁用账号后立刻让已签发的 JWT 失效（F012 ``token_version``），并清理管理端 scope 缓存。"""
         try:
             await UserDao.aincrement_token_version(user_id)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             logger.warning(
-                'aincrement_token_version failed after account disabled user_id=%s: %s',
-                user_id, exc,
+                "aincrement_token_version failed after account disabled user_id=%s: %s",
+                user_id,
+                exc,
             )
             return
         try:
             from bisheng.admin.domain.services.tenant_scope import TenantScopeService
+
             await TenantScopeService.clear_on_token_version_bump(user_id)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             logger.debug(
-                'clear_on_token_version_bump failed after account disabled user_id=%s: %s',
-                user_id, exc,
+                "clear_on_token_version_bump failed after account disabled user_id=%s: %s",
+                user_id,
+                exc,
             )
 
     @classmethod
@@ -91,11 +95,11 @@ class UserService:
             return avatar
 
         avatar = avatar.strip()
-        path = urlsplit(avatar).path if '://' in avatar or avatar.startswith('/') else avatar.split('?', 1)[0]
-        path = unquote(path).lstrip('/')
+        path = urlsplit(avatar).path if "://" in avatar or avatar.startswith("/") else avatar.split("?", 1)[0]
+        path = unquote(path).lstrip("/")
 
-        if bucket and path.startswith(f'{bucket}/'):
-            path = path[len(bucket) + 1:]
+        if bucket and path.startswith(f"{bucket}/"):
+            path = path[len(bucket) + 1 :]
 
         if path.startswith(AVATAR_OBJECT_PREFIX):
             return path
@@ -127,14 +131,14 @@ class UserService:
     async def build_user_read(cls, user: User, **kwargs) -> UserRead:
         user_data = user.model_dump()
         user_data.update(kwargs)
-        user_data['avatar'] = await cls.get_avatar_share_link(user_data.get('avatar'))
+        user_data["avatar"] = await cls.get_avatar_share_link(user_data.get("avatar"))
         return UserRead(**user_data)
 
     @classmethod
     def build_user_read_sync(cls, user: User, **kwargs) -> UserRead:
         user_data = user.model_dump()
         user_data.update(kwargs)
-        user_data['avatar'] = cls.get_avatar_share_link_sync(user_data.get('avatar'))
+        user_data["avatar"] = cls.get_avatar_share_link_sync(user_data.get("avatar"))
         return UserRead(**user_data)
 
     @classmethod
@@ -142,7 +146,7 @@ class UserService:
         """RSA 解密得到明文密码（未做 MD5）；无 RSA 配置时视为明文开发模式。"""
         if value := get_redis_client_sync().get(RSA_KEY):
             private_key = value[1]
-            return rsa.decrypt(b64decode(password), private_key).decode('utf-8')
+            return rsa.decrypt(b64decode(password), private_key).decode("utf-8")
         return password
 
     @classmethod
@@ -154,7 +158,7 @@ class UserService:
     def password_meets_policy(cls, plain: str) -> bool:
         """Same rule as the frontend: >=8 chars with lowercase, uppercase,
         digit and symbol."""
-        return bool(PASSWORD_STRENGTH_PATTERN.match(plain or ''))
+        return bool(PASSWORD_STRENGTH_PATTERN.match(plain or ""))
 
     @classmethod
     def validate_password_strength(cls, plain: str) -> None:
@@ -174,14 +178,14 @@ class UserService:
         return md5_hash(plain)
 
     @classmethod
-    def create_user(cls, request: Request, login_user: LoginUser, req_data: 'CreateUserReq'):
+    def create_user(cls, request: Request, login_user: LoginUser, req_data: "CreateUserReq"):
         """
         Create User
         """
         user = User(
             user_name=req_data.user_name,
             password=cls.decrypt_md5_password_strict(req_data.password),
-            source='local',
+            source="local",
             # Default external_id to user_name so password login (which queries
             # external_id only since 94323e3ec) works out of the box. SSO-synced
             # users set their own external_id via org_sync, not through here.
@@ -243,17 +247,17 @@ class UserService:
     @classmethod
     async def user_register(cls, user: UserCreate):
         # Captcha Verification
-        if settings.get_from_db('use_captcha'):
+        if settings.get_from_db("use_captcha"):
             if not user.captcha_key or not await verify_captcha(user.captcha, user.captcha_key):
                 raise CaptchaError()
 
         db_user = User.model_validate(user)
         person_id = (db_user.external_id or "").strip()
         if not person_id:
-            raise UserValidateError(msg='Person ID is required')
+            raise UserValidateError(msg="Person ID is required")
         existing_pid = await UserDao.aget_by_external_id(person_id)
         if existing_pid:
-            raise UserValidateError(msg='Person ID already exists')
+            raise UserValidateError(msg="Person ID already exists")
         db_user.external_id = person_id
 
         # 允许用户名重复；人员唯一性由 external_id / user_id 等保证
@@ -295,16 +299,17 @@ class UserService:
     @classmethod
     async def _ensure_user_guest_department_membership(cls, user_id: int) -> None:
         """注册完成后自动加入“临时访客”部门（主部门）。"""
-        from bisheng.database.models.department import Department, UserDepartment
         from sqlmodel import select
 
-        guest_dept_id = 'BS@guest'
+        from bisheng.database.models.department import Department, UserDepartment
+
+        guest_dept_id = "BS@guest"
         async with get_async_db_session() as session:
             dept = (
                 await session.exec(
                     select(Department).where(
                         Department.dept_id == guest_dept_id,
-                        Department.status == 'active',
+                        Department.status == "active",
                     )
                 )
             ).first()
@@ -320,16 +325,19 @@ class UserService:
             ).first()
             if exists:
                 return
-            session.add(UserDepartment(
-                user_id=user_id,
-                department_id=dept.id,
-                is_primary=1,
-                source='local',
-            ))
+            session.add(
+                UserDepartment(
+                    user_id=user_id,
+                    department_id=dept.id,
+                    is_primary=1,
+                    source="local",
+                )
+            )
             await session.commit()
             from bisheng.department.domain.services.department_change_handler import (
                 DepartmentChangeHandler,
             )
+
             ops = DepartmentChangeHandler.on_members_added(dept.id, [user_id])
             await DepartmentChangeHandler.execute_async(ops)
 
@@ -338,14 +346,16 @@ class UserService:
         """Resolve the tenant_id used for anonymous self-registration."""
         from bisheng.database.models.tenant import TenantDao
 
-        code = (settings.multi_tenant.default_tenant_code or 'default').strip() or 'default'
+        code = (settings.multi_tenant.default_tenant_code or "default").strip() or "default"
         with bypass_tenant_filter():
             tenant = await TenantDao.aget_by_code(code)
         return int(tenant.id) if tenant else DEFAULT_TENANT_ID
 
     @classmethod
     async def _ensure_user_default_tenant_association(
-        cls, user_id: int, tenant_id: Optional[int] = None,
+        cls,
+        user_id: int,
+        tenant_id: int | None = None,
     ) -> int:
         """Ensure a user has a default tenant row and return its tenant_id."""
         from bisheng.database.models.tenant import UserTenantDao
@@ -353,7 +363,7 @@ class UserService:
         existing = await UserTenantDao.aget_user_tenants(user_id)
         if existing:
             for row in existing:
-                if getattr(row, 'is_default', 0) == 1:
+                if getattr(row, "is_default", 0) == 1:
                     return int(row.tenant_id)
             return int(existing[0].tenant_id)
         if tenant_id is None:
@@ -363,8 +373,12 @@ class UserService:
 
     @classmethod
     async def _reject_login_if_user_has_no_usable_access(
-        cls, db_user: User,
-    ) -> Optional[UnifiedResponseModel]:
+        cls,
+        db_user: User,
+        *,
+        role_ids: list[int] | None = None,
+        is_department_admin: bool | None = None,
+    ) -> UnifiedResponseModel | None:
         """无角色且非部门/用户组管理员时拒绝登录；有角色但生效菜单既不包含工作台也不包含管理后台时拒绝登录。
 
         需审批模式下角色可仅勾选一级菜单（workstation/admin）而无二级项，仍视为有菜单权限，允许登录。
@@ -374,22 +388,33 @@ class UserService:
         # NoTenantContextError because do_orm_execute can't infer a tenant for
         # tenant-aware tables (userrole/department/usergroup/roleaccess).
         from bisheng.core.context.tenant import bypass_tenant_filter
+
         with bypass_tenant_filter():
-            roles = await UserRoleDao.aget_user_roles(db_user.user_id)
-            if not roles:
-                if await DepartmentDao.aget_user_admin_departments(db_user.user_id):
+            if role_ids is None:
+                roles = await UserRoleDao.aget_user_roles(db_user.user_id)
+                resolved_role_ids = [role.role_id for role in roles]
+            else:
+                resolved_role_ids = list(role_ids)
+            if AdminRole in resolved_role_ids:
+                return None
+            if is_department_admin is None:
+                is_department_admin = bool(await DepartmentDao.aget_user_admin_departments(db_user.user_id))
+            if not resolved_role_ids:
+                if is_department_admin:
                     return None
                 group_admins = await UserGroupDao.aget_user_admin_group(db_user.user_id)
                 if group_admins:
                     return None
                 return UserNoRoleForLoginError.return_resp()
 
-            if any(ur.role_id == AdminRole for ur in roles):
-                return None
-            if await DepartmentDao.aget_user_admin_departments(db_user.user_id):
+            if is_department_admin:
                 return None
 
-            if not await LoginUser.user_has_workbench_or_admin_effective_menu(db_user):
+            if not await LoginUser.user_has_workbench_or_admin_effective_menu(
+                db_user,
+                role_ids=resolved_role_ids,
+                is_department_admin=is_department_admin,
+            ):
                 return UserNoWebMenuForLoginError.return_resp()
             return None
 
@@ -397,7 +422,7 @@ class UserService:
     async def user_login(cls, request: Request, user: UserLogin, auth_jwt: AuthJwt = Depends()):
         from bisheng.api.services.audit_log import AuditLogService
 
-        if await settings.aget_from_db('use_captcha'):
+        if await settings.aget_from_db("use_captcha"):
             if not user.captcha_key or not await verify_captcha(user.captcha, user.captcha_key):
                 raise CaptchaError()
 
@@ -430,7 +455,17 @@ class UserService:
 
         await cls.clear_error_password_key(db_user.user_id)
 
-        no_role_resp = await cls._reject_login_if_user_has_no_usable_access(db_user)
+        with bypass_tenant_filter():
+            login_user_roles = await UserRoleDao.aget_user_roles(db_user.user_id)
+            login_role_ids = [role.role_id for role in login_user_roles]
+            login_is_department_admin = False
+            if AdminRole not in login_role_ids:
+                login_is_department_admin = bool(await DepartmentDao.aget_user_admin_departments(db_user.user_id))
+            no_role_resp = await cls._reject_login_if_user_has_no_usable_access(
+                db_user,
+                role_ids=login_role_ids,
+                is_department_admin=login_is_department_admin,
+            )
         if no_role_resp is not None:
             return no_role_resp
 
@@ -440,9 +475,9 @@ class UserService:
         tenants_list = None
 
         if settings.multi_tenant.enabled:
-            from bisheng.database.models.tenant import UserTenantDao, TenantDao
             from bisheng.common.errcode.tenant import (
-                NoTenantsAvailableError, TenantDisabledError,
+                NoTenantsAvailableError,
+                TenantDisabledError,
             )
             from bisheng.core.context.tenant import DEFAULT_TENANT_ID, bypass_tenant_filter
 
@@ -454,8 +489,10 @@ class UserService:
             # of a disabled tenant would get fallback-routed to Root and log
             # in normally, which contradicts the operator's intent.
             from bisheng.database.models.department import (
-                DepartmentDao, UserDepartmentDao,
+                UserDepartmentDao,
             )
+            from bisheng.database.models.tenant import TenantDao, UserTenantDao
+
             with bypass_tenant_filter():
                 primary_dept = await UserDepartmentDao.aget_user_primary_department(
                     db_user.user_id,
@@ -468,31 +505,31 @@ class UserService:
                         mount_tenant = await TenantDao.aget_by_id(
                             mount_dept.mounted_tenant_id,
                         )
-                        if mount_tenant is not None and mount_tenant.status == 'disabled':
+                        if mount_tenant is not None and mount_tenant.status == "disabled":
                             raise TenantDisabledError()
 
             user_tenants = await UserTenantDao.aget_user_tenants_with_details(db_user.user_id)
-            active_tenants = [t for t in user_tenants if t.get('status') == 'active']
+            active_tenants = [t for t in user_tenants if t.get("status") == "active"]
 
             if len(active_tenants) == 0:
                 # 注册/历史数据可能未写入 user_tenant；若默认租户存在则自动挂接，避免无法登录
-                code = (settings.multi_tenant.default_tenant_code or 'default').strip() or 'default'
+                code = (settings.multi_tenant.default_tenant_code or "default").strip() or "default"
                 with bypass_tenant_filter():
                     default_tenant = await TenantDao.aget_by_code(code)
                     if default_tenant is None:
                         default_tenant = await TenantDao.aget_by_id(DEFAULT_TENANT_ID)
-                if default_tenant and default_tenant.status == 'active':
+                if default_tenant and default_tenant.status == "active":
                     await UserTenantDao.aadd_user_to_tenant(
                         user_id=db_user.user_id,
                         tenant_id=default_tenant.id,
                         is_default=1,
                     )
                     user_tenants = await UserTenantDao.aget_user_tenants_with_details(db_user.user_id)
-                    active_tenants = [t for t in user_tenants if t.get('status') == 'active']
+                    active_tenants = [t for t in user_tenants if t.get("status") == "active"]
                 if len(active_tenants) == 0:
                     raise NoTenantsAvailableError()
             elif len(active_tenants) == 1:
-                tenant_id = active_tenants[0]['tenant_id']
+                tenant_id = active_tenants[0]["tenant_id"]
                 await UserTenantDao.aupdate_last_access_time(db_user.user_id, tenant_id)
             else:
                 # Multiple tenants: issue temporary JWT with tenant_id=0
@@ -511,18 +548,20 @@ class UserService:
             from bisheng.tenant.domain.services.user_tenant_sync_service import (
                 UserTenantSyncService,
             )
+
             leaf = await UserTenantSyncService.sync_user(
-                db_user.user_id, trigger=UserTenantSyncTrigger.LOGIN,
+                db_user.user_id,
+                trigger=UserTenantSyncTrigger.LOGIN,
             )
             # Override tenant_id for JWT payload — the resolver is the
             # authoritative source for leaf tenancy in v2.5.1.
-            if leaf is not None and getattr(leaf, 'id', None):
+            if leaf is not None and getattr(leaf, "id", None):
                 tenant_id = leaf.id
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             logger.warning(
-                'F012 login-time tenant sync failed for user %d: %s — '
-                'falling back to legacy tenant resolution',
-                db_user.user_id, exc,
+                "F012 login-time tenant sync failed for user %d: %s — falling back to legacy tenant resolution",
+                db_user.user_id,
+                exc,
             )
 
         # Block login when the resolved leaf tenant is disabled/archived.
@@ -531,24 +570,27 @@ class UserService:
         # immediately kicked out" instead of a clear error on the login form.
         # DB.status is authoritative; Redis blacklist is a defensive cross-check.
         if settings.multi_tenant.enabled and tenant_id and tenant_id > 0:
-            from bisheng.database.models.tenant import TenantDao
             from bisheng.common.errcode.tenant import TenantDisabledError
             from bisheng.core.context.tenant import bypass_tenant_filter
+            from bisheng.database.models.tenant import TenantDao
+
             with bypass_tenant_filter():
                 tenant_obj = await TenantDao.aget_by_id(tenant_id)
-            if not tenant_obj or tenant_obj.status != 'active':
+            if not tenant_obj or tenant_obj.status != "active":
                 raise TenantDisabledError()
             try:
                 from bisheng.tenant.domain.services.tenant_service import DISABLED_TENANT_KEY
+
                 redis_client = await get_redis_client()
                 if await redis_client.aget(DISABLED_TENANT_KEY.format(tenant_id)):
                     raise TenantDisabledError()
             except TenantDisabledError:
                 raise
-            except Exception as exc:  # noqa: BLE001
+            except Exception as exc:
                 logger.debug(
-                    'tenant blacklist check skipped for tenant %s: %s',
-                    tenant_id, exc,
+                    "tenant blacklist check skipped for tenant %s: %s",
+                    tenant_id,
+                    exc,
                 )
 
         # Fetch fresh token_version (sync_user may have just bumped it) and
@@ -556,12 +598,14 @@ class UserService:
         fresh_token_version = 0
         try:
             fresh_token_version = await UserDao.aget_token_version(db_user.user_id)
-        except Exception as exc:  # noqa: BLE001
-            logger.debug('aget_token_version failed for %d: %s', db_user.user_id, exc)
+        except Exception as exc:
+            logger.debug("aget_token_version failed for %d: %s", db_user.user_id, exc)
 
         # gen jwt token
         access_token = LoginUser.create_access_token(
-            user=db_user, auth_jwt=auth_jwt, tenant_id=tenant_id,
+            user=db_user,
+            auth_jwt=auth_jwt,
+            tenant_id=tenant_id,
             token_version=fresh_token_version,
         )
 
@@ -570,58 +614,81 @@ class UserService:
 
         # Set the logged in user's currentcookie, .jwtValid for an additional hour
         redis_client = await get_redis_client()
-        await redis_client.aset(USER_CURRENT_SESSION.format(db_user.user_id), access_token,
-                                auth_jwt.cookie_conf.jwt_token_expire_time + 3600)
+        await redis_client.aset(
+            USER_CURRENT_SESSION.format(db_user.user_id),
+            access_token,
+            auth_jwt.cookie_conf.jwt_token_expire_time + 3600,
+        )
 
         # Log Audit Logs
-        login_user = await LoginUser.init_login_user(db_user.user_id, db_user.user_name)
+        login_user = await LoginUser.init_login_user(
+            db_user.user_id,
+            db_user.user_name,
+            role_ids=login_role_ids,
+        )
         AuditLogService.user_login(login_user, get_request_ip(request))
 
         # RecordTelemetryJournal
-        await telemetry_service.log_event(user_id=db_user.user_id, event_type=BaseTelemetryTypeEnum.USER_LOGIN,
-                                          trace_id=trace_id_var.get(),
-                                          event_data=UserLoginEventData(method="password"))
+        await telemetry_service.log_event(
+            user_id=db_user.user_id,
+            event_type=BaseTelemetryTypeEnum.USER_LOGIN,
+            trace_id=trace_id_var.get(),
+            event_data=UserLoginEventData(method="password"),
+        )
 
         # Build response with tenant info. is_global_super is already a bool
         # on LoginUser, populated by init_login_user; surface it so the
         # frontend can render admin menus without a /user/info round-trip.
-        entry = await LoginUser.user_entry_payload_for_read(db_user)
+        entry = await LoginUser.user_entry_payload_for_read(
+            db_user,
+            role_ids=login_role_ids,
+            is_department_admin=login_is_department_admin,
+        )
         extra_fields = {
-            'access_token': access_token,
-            'is_global_super': login_user.is_global_super,
+            "access_token": access_token,
+            "is_global_super": login_user.is_global_super,
             **entry,
         }
         if requires_tenant_selection:
-            extra_fields['requires_tenant_selection'] = True
-            extra_fields['tenants'] = tenants_list
+            extra_fields["requires_tenant_selection"] = True
+            extra_fields["tenants"] = tenants_list
         if tenant_id and tenant_id > 0:
-            extra_fields['tenant_id'] = tenant_id
-            tenant_info = next((t for t in (tenants_list or []) if t.get('tenant_id') == tenant_id), None)
+            extra_fields["tenant_id"] = tenant_id
+            tenant_info = next((t for t in (tenants_list or []) if t.get("tenant_id") == tenant_id), None)
             if not tenant_info and settings.multi_tenant.enabled:
-                from bisheng.database.models.tenant import TenantDao
                 from bisheng.core.context.tenant import bypass_tenant_filter
+                from bisheng.database.models.tenant import TenantDao
+
                 with bypass_tenant_filter():
                     t_obj = await TenantDao.aget_by_id(tenant_id)
                 if t_obj:
-                    extra_fields['tenant_name'] = t_obj.tenant_name
+                    extra_fields["tenant_name"] = t_obj.tenant_name
 
         return resp_200(await cls.build_user_read(db_user, **extra_fields))
 
     @classmethod
-    def get_user_all_info(cls, *, start_time: datetime = None, end_time: datetime = None, user_ids: List[int] = None,
-                          page: int = 1, page_size: int = 100) -> List[User]:
-        """ Get user information, including user group and role information """
-        return UserDao.get_user_with_group_role(page=page, page_size=page_size, user_ids=user_ids,
-                                                start_time=start_time, end_time=end_time)
+    def get_user_all_info(
+        cls,
+        *,
+        start_time: datetime = None,
+        end_time: datetime = None,
+        user_ids: list[int] = None,
+        page: int = 1,
+        page_size: int = 100,
+    ) -> list[User]:
+        """Get user information, including user group and role information"""
+        return UserDao.get_user_with_group_role(
+            page=page, page_size=page_size, user_ids=user_ids, start_time=start_time, end_time=end_time
+        )
 
     @classmethod
     def get_first_user(cls) -> User | None:
-        """ Get the first user """
+        """Get the first user"""
         return UserDao.get_first_user()
 
     @classmethod
     async def get_user_by_id(cls, user_id: int) -> User | None:
-        """ Get user by username """
+        """Get user by username"""
         return await UserDao.aget_user(user_id)
 
     @classmethod
@@ -634,22 +701,16 @@ class UserService:
         """
         # Validate file type
         if file.content_type not in ALLOWED_AVATAR_TYPES:
-            raise HTTPException(
-                status_code=400,
-                detail=f'Invalid file type. Allowed types: jpg, png, webp, gif'
-            )
+            raise HTTPException(status_code=400, detail="Invalid file type. Allowed types: jpg, png, webp, gif")
 
         # Read file content to check size
         content = await file.read()
         if len(content) > MAX_AVATAR_SIZE:
-            raise HTTPException(
-                status_code=400,
-                detail='File size exceeds limit. Maximum size: 10MB'
-            )
+            raise HTTPException(status_code=400, detail="File size exceeds limit. Maximum size: 10MB")
 
         # Generate object name for MinIO
         file_ext = ALLOWED_AVATAR_TYPES[file.content_type]
-        object_name = f'avatar/{user_id}/{generate_uuid()}{file_ext}'
+        object_name = f"avatar/{user_id}/{generate_uuid()}{file_ext}"
 
         # Upload to MinIO
         minio_client = await get_minio_storage()

src/backend/bisheng/utils/http_middleware.py

@@ -1,5 +1,6 @@
 # Define a custom middleware class
 import http.cookies
+from collections.abc import Iterable
 from time import time
 
 from fastapi import Request
@@ -141,7 +142,11 @@ async def _validate_token_version(
     return int(current) == int(payload_token_version)
 
 
-async def _check_is_global_super(user_id: int) -> bool:
+async def _check_is_global_super(
+    user_id: int,
+    *,
+    role_ids: Iterable[int] | None = None,
+) -> bool:
     """FGA check: ``user:{id} super_admin system:global`` with Redis caching.
 
     Used by the middleware to decide whether to inject an IN-list filter
@@ -179,10 +184,15 @@ async def _check_is_global_super(user_id: int) -> bool:
     if not is_super:
         try:
             from bisheng.database.constants import AdminRole
-            from bisheng.user.domain.models.user_role import UserRoleDao
 
-            roles = await UserRoleDao.aget_user_roles(user_id)
-            is_super = any(int(role.role_id) == AdminRole for role in roles)
+            if role_ids is None:
+                from bisheng.user.domain.models.user_role import UserRoleDao
+
+                roles = await UserRoleDao.aget_user_roles(user_id)
+                resolved_role_ids = [role.role_id for role in roles]
+            else:
+                resolved_role_ids = role_ids
+            is_super = any(int(role_id) == AdminRole for role_id in resolved_role_ids)
         except Exception as exc:
             logger.debug("legacy AdminRole fallback failed for user %d: %s", user_id, exc)
 

src/backend/test/user/test_user_access_request_dedup.py

@@ -0,0 +1,107 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import bisheng.user.domain.services.auth as auth_module
+import bisheng.user.domain.services.user as user_service_module
+
+
+async def test_user_entry_queries_each_access_source_once():
+    user = SimpleNamespace(user_id=7, tenant_id=1)
+    role_rows = [SimpleNamespace(role_id=2)]
+
+    with (
+        patch.object(
+            auth_module.UserRoleDao,
+            "aget_user_roles",
+            new_callable=AsyncMock,
+            return_value=role_rows,
+        ) as mock_roles,
+        patch.object(
+            auth_module.DepartmentDao,
+            "aget_user_admin_departments",
+            new_callable=AsyncMock,
+            return_value=[],
+        ) as mock_departments,
+        patch.object(
+            auth_module.RoleAccessDao,
+            "aget_role_access",
+            new_callable=AsyncMock,
+            return_value=[SimpleNamespace(third_id="workstation")],
+        ),
+        patch.object(
+            auth_module.UserMenuAccessService,
+            "list_effective_menu_grants",
+            new_callable=AsyncMock,
+            return_value=[],
+        ),
+    ):
+        entry = await auth_module.LoginUser.user_entry_payload_for_read(user)
+
+    assert entry == {
+        "has_workbench": True,
+        "has_admin_console": False,
+        "default_entry": "workspace",
+    }
+    mock_roles.assert_awaited_once_with(7)
+    mock_departments.assert_awaited_once_with(7)
+
+
+async def test_init_login_user_reuses_explicit_empty_role_snapshot():
+    with (
+        patch.object(
+            auth_module.UserRoleDao,
+            "aget_user_roles",
+            new_callable=AsyncMock,
+        ) as mock_async_roles,
+        patch.object(auth_module.UserRoleDao, "get_user_roles") as mock_sync_roles,
+        patch(
+            "bisheng.utils.http_middleware._check_is_global_super",
+            new_callable=AsyncMock,
+            return_value=False,
+        ) as mock_global_super,
+    ):
+        login_user = await auth_module.LoginUser.init_login_user(
+            9,
+            "user-9",
+            role_ids=[],
+        )
+
+    assert login_user.user_role == []
+    mock_async_roles.assert_not_awaited()
+    mock_sync_roles.assert_not_called()
+    mock_global_super.assert_awaited_once_with(9, role_ids=[])
+
+
+async def test_login_access_guard_reuses_resolved_roles_and_department_flag():
+    user = SimpleNamespace(user_id=11)
+
+    with (
+        patch.object(
+            user_service_module.UserRoleDao,
+            "aget_user_roles",
+            new_callable=AsyncMock,
+            return_value=[SimpleNamespace(role_id=2)],
+        ) as mock_roles,
+        patch.object(
+            user_service_module.DepartmentDao,
+            "aget_user_admin_departments",
+            new_callable=AsyncMock,
+            return_value=[],
+        ) as mock_departments,
+        patch.object(
+            user_service_module.LoginUser,
+            "user_has_workbench_or_admin_effective_menu",
+            new_callable=AsyncMock,
+            return_value=True,
+        ) as mock_effective_menu,
+    ):
+        result = await user_service_module.UserService._reject_login_if_user_has_no_usable_access(user)
+
+    assert result is None
+    mock_roles.assert_awaited_once_with(11)
+    mock_departments.assert_awaited_once_with(11)
+    mock_effective_menu.assert_awaited_once_with(
+        user,
+        role_ids=[2],
+        is_department_admin=False,
+    )