diff --git a/mcpgateway/cache/auth_cache.py b/mcpgateway/cache/auth_cache.py
index 450952f191c..4c0771028dc 100644
--- a/mcpgateway/cache/auth_cache.py
+++ b/mcpgateway/cache/auth_cache.py
@@ -623,6 +623,118 @@ async def set_user_role(self, email: str, team_id: str, role: Optional[str]) ->
                 expiry=time.time() + self._role_ttl,
             )
 
+    async def get_user_roles_list(self, email: str) -> Optional[List[str]]:
+        """Get the cached list of all role names for a user across all teams.
+
+        Returns:
+            - None: Cache miss (caller should query DB)
+            - list[str]: Cached role names (may be empty if user has no roles)
+
+        Args:
+            email: User email
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> result = asyncio.run(cache.get_user_roles_list("test@example.com"))
+            >>> result is None  # Cache miss
+            True
+        """
+        if not self._enabled:
+            return None
+
+        cache_key = f"roles_list:{email}"
+
+        # Check L1 in-memory cache
+        entry = self._role_cache.get(cache_key)
+        if entry and not entry.is_expired():
+            self._hit_count += 1
+            return entry.value if entry.value is not None else []
+
+        # Check L2 Redis cache
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                import json  # pylint: disable=import-outside-toplevel
+
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                data = await redis.get(redis_key)
+                if data is not None:
+                    self._hit_count += 1
+                    self._redis_hit_count += 1
+                    roles = json.loads(data.decode() if isinstance(data, bytes) else data)
+                    with self._lock:
+                        self._role_cache[cache_key] = CacheEntry(value=roles, expiry=time.time() + self._role_ttl)
+                    return roles
+                self._redis_miss_count += 1
+            except Exception as e:
+                logger.warning(f"AuthCache Redis get_user_roles_list failed: {e}")
+
+        self._miss_count += 1
+        return None
+
+    async def set_user_roles_list(self, email: str, roles: List[str]) -> None:
+        """Store the list of all role names for a user.
+
+        Uses a long safety-net TTL; the primary invalidation mechanism is
+        explicit invalidation via invalidate_user_roles_list() on role changes.
+
+        Args:
+            email: User email
+            roles: List of role name strings
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> asyncio.run(cache.set_user_roles_list("test@example.com", ["developer"]))
+        """
+        if not self._enabled:
+            return
+
+        import json  # pylint: disable=import-outside-toplevel
+
+        cache_key = f"roles_list:{email}"
+        safety_net_ttl = 3600  # 1 hour fallback; write-invalidation is the primary mechanism
+
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                await redis.setex(redis_key, safety_net_ttl, json.dumps(roles))
+            except Exception as e:
+                logger.warning(f"AuthCache Redis set_user_roles_list failed: {e}")
+
+        with self._lock:
+            self._role_cache[cache_key] = CacheEntry(value=roles, expiry=time.time() + safety_net_ttl)
+
+    async def invalidate_user_roles_list(self, email: str) -> None:
+        """Invalidate the cached roles list for a user.
+
+        Call this whenever a role is assigned to or revoked from the user.
+
+        Args:
+            email: User email
+
+        Examples:
+            >>> import asyncio
+            >>> cache = AuthCache()
+            >>> asyncio.run(cache.invalidate_user_roles_list("test@example.com"))
+        """
+        cache_key = f"roles_list:{email}"
+        logger.debug(f"AuthCache: Invalidating roles_list cache for {email}")
+
+        with self._lock:
+            self._role_cache.pop(cache_key, None)
+
+        redis = await self._get_redis_client()
+        if redis:
+            try:
+                redis_key = self._get_redis_key("roles_list", cache_key)
+                await redis.delete(redis_key)
+                await redis.publish("mcpgw:auth:invalidate", f"roles_list:{email}")
+            except Exception as e:
+                logger.warning(f"AuthCache Redis invalidate_user_roles_list failed: {e}")
+
     async def invalidate_user_role(self, email: str, team_id: str) -> None:
         """Invalidate cached role for a user in a team.
 
diff --git a/mcpgateway/main.py b/mcpgateway/main.py
index f791eeecb02..60d5823ce7f 100644
--- a/mcpgateway/main.py
+++ b/mcpgateway/main.py
@@ -95,7 +95,7 @@
 from mcpgateway.middleware.token_scoping import token_scoping_middleware
 from mcpgateway.middleware.validation_middleware import ValidationMiddleware
 from mcpgateway.observability import init_telemetry
-from mcpgateway.plugins.framework import HttpHookType, PluginError, PluginManager, PluginViolationError, PromptHookType, ResourceHookType
+from mcpgateway.plugins.framework import GlobalContext, HttpAuthCheckPermissionPayload, HttpHookType, PluginError, PluginManager, PluginViolationError, PromptHookType, ResourceHookType
 from mcpgateway.plugins.framework.constants import PLUGIN_VIOLATION_CODE_MAPPING, PluginViolationCode, VALID_HTTP_STATUS_CODES
 from mcpgateway.routers.server_well_known import router as server_well_known_router
 from mcpgateway.routers.well_known import router as well_known_router
@@ -1004,6 +1004,49 @@ async def _ensure_rpc_permission(user, db: Session, permission: str, method: str
     if permission == "admin.system_config" and _is_permission_admin_user(user):
         return
 
+    # Layer 2a: Plugin hook — HTTP_AUTH_CHECK_PERMISSION
+    # Mirrors the same hook invoked by @require_permission in rbac.py.
+    if plugin_manager and plugin_manager.has_hooks_for(HttpHookType.HTTP_AUTH_CHECK_PERMISSION):
+        user_email = get_user_email(user) if not isinstance(user, dict) else (user.get("email") or "")
+        is_admin = _is_permission_admin_user(user)
+        user_role_names: list[str] = []
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache as _rpc_auth_cache  # pylint: disable=import-outside-toplevel
+
+            _cached_roles = await _rpc_auth_cache.get_user_roles_list(user_email)
+            if _cached_roles is not None:
+                user_role_names = _cached_roles
+            else:
+                _role_assignments = await PermissionService(db).get_user_roles(user_email)
+                user_role_names = [ur.role.name for ur in _role_assignments]
+                await _rpc_auth_cache.set_user_roles_list(user_email, user_role_names)
+        except Exception:  # nosec B110 - role fetch failure must never block the request
+            pass
+        _rpc_global_ctx = GlobalContext(
+            request_id=getattr(request, "state", None) and getattr(request.state, "request_id", None) or "",
+            user={"email": user_email, "is_admin": is_admin, "roles": user_role_names},
+        )
+        _rpc_result, _ = await plugin_manager.invoke_hook(
+            HttpHookType.HTTP_AUTH_CHECK_PERMISSION,
+            payload=HttpAuthCheckPermissionPayload(
+                user_email=user_email,
+                permission=permission,
+                resource_type="rpc",
+                is_admin=is_admin,
+                roles=user_role_names,
+                auth_method="rpc",
+            ),
+            global_context=_rpc_global_ctx,
+        )
+        if _rpc_result and _rpc_result.modified_payload and hasattr(_rpc_result.modified_payload, "granted"):
+            decision_reason = getattr(_rpc_result.modified_payload, "reason", None)
+            if _rpc_result.modified_payload.granted is False:
+                logger.warning(
+                    "RPC permission denied by plugin: method=%s permission=%s user=%s reason=%s",
+                    method, permission, user_email, decision_reason,
+                )
+                raise JSONRPCError(-32003, _ACCESS_DENIED_MSG, {"method": method})
+
     # Layer 2: RBAC check
     # Session tokens have no explicit team_id, so check across all team-scoped roles.
     # Mirrors the @require_permission decorator's check_any_team fallback (rbac.py:562-576).
diff --git a/mcpgateway/middleware/rbac.py b/mcpgateway/middleware/rbac.py
index 9fe315ad2f3..83e84bf088c 100644
--- a/mcpgateway/middleware/rbac.py
+++ b/mcpgateway/middleware/rbac.py
@@ -653,6 +653,29 @@ async def wrapper(*args, **kwargs):
                 plugin_context_table = user_context.get("plugin_context_table")
                 plugin_global_context = user_context.get("plugin_global_context")
 
+                # Fetch the user's role names so plugins (e.g. Cedar) can act on them.
+                # Try the roles list cache first; fall back to DB on miss.
+                # The cache is invalidated on assign/revoke, so it is always fresh on normal paths.
+                # Failure is non-fatal — plugins still receive email and is_admin.
+                user_role_names: List[str] = []
+                try:
+                    from mcpgateway.cache.auth_cache import auth_cache as _rbac_auth_cache  # pylint: disable=import-outside-toplevel
+
+                    _cached_roles = await _rbac_auth_cache.get_user_roles_list(user_context["email"])
+                    if _cached_roles is not None:
+                        user_role_names = _cached_roles
+                    else:
+                        _db_for_roles = kwargs.get("db") or user_context.get("db")
+                        if _db_for_roles:
+                            _role_assignments = await PermissionService(_db_for_roles).get_user_roles(user_context["email"], team_id=team_id)
+                        else:
+                            with fresh_db_session() as _db_roles:
+                                _role_assignments = await PermissionService(_db_roles).get_user_roles(user_context["email"], team_id=team_id)
+                        user_role_names = [ur.role.name for ur in _role_assignments]
+                        await _rbac_auth_cache.set_user_roles_list(user_context["email"], user_role_names)
+                except Exception:  # nosec B110 - role fetch failure must never block the request
+                    pass
+
                 # Reuse existing global context from middleware if available for consistency
                 # Otherwise create a new one (fallback for cases where middleware didn't run)
                 if plugin_global_context:
@@ -662,7 +685,13 @@ async def wrapper(*args, **kwargs):
                     global_context = GlobalContext(
                         request_id=request_id,
                         server_id=None,
-                        tenant_id=None,
+                        tenant_id=team_id,
+                        user={
+                            "email": user_context["email"],
+                            "is_admin": user_context.get("is_admin", False),
+                            "roles": user_role_names,
+                            "team_id": team_id,
+                        },
                     )
 
                 # Invoke permission check hook, passing plugin contexts from HTTP_PRE_REQUEST hook
@@ -674,6 +703,7 @@ async def wrapper(*args, **kwargs):
                         resource_type=resource_type,
                         team_id=team_id,
                         is_admin=user_context.get("is_admin", False),
+                        roles=user_role_names,
                         auth_method=user_context.get("auth_method"),
                         client_host=user_context.get("ip_address"),
                         user_agent=user_context.get("user_agent"),
diff --git a/mcpgateway/plugins/framework/hooks/http.py b/mcpgateway/plugins/framework/hooks/http.py
index 7e016cb5f4a..20dea5b3cbb 100644
--- a/mcpgateway/plugins/framework/hooks/http.py
+++ b/mcpgateway/plugins/framework/hooks/http.py
@@ -155,6 +155,7 @@ class HttpAuthCheckPermissionPayload(PluginPayload):
         resource_type: Type of resource being accessed (e.g., "tool", "server", "prompt")
         team_id: Team context for the permission check (if applicable)
         is_admin: Whether the user has admin privileges
+        roles: List of role names assigned to the user (e.g., ["developer", "viewer"])
         auth_method: Authentication method used (e.g., "simple_token", "jwt", "oauth")
         client_host: Client IP address for IP-based permission checks
         user_agent: User agent string for device-based permission checks
@@ -165,6 +166,7 @@ class HttpAuthCheckPermissionPayload(PluginPayload):
     resource_type: str | None = None
     team_id: str | None = None
     is_admin: bool = False
+    roles: list[str] = []
     auth_method: str | None = None
     client_host: str | None = None
     user_agent: str | None = None
diff --git a/mcpgateway/services/role_service.py b/mcpgateway/services/role_service.py
index f5e24211ad9..e7166fc907e 100644
--- a/mcpgateway/services/role_service.py
+++ b/mcpgateway/services/role_service.py
@@ -628,6 +628,14 @@ async def assign_role_to_user(
         self.db.commit()
         self.db.refresh(user_role)
 
+        # Invalidate roles list cache so next request fetches fresh roles from DB
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel
+
+            await auth_cache.invalidate_user_roles_list(user_email)
+        except Exception:  # nosec B110 - cache invalidation failure must never block the request
+            pass
+
         logger.info(f"Assigned role {role.name} to {user_email} (scope: {scope}, scope_id: {scope_id})")
         return user_role
 
@@ -671,6 +679,14 @@ async def revoke_role_from_user(self, user_email: str, role_id: str, scope: str,
         user_role.is_active = False
         self.db.commit()
 
+        # Invalidate roles list cache so next request fetches fresh roles from DB
+        try:
+            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel
+
+            await auth_cache.invalidate_user_roles_list(user_email)
+        except Exception:  # nosec B110 - cache invalidation failure must never block the request
+            pass
+
         logger.info(f"Revoked role {role_id} from {user_email} (scope: {scope}, scope_id: {scope_id})")
         return True
 
diff --git a/plugins/config.yaml b/plugins/config.yaml
index 10e28f072d9..ef27f0e2555 100644
--- a/plugins/config.yaml
+++ b/plugins/config.yaml
@@ -16,6 +16,55 @@ plugin_settings:
   include_user_info: false
 
 plugins:
+  # Cedar Policy Plugin - evaluates HTTP permission checks against Cedar policies
+  - name: "CedarPolicyPlugin"
+    kind: "cedarpolicyplugin.plugin.CedarPolicyPlugin"
+    description: "Evaluates user role-based permissions using Cedar policy language"
+    version: "0.1.0"
+    author: "ContextForge"
+    hooks: ["http_auth_check_permission", "tool_pre_invoke"]
+    tags: ["security", "cedar", "policy", "rbac", "auth"]
+    mode: "permissive"  # set to "enforce" to block on Deny
+    priority: 5
+    conditions: []
+    config:
+      policy_lang: "cedar"
+      # HTTP-level RBAC permits (used by http_auth_check_permission):
+      # Permit roles to perform gateway-level permission checks.
+      # Anything not permitted is implicitly denied.
+      #
+      # Tool-level permits (used by tool_pre_invoke):
+      # principal = Role::"<role>", action = Action::"<tool_name>", resource = Server::"<server_name>"
+      policy:
+        # HTTP-level: data_scientist can perform data:read permission checks
+        - principal: 'Role::"data_scientist"'
+          action: 'Action::"data:read"'
+          resource: 'Resource::"*"'
+        # HTTP-level: team_admin can perform multiple permission checks
+        - principal: 'Role::"team_admin"'
+          action:
+            - 'Action::"data:read"'
+            - 'Action::"security:read"'
+            - 'Action::"tools:read"'
+          resource: 'Resource::"*"'
+        # Tool-level: data_scientist can call ping and flaky_tool on any server
+        - principal: 'Role::"data_scientist"'
+          action:
+            - 'Action::"ping"'
+            - 'Action::"flaky_tool"'
+            - 'Action::"recovers_quickly"'
+          resource: 'Server::"*"'
+        # Tool-level: team_admin can call all dummy_mcp tools
+        - principal: 'Role::"team_admin"'
+          action:
+            - 'Action::"ping"'
+            - 'Action::"flaky_tool"'
+            - 'Action::"reset_counter"'
+            - 'Action::"status_code_tool"'
+            - 'Action::"recovers_quickly"'
+            - 'Action::"always_fails"'
+          resource: 'Server::"*"'
+
   # Argument Normalizer - stabilize inputs before anything else
   - name: "ArgumentNormalizer"
     kind: "plugins.argument_normalizer.argument_normalizer.ArgumentNormalizerPlugin"
diff --git a/plugins/external/cedar/cedarpolicyplugin/plugin-manifest.yaml b/plugins/external/cedar/cedarpolicyplugin/plugin-manifest.yaml
index 38ec3ceea82..da4abb320aa 100644
--- a/plugins/external/cedar/cedarpolicyplugin/plugin-manifest.yaml
+++ b/plugins/external/cedar/cedarpolicyplugin/plugin-manifest.yaml
@@ -2,6 +2,7 @@ description: "A plugin that does policy decision and enforcement using cedar"
 author: "Shriti Priya"
 version: "0.1.0"
 available_hooks:
+  - "http_auth_check_permission"
   - "prompt_pre_hook"
   - "prompt_post_hook"
   - "tool_pre_hook"
diff --git a/plugins/external/cedar/cedarpolicyplugin/plugin.py b/plugins/external/cedar/cedarpolicyplugin/plugin.py
index 37e5e87e148..b34f35d598f 100644
--- a/plugins/external/cedar/cedarpolicyplugin/plugin.py
+++ b/plugins/external/cedar/cedarpolicyplugin/plugin.py
@@ -29,6 +29,7 @@
     PluginContext,
     PluginError,
     PluginErrorModel,
+    PluginMode,
     PluginViolation,
     PromptPosthookPayload,
     PromptPosthookResult,
@@ -46,6 +47,11 @@
     ToolPreInvokePayload,
     ToolPreInvokeResult,
 )
+from mcpgateway.plugins.framework.hooks.http import (
+    HttpAuthCheckPermissionPayload,
+    HttpAuthCheckPermissionResult,
+    HttpAuthCheckPermissionResultPayload,
+)
 
 # Initialize logging service first
 logger = logging.getLogger(__name__)
@@ -283,13 +289,14 @@ def _preprocess_request(self, user: Any = None, action: str = "", resource: str
         if context is None:
             context = {}
         user_role = None
-        if "tool" in hook_type:
+        _hook_lower = hook_type.lower()
+        if "tool" in _hook_lower:
             resource_expr = CedarResourceTemplates.SERVER.format(resource_type=resource)
-        elif "agent" in hook_type:
+        elif "agent" in _hook_lower:
             resource_expr = CedarResourceTemplates.AGENT.format(resource_type=resource)
-        elif "resource" in hook_type:
+        elif "resource" in _hook_lower:
             resource_expr = CedarResourceTemplates.RESOURCE.format(resource_type=resource)
-        elif "prompt" in hook_type:
+        elif "prompt" in _hook_lower:
             resource_expr = CedarResourceTemplates.PROMPT.format(resource_type=resource)
         else:
             logger.error(f"{CedarErrorCodes.UNSUPPORTED_RESOURCE_TYPE.value}: {hook_type}")
@@ -308,6 +315,49 @@ def _preprocess_request(self, user: Any = None, action: str = "", resource: str
         request = CedarInput(principal=principal_expr, action=action_expr, resource=resource_expr, context=context, correlation_id=correlation_id).model_dump()
         return request
 
+    def _preprocess_requests_for_roles(self, user: Any, action: str, resource: str, hook_type: str) -> list[dict]:
+        """Build one CedarInput per role the user holds, for role-aware evaluation.
+
+        Falls back to a User:: principal when the user has no roles assigned.
+        Used by hooks that need per-role any-allow semantics (e.g. tool_pre_invoke).
+
+        Args:
+            user: The user dict from global_context (must have 'email' and optionally 'roles', 'is_admin').
+            action: Cedar action string (e.g. the tool name).
+            resource: Cedar resource string (e.g. server id).
+            hook_type: Hook type string used to pick the resource template.
+
+        Returns:
+            List of CedarInput dicts, one per principal to evaluate.
+        """
+        if user is None:
+            user = {}
+        _hook_lower = hook_type.lower()
+        if "tool" in _hook_lower:
+            resource_expr = CedarResourceTemplates.SERVER.format(resource_type=resource)
+        elif "agent" in _hook_lower:
+            resource_expr = CedarResourceTemplates.AGENT.format(resource_type=resource)
+        elif "resource" in _hook_lower:
+            resource_expr = CedarResourceTemplates.RESOURCE.format(resource_type=resource)
+        elif "prompt" in _hook_lower:
+            resource_expr = CedarResourceTemplates.PROMPT.format(resource_type=resource)
+        else:
+            logger.error(f"{CedarErrorCodes.UNSUPPORTED_RESOURCE_TYPE.value}: {hook_type}")
+            raise PluginError(PluginErrorModel(message=CedarErrorCodes.UNSUPPORTED_RESOURCE_TYPE.value, plugin_name="CedarPolicyPlugin"))
+
+        action_expr = f'Action::"{action}"'
+        roles: list[str] = user.get("roles", []) if isinstance(user, dict) else []
+        if roles:
+            principals = [f'Role::"{r}"' for r in roles]
+        else:
+            email = user.get("email", "unknown") if isinstance(user, dict) else str(user)
+            principals = [f'User::"{email}"']
+
+        return [
+            CedarInput(principal=p, action=action_expr, resource=resource_expr, context={}).model_dump(exclude_none=True)
+            for p in principals
+        ]
+
     def _redact_output(self, payload: Any) -> Any:
         """Function that redacts the output of prompt, tool or resource
         NOTE: It's an extremely simple logic for redaction, could be replaced with more advanced
@@ -461,6 +511,11 @@ async def prompt_post_fetch(self, payload: PromptPosthookPayload, context: Plugi
     async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
         """Plugin hook run before a tool is invoked.
 
+        Evaluates role-based Cedar policy for tool invocation using the same
+        any-allow semantics as http_auth_check_permission: if any of the user's
+        roles is permitted, the tool call proceeds.  Admin users bypass Cedar.
+        Permissive mode logs denials but never blocks.
+
         Args:
             payload: The tool payload to be analyzed.
             context: Contextual information about the hook call.
@@ -469,39 +524,75 @@ async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginCo
             The result of the plugin's analysis, including whether the tool can proceed.
         """
         hook_type = ToolHookType.TOOL_PRE_INVOKE
-        logger.info(f"Processing {hook_type} for '{payload.args}' with {len(payload.args) if payload.args else 0}")
-        logger.info(f"Processing context {context}")
 
-        if not payload.args:
-            return ToolPreInvokeResult()
+        if not self._cedar_policy:
+            return ToolPreInvokeResult(continue_processing=True)
 
-        user = ""
-        server_id = ""
-        if context.global_context.user:
-            user = context.global_context.user
+        user = context.global_context.user if context.global_context.user else {}
 
+        # Admins bypass Cedar evaluation
+        if isinstance(user, dict) and user.get("is_admin"):
+            logger.debug("[CedarPolicyPlugin] tool_pre_invoke: admin %s — bypassing Cedar", user.get("email"))
+            return ToolPreInvokeResult(continue_processing=True)
+
+        server_id = ""
         if "gateway" in context.global_context.metadata and "name" in context.global_context.metadata["gateway"]:
             server_id = context.global_context.metadata["gateway"]["name"]
         else:
-            server_id = context.global_context.server_id
+            server_id = context.global_context.server_id or ""
 
-        if server_id:
-            request = self._preprocess_request(user, payload.name, server_id, hook_type)
-        else:
+        if not server_id:
             logger.error(f"{CedarErrorCodes.UNSPECIFIED_SERVER.value}")
             raise PluginError(PluginErrorModel(message=CedarErrorCodes.UNSPECIFIED_SERVER.value, plugin_name="CedarPolicyPlugin"))
 
-        if self._cedar_policy:
-            decision = await self._evaluate_policy(request, self._cedar_policy)
-            if decision == "Deny":
-                violation = PluginViolation(
-                    reason=CedarResponseTemplates.CEDAR_REASON.format(hook_type=hook_type),
-                    description=CedarResponseTemplates.CEDAR_DESC.format(hook_type=hook_type),
-                    code=CedarCodes.DENIAL_CODE,
-                    details={},
-                )
-                return ToolPreInvokeResult(modified_payload=payload, violation=violation, continue_processing=False)
-        return ToolPreInvokeResult(continue_processing=True)
+        # Legacy jwt_info path: user is a string username (e.g. from older tests/setups)
+        if not isinstance(user, dict) and len(self.jwt_info) > 0 and "users" in self.jwt_info:
+            try:
+                single_req = self._preprocess_request(user, payload.name, server_id, str(hook_type), context={})
+                requests = [single_req]
+            except PluginError as exc:
+                logger.warning("[CedarPolicyPlugin] tool_pre_invoke jwt_info preprocess error: %s", exc)
+                requests = self._preprocess_requests_for_roles(user, payload.name, server_id, str(hook_type))
+        else:
+            requests = self._preprocess_requests_for_roles(user, payload.name, server_id, str(hook_type))
+
+        last_denying_principal = requests[0]["principal"] if requests else "unknown"
+        for cedar_request in requests:
+            try:
+                decision = await self._evaluate_policy(cedar_request, self._cedar_policy)
+            except PluginError as exc:
+                logger.warning("[CedarPolicyPlugin] tool_pre_invoke evaluation error principal=%s: %s — treating as Deny", cedar_request["principal"], exc)
+                decision = "Deny"
+
+            logger.info(
+                "[CedarPolicyPlugin] tool_pre_invoke: user=%s principal=%s tool=%s server=%s => %s",
+                user.get("email") if isinstance(user, dict) else user,
+                cedar_request["principal"],
+                payload.name,
+                server_id,
+                decision,
+            )
+
+            if decision != "Deny":
+                return ToolPreInvokeResult(continue_processing=True)
+
+            last_denying_principal = cedar_request["principal"]
+
+        # All roles denied
+        deny_reason = f"Cedar denied tool '{payload.name}' on server '{server_id}' for all roles (last principal: {last_denying_principal})"
+        logger.info("[CedarPolicyPlugin] tool_pre_invoke DENIED: %s", deny_reason)
+
+        if self._config.mode != PluginMode.ENFORCE:
+            logger.info("[CedarPolicyPlugin] Permissive mode — not enforcing Cedar denial for tool_pre_invoke")
+            return ToolPreInvokeResult(continue_processing=True)
+
+        violation = PluginViolation(
+            reason=CedarResponseTemplates.CEDAR_REASON.format(hook_type=hook_type),
+            description=deny_reason,
+            code=CedarCodes.DENIAL_CODE,
+            details={},
+        )
+        return ToolPreInvokeResult(modified_payload=payload, violation=violation, continue_processing=False)
 
     async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: PluginContext) -> ToolPostInvokeResult:
         """Plugin hook run after a tool is invoked.
@@ -668,3 +759,82 @@ async def resource_post_fetch(self, payload: ResourcePostFetchPayload, context:
                 raise PluginError(PluginErrorModel(message=CedarErrorCodes.UNSPECIFIED_OUTPUT_ACTION.value, plugin_name="CedarPolicyPlugin"))
 
         return ResourcePostFetchResult(continue_processing=True)
+
+    async def http_auth_check_permission(self, payload: HttpAuthCheckPermissionPayload, context: PluginContext) -> HttpAuthCheckPermissionResult:
+        """Cedar policy hook for HTTP permission checks.
+
+        Evaluates whether a user (by roles) may perform an action using the
+        Cedar policy defined in this plugin's config.  Each role the user holds
+        is evaluated independently — a Cedar Allow for *any* role is sufficient
+        to permit the action (fail-open per role, fail-closed only when all
+        roles are denied).
+
+        Permissive mode: logs the Cedar decision, always passes through.
+        Enforce mode: returns Deny if Cedar denies for *all* roles.
+
+        Args:
+            payload: Contains user_email, roles, permission, is_admin.
+            context: Plugin execution context.
+
+        Returns:
+            HttpAuthCheckPermissionResult.
+        """
+        if not self._cedar_policy:
+            logger.debug("[CedarPolicyPlugin] No cedar policy configured — skipping http_auth_check_permission")
+            return HttpAuthCheckPermissionResult(continue_processing=True)
+
+        # Admins bypass Cedar evaluation
+        if payload.is_admin:
+            logger.debug("[CedarPolicyPlugin] Admin %s — bypassing Cedar", payload.user_email)
+            return HttpAuthCheckPermissionResult(continue_processing=True)
+
+        action = f'Action::"{payload.permission}"'
+        resource = 'Resource::"*"'
+
+        # Evaluate each role; Allow for any role is sufficient
+        roles_to_check = payload.roles if payload.roles else []
+        # Fall back to user-principal evaluation when the user has no roles
+        principals = [f'Role::"{r}"' for r in roles_to_check] or [f'User::"{payload.user_email}"']
+
+        last_denying_principal = principals[0]
+        for principal in principals:
+            cedar_request = {"principal": principal, "action": action, "resource": resource, "context": {}}
+            try:
+                decision = await self._evaluate_policy(cedar_request, self._cedar_policy)
+            except PluginError as exc:
+                logger.warning(
+                    "[CedarPolicyPlugin] Evaluation error for %s/%s principal=%s: %s — treating as Deny",
+                    payload.user_email, payload.permission, principal, exc,
+                )
+                decision = "Deny"
+
+            logger.info(
+                "[CedarPolicyPlugin] http_auth: user=%s principal=%s permission=%s => %s",
+                payload.user_email,
+                principal,
+                payload.permission,
+                decision,
+            )
+
+            if decision != "Deny":
+                # At least one role is permitted — allow immediately
+                return HttpAuthCheckPermissionResult(continue_processing=True)
+
+            last_denying_principal = principal
+
+        # All role evaluations denied
+        deny_reason = f"Cedar denied '{payload.permission}' for all roles of {payload.user_email} (last principal: {last_denying_principal})"
+        logger.info("[CedarPolicyPlugin] http_auth DENIED: %s", deny_reason)
+
+        if self._config.mode != PluginMode.ENFORCE:
+            # Permissive / audit mode — log and pass through without blocking
+            logger.info("[CedarPolicyPlugin] Permissive mode — not enforcing Cedar denial for %s", payload.user_email)
+            return HttpAuthCheckPermissionResult(continue_processing=True)
+
+        return HttpAuthCheckPermissionResult(
+            continue_processing=False,
+            modified_payload=HttpAuthCheckPermissionResultPayload(
+                granted=False,
+                reason=deny_reason,
+            ),
+        )
diff --git a/plugins/external/cedar/tests/test_cedarpolicyplugin.py b/plugins/external/cedar/tests/test_cedarpolicyplugin.py
index 34be7f115e4..711c2a2aa57 100644
--- a/plugins/external/cedar/tests/test_cedarpolicyplugin.py
+++ b/plugins/external/cedar/tests/test_cedarpolicyplugin.py
@@ -640,3 +640,310 @@ async def test_cedarpolicyplugin_resource_post_fetch_custom_dsl_rbac():
     assert allow_count == 2
     assert deny_count == 1
     assert redact_count == 1
+
+
+# ---------------------------------------------------------------------------
+# http_auth_check_permission tests
+# ---------------------------------------------------------------------------
+
+CEDAR_HTTP_POLICY = [
+    {
+        "principal": 'Role::"data_scientist"',
+        "action": ['Action::"data:read"', 'Action::"tools:read"'],
+        "resource": 'Resource::"*"',
+    },
+    {
+        "principal": 'Role::"team_admin"',
+        "action": ['Action::"data:read"', 'Action::"security:read"', 'Action::"tools:read"'],
+        "resource": 'Resource::"*"',
+    },
+]
+
+
+def _make_cedar_http_plugin(policy=None, mode="permissive"):
+    """Helper: build a CedarPolicyPlugin configured for HTTP permission checks."""
+    config = PluginConfig(
+        name="test-cedar-http",
+        kind="cedarpolicyplugin.plugin.CedarPolicyPlugin",
+        hooks=["http_auth_check_permission"],
+        mode=mode,
+        config={"policy_lang": "cedar", "policy": policy if policy is not None else CEDAR_HTTP_POLICY},
+    )
+    return CedarPolicyPlugin(config)
+
+
+def _make_http_context():
+    return PluginContext(global_context=GlobalContext(request_id="test-req"))
+
+
+from mcpgateway.plugins.framework.hooks.http import HttpAuthCheckPermissionPayload  # noqa: E402
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_allow_matching_role_and_permission():
+    """data_scientist requesting data:read should be allowed."""
+    plugin = _make_cedar_http_plugin()
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="alice@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["data_scientist"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is True
+    assert result.modified_payload is None
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_deny_role_without_permission():
+    """viewer role has no Cedar permit — enforce mode must deny."""
+    plugin = _make_cedar_http_plugin(mode="enforce")
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="bob@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["viewer"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is False
+    assert result.modified_payload is not None
+    assert result.modified_payload.granted is False
+    assert result.modified_payload.reason is not None
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_admin_bypass_skips_cedar():
+    """is_admin=True must bypass Cedar entirely and allow."""
+    plugin = _make_cedar_http_plugin()
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="admin@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=True,
+        roles=[],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is True
+    assert result.modified_payload is None
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_no_policy_passes_through():
+    """When _cedar_policy is None (no compiled policy), must pass through without blocking."""
+    plugin = _make_cedar_http_plugin()
+    plugin._cedar_policy = None  # simulate unconfigured state
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="carol@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["data_scientist"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is True
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_any_role_allows_access():
+    """Any permitted role is sufficient: first role denied, second role allowed — should allow."""
+    plugin = _make_cedar_http_plugin(mode="enforce")
+    # viewer is denied; data_scientist is permitted for data:read
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="dave@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["viewer", "data_scientist"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is True
+    assert result.modified_payload is None
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_first_role_allows_short_circuits():
+    """When first role is permitted, remaining roles are not evaluated (any-allow wins)."""
+    plugin = _make_cedar_http_plugin(mode="enforce")
+    # First role is team_admin (has security:read permit), second is viewer (doesn't)
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="dave@example.com",
+        permission="security:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["team_admin", "viewer"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is True
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_permissive_mode_passes_through_on_denial():
+    """Permissive mode: even when Cedar denies all roles, the request passes through."""
+    plugin = _make_cedar_http_plugin(mode="permissive")
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="eve@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=["viewer"],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    # Permissive mode must not block, regardless of Cedar decision
+    assert result.continue_processing is True
+
+
+@pytest.mark.asyncio
+async def test_cedar_http_auth_no_roles_falls_back_to_user_principal():
+    """User with no roles falls back to User:: principal; no Cedar permit — enforce denies."""
+    plugin = _make_cedar_http_plugin(mode="enforce")
+    payload = HttpAuthCheckPermissionPayload(
+        user_email="frank@example.com",
+        permission="data:read",
+        resource_type="tool",
+        team_id="team1",
+        is_admin=False,
+        roles=[],
+        auth_method="jwt",
+        client_host="127.0.0.1",
+        user_agent="test",
+    )
+    result = await plugin.http_auth_check_permission(payload, _make_http_context())
+    assert result.continue_processing is False
+    assert result.modified_payload.granted is False
+
+
+# ---------------------------------------------------------------------------
+# tool_pre_invoke role-aware tests
+# ---------------------------------------------------------------------------
+
+CEDAR_TOOL_POLICY = [
+    {
+        "principal": 'Role::"data_scientist"',
+        "action": ['Action::"ping"', 'Action::"flaky_tool"'],
+        "resource": 'Server::"dummy-server"',
+    },
+    {
+        "principal": 'Role::"team_admin"',
+        "action": ['Action::"ping"', 'Action::"reset_counter"'],
+        "resource": 'Server::"dummy-server"',
+    },
+]
+
+
+def _make_cedar_tool_plugin(mode="enforce"):
+    config = PluginConfig(
+        name="test-cedar-tool",
+        kind="cedarpolicyplugin.plugin.CedarPolicyPlugin",
+        hooks=["tool_pre_invoke"],
+        mode=mode,
+        config={"policy_lang": "cedar", "policy": CEDAR_TOOL_POLICY},
+    )
+    return CedarPolicyPlugin(config)
+
+
+def _make_tool_context(user_dict, server_id="dummy-server"):
+    return PluginContext(global_context=GlobalContext(request_id="test-req", server_id=server_id, user=user_dict))
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_role_allowed():
+    """data_scientist calling ping should be allowed."""
+    plugin = _make_cedar_tool_plugin()
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "alice@example.com", "is_admin": False, "roles": ["data_scientist"]})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is True
+    assert result.violation is None
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_role_denied_enforce():
+    """viewer role has no Cedar tool permit — enforce mode blocks."""
+    plugin = _make_cedar_tool_plugin(mode="enforce")
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "bob@example.com", "is_admin": False, "roles": ["viewer"]})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is False
+    assert result.violation is not None
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_role_denied_permissive():
+    """viewer role has no Cedar tool permit — permissive mode passes through."""
+    plugin = _make_cedar_tool_plugin(mode="permissive")
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "bob@example.com", "is_admin": False, "roles": ["viewer"]})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is True
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_admin_bypass():
+    """Admin users bypass Cedar tool evaluation entirely."""
+    plugin = _make_cedar_tool_plugin(mode="enforce")
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "admin@example.com", "is_admin": True, "roles": []})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is True
+    assert result.violation is None
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_any_role_allows():
+    """viewer + data_scientist: viewer denied, data_scientist allowed — should allow."""
+    plugin = _make_cedar_tool_plugin(mode="enforce")
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "carol@example.com", "is_admin": False, "roles": ["viewer", "data_scientist"]})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is True
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_no_roles_denied():
+    """No roles falls back to User:: principal; not in policy — enforce blocks."""
+    plugin = _make_cedar_tool_plugin(mode="enforce")
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "nobody@example.com", "is_admin": False, "roles": []})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is False
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_no_policy_passes_through():
+    """When _cedar_policy is None, tool_pre_invoke passes through."""
+    plugin = _make_cedar_tool_plugin(mode="enforce")
+    plugin._cedar_policy = None
+    payload = ToolPreInvokePayload(name="ping", args={})
+    context = _make_tool_context({"email": "alice@example.com", "is_admin": False, "roles": ["data_scientist"]})
+    result = await plugin.tool_pre_invoke(payload, context)
+    assert result.continue_processing is True
diff --git a/tests/unit/mcpgateway/cache/test_auth_cache_l1_l2.py b/tests/unit/mcpgateway/cache/test_auth_cache_l1_l2.py
index ebe82f029e6..6854c534154 100644
--- a/tests/unit/mcpgateway/cache/test_auth_cache_l1_l2.py
+++ b/tests/unit/mcpgateway/cache/test_auth_cache_l1_l2.py
@@ -1088,3 +1088,72 @@ async def test_get_team_membership_redis_miss(self, auth_cache, mock_redis):
 
         assert result is None
         assert auth_cache._redis_miss_count == initial + 1
+
+
+class TestUserRolesListCache:
+    """Test get/set/invalidate_user_roles_list L1/L2 behaviour."""
+
+    @pytest.mark.asyncio
+    async def test_get_returns_none_on_miss(self, auth_cache):
+        """Fresh cache returns None for unknown user."""
+        result = await auth_cache.get_user_roles_list("test@example.com")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_set_then_get_returns_list_from_l1(self, auth_cache):
+        """Set populates L1; subsequent get returns the list without touching Redis."""
+        await auth_cache.set_user_roles_list("test@example.com", ["data_scientist", "team_admin"])
+        result = await auth_cache.get_user_roles_list("test@example.com")
+        assert result == ["data_scientist", "team_admin"]
+
+    @pytest.mark.asyncio
+    async def test_l2_hit_populates_l1_write_through(self, auth_cache, mock_redis):
+        """Redis hit returns JSON-encoded list and writes it through to L1."""
+        import json
+
+        mock_redis.get = AsyncMock(return_value=json.dumps(["data_scientist"]).encode())
+        with patch.object(auth_cache, "_get_redis_client", return_value=mock_redis):
+            result = await auth_cache.get_user_roles_list("test@example.com")
+
+        assert result == ["data_scientist"]
+        # L1 should now be populated so a second call returns from L1
+        result2 = await auth_cache.get_user_roles_list("test@example.com")
+        assert result2 == ["data_scientist"]
+
+    @pytest.mark.asyncio
+    async def test_invalidate_clears_l1(self, auth_cache):
+        """After invalidation get returns None again."""
+        await auth_cache.set_user_roles_list("test@example.com", ["developer"])
+        await auth_cache.invalidate_user_roles_list("test@example.com")
+        result = await auth_cache.get_user_roles_list("test@example.com")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_invalidate_deletes_redis_key(self, auth_cache, mock_redis):
+        """Invalidate calls Redis delete for the roles_list key."""
+        with patch.object(auth_cache, "_get_redis_client", return_value=mock_redis):
+            await auth_cache.invalidate_user_roles_list("test@example.com")
+        mock_redis.delete.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_get_returns_none_when_cache_disabled(self):
+        """Disabled cache always returns None."""
+        cache = AuthCache(enabled=False)
+        result = await cache.get_user_roles_list("test@example.com")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_set_does_not_raise_when_disabled(self):
+        """set_user_roles_list on disabled cache completes without error."""
+        cache = AuthCache(enabled=False)
+        await cache.set_user_roles_list("test@example.com", ["viewer"])
+        result = await cache.get_user_roles_list("test@example.com")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_l2_invalid_json_returns_none(self, auth_cache, mock_redis):
+        """Corrupt Redis value is treated as miss (no crash)."""
+        mock_redis.get = AsyncMock(return_value=b"not-valid-json")
+        with patch.object(auth_cache, "_get_redis_client", return_value=mock_redis):
+            result = await auth_cache.get_user_roles_list("test@example.com")
+        assert result is None
diff --git a/tests/unit/mcpgateway/services/test_role_service.py b/tests/unit/mcpgateway/services/test_role_service.py
index a9dd50e6dc0..f5f5933c7ac 100644
--- a/tests/unit/mcpgateway/services/test_role_service.py
+++ b/tests/unit/mcpgateway/services/test_role_service.py
@@ -592,6 +592,33 @@ async def test_assign_personal_role_with_scope_id(self, role_service, sample_rol
             with pytest.raises(ValueError, match="scope_id not allowed"):
                 await role_service.assign_role_to_user(user_email="user@example.com", role_id="role-123", scope="personal", scope_id="should-not-have", granted_by="admin@example.com")
 
+    @pytest.mark.asyncio
+    async def test_assign_role_invalidates_roles_list_cache(self, role_service, mock_db, sample_role, sample_user_role):
+        """Successful assignment invalidates the user roles list cache."""
+        mock_cache = AsyncMock()
+
+        with patch.object(role_service, "get_role_by_id", new=AsyncMock(return_value=sample_role)):
+            with patch.object(role_service, "get_user_role_assignment", new=AsyncMock(return_value=None)):
+                with patch("mcpgateway.services.role_service.UserRole", return_value=sample_user_role):
+                    with patch("mcpgateway.cache.auth_cache.auth_cache", mock_cache):
+                        await role_service.assign_role_to_user(user_email="user@example.com", role_id="role-123", scope="team", scope_id="team-789", granted_by="admin@example.com")
+
+        mock_cache.invalidate_user_roles_list.assert_called_once_with("user@example.com")
+
+    @pytest.mark.asyncio
+    async def test_assign_role_cache_invalidation_failure_is_non_fatal(self, role_service, mock_db, sample_role, sample_user_role):
+        """Cache invalidation failure on assign does not prevent successful assignment."""
+        mock_cache = AsyncMock()
+        mock_cache.invalidate_user_roles_list = AsyncMock(side_effect=Exception("Redis down"))
+
+        with patch.object(role_service, "get_role_by_id", new=AsyncMock(return_value=sample_role)):
+            with patch.object(role_service, "get_user_role_assignment", new=AsyncMock(return_value=None)):
+                with patch("mcpgateway.services.role_service.UserRole", return_value=sample_user_role):
+                    with patch("mcpgateway.cache.auth_cache.auth_cache", mock_cache):
+                        result = await role_service.assign_role_to_user(user_email="user@example.com", role_id="role-123", scope="team", scope_id="team-789", granted_by="admin@example.com")
+
+        assert result == sample_user_role
+
 
 class TestRevokeRoleFromUser:
     """Test revoke_role_from_user method."""
@@ -626,6 +653,32 @@ async def test_revoke_inactive_role(self, role_service, sample_user_role):
 
             assert result is False
 
+    @pytest.mark.asyncio
+    async def test_revoke_role_invalidates_roles_list_cache(self, role_service, sample_user_role):
+        """Successful revocation invalidates the user roles list cache."""
+        sample_user_role.is_active = True
+        mock_cache = AsyncMock()
+
+        with patch.object(role_service, "get_user_role_assignment", new=AsyncMock(return_value=sample_user_role)):
+            with patch("mcpgateway.cache.auth_cache.auth_cache", mock_cache):
+                result = await role_service.revoke_role_from_user(user_email="user@example.com", role_id="role-123", scope="team", scope_id="team-789")
+
+        assert result is True
+        mock_cache.invalidate_user_roles_list.assert_called_once_with("user@example.com")
+
+    @pytest.mark.asyncio
+    async def test_revoke_role_cache_invalidation_failure_is_non_fatal(self, role_service, sample_user_role):
+        """Cache invalidation failure on revoke does not prevent successful revocation."""
+        sample_user_role.is_active = True
+        mock_cache = AsyncMock()
+        mock_cache.invalidate_user_roles_list = AsyncMock(side_effect=Exception("Redis down"))
+
+        with patch.object(role_service, "get_user_role_assignment", new=AsyncMock(return_value=sample_user_role)):
+            with patch("mcpgateway.cache.auth_cache.auth_cache", mock_cache):
+                result = await role_service.revoke_role_from_user(user_email="user@example.com", role_id="role-123", scope="team", scope_id="team-789")
+
+        assert result is True
+
 
 class TestGetUserRoleAssignment:
     """Test get_user_role_assignment method."""