diff --git a/keepercommander/service/commands/integrations/sailpoint/command_hook.py b/keepercommander/service/commands/integrations/sailpoint/command_hook.py
index 30783a1ee..b079514bf 100644
--- a/keepercommander/service/commands/integrations/sailpoint/command_hook.py
+++ b/keepercommander/service/commands/integrations/sailpoint/command_hook.py
@@ -81,7 +81,7 @@ def before_command(self, params: KeeperParams, command: str) -> Tuple[str, Optio
 
         mutation = SailPointCommandParser.parse_identity_mutation(command)
         if mutation:
-            err = self._first_scim_identity_error(params, mutation.emails)
+            err = self._first_scim_identity_error(params, mutation.identifiers)
             if err:
                 return self._reject(command, err, 403)
         return command, None
diff --git a/keepercommander/service/commands/integrations/sailpoint/command_parse.py b/keepercommander/service/commands/integrations/sailpoint/command_parse.py
index a90b22b62..9de73cd92 100644
--- a/keepercommander/service/commands/integrations/sailpoint/command_parse.py
+++ b/keepercommander/service/commands/integrations/sailpoint/command_parse.py
@@ -63,7 +63,7 @@ def is_grant(self) -> bool:
 class ParsedIdentityMutation:
     """enterprise-user identity change (role/team/node) — used for SCIM coexistence."""
 
-    emails: List[str] = field(default_factory=list)
+    identifiers: List[str] = field(default_factory=list)
     has_role_change: bool = False
     has_team_change: bool = False
     has_node_change: bool = False
@@ -134,14 +134,15 @@ def parse_identity_mutation(cls, command: str) -> Optional[ParsedIdentityMutatio
         if not parsed:
             return None
         ns, _unknown = parsed
-        emails = [e for e in (ns.email or []) if isinstance(e, str) and '@' in e]
+        # No '@' filter here (unlike parse_invite/parse_transfer): the SCIM guard must see IDs and '@all' too.
+        identifiers = [e for e in (ns.email or []) if isinstance(e, str) and e.strip()]
         has_role = bool(ns.add_role or ns.remove_role)
         has_team = bool(ns.add_team or ns.remove_team)
         has_node = bool(ns.node)
-        if not (has_role or has_team or has_node) or not emails:
+        if not (has_role or has_team or has_node) or not identifiers:
             return None
         return ParsedIdentityMutation(
-            emails=emails,
+            identifiers=identifiers,
             has_role_change=has_role,
             has_team_change=has_team,
             has_node_change=has_node,
diff --git a/keepercommander/service/commands/integrations/sailpoint/scim_guard.py b/keepercommander/service/commands/integrations/sailpoint/scim_guard.py
index 0e2a42812..b774fc192 100644
--- a/keepercommander/service/commands/integrations/sailpoint/scim_guard.py
+++ b/keepercommander/service/commands/integrations/sailpoint/scim_guard.py
@@ -13,11 +13,13 @@
 
 from __future__ import annotations
 
-from typing import Iterable, Optional, Set
+from typing import Any, Dict, Iterable, Optional, Set
 
 from ..... import api
 from .....params import KeeperParams
 
+_ALL_PSEUDO_USER = '@all'
+
 
 class SailPointScimGuard:
     """Detect SCIM-managed users/nodes and block identity stomps."""
@@ -40,15 +42,21 @@ def _scim_node_ids(params: KeeperParams) -> Set[int]:
         return nodes
 
     @staticmethod
-    def _node_ancestors(params: KeeperParams, node_id: int) -> Iterable[int]:
-        by_id = {int(n['node_id']): n for n in params.enterprise.get('nodes') or []}
+    def _node_parents(params: KeeperParams) -> Dict[int, Optional[int]]:
+        return {
+            int(n['node_id']): (int(n['parent_id']) if n.get('parent_id') else None)
+            for n in params.enterprise.get('nodes') or []
+        }
+
+    @staticmethod
+    def _walk_ancestors(node_id: int, parents: Dict[int, Optional[int]]) -> Iterable[int]:
         current: Optional[int] = node_id
         seen = set()
         while current and current not in seen:
             yield current
             seen.add(current)
-            parent = by_id.get(current, {}).get('parent_id')
-            current = int(parent) if parent and int(parent) != current else None
+            parent = parents.get(current)
+            current = parent if parent != current else None
 
     @classmethod
     def is_scim_managed_node(cls, params: KeeperParams, node_id: Optional[int]) -> bool:
@@ -58,35 +66,82 @@ def is_scim_managed_node(cls, params: KeeperParams, node_id: Optional[int]) -> b
         scim_nodes = cls._scim_node_ids(params)
         if not scim_nodes:
             return False
-        return any(n in scim_nodes for n in cls._node_ancestors(params, int(node_id)))
+        parents = cls._node_parents(params)
+        return any(n in scim_nodes for n in cls._walk_ancestors(int(node_id), parents))
+
+    @staticmethod
+    def _normalize(value: Any) -> str:
+        return str(value).strip().lower() if isinstance(value, str) else ''
 
     @classmethod
-    def find_user(cls, params: KeeperParams, email: str):
+    def _build_user_lookup(cls, params: KeeperParams) -> Dict[str, Any]:
+        """Mirrors EnterpriseUserCommand's own lookup: id, username, then aliases."""
+        lookup: Dict[str, Any] = {}
+        for user in params.enterprise.get('users') or []:
+            if 'enterprise_user_id' in user:
+                lookup[str(user['enterprise_user_id']).strip()] = user
+            username = cls._normalize(user.get('username'))
+            if username:
+                lookup[username] = user
+        for alias in params.enterprise.get('user_aliases') or []:
+            username = cls._normalize(alias.get('username'))
+            if username and username not in lookup:
+                user_id = str(alias.get('enterprise_user_id')).strip()
+                if user_id in lookup:
+                    lookup[username] = lookup[user_id]
+        return lookup
+
+    @classmethod
+    def _is_all_pseudo_user(cls, identifier: str) -> bool:
+        return cls._normalize(identifier) == _ALL_PSEUDO_USER
+
+    @classmethod
+    def find_user(cls, params: KeeperParams, identifier: str) -> Optional[Dict[str, Any]]:
         cls.ensure_enterprise(params)
         if not params.enterprise:
             return None
-        target = email.strip().lower()
-        return next(
-            (
-                user for user in params.enterprise.get('users') or []
-                if (user.get('username') or '').lower() == target
-            ),
-            None,
-        )
+        target = cls._normalize(identifier)
+        if not target:
+            return None
+        return cls._build_user_lookup(params).get(target)
 
     @classmethod
-    def is_scim_managed_user(cls, params: KeeperParams, email: str) -> bool:
-        user = cls.find_user(params, email)
+    def is_scim_managed_user(cls, params: KeeperParams, identifier: str) -> bool:
+        user = cls.find_user(params, identifier)
         if not user:
             return False
         return cls.is_scim_managed_node(params, user.get('node_id'))
 
     @classmethod
-    def identity_change_error(cls, params: KeeperParams, email: str) -> Optional[str]:
-        if cls.is_scim_managed_user(params, email):
+    def is_scim_managed_identifier(cls, params: KeeperParams, identifier: str) -> bool:
+        """Like is_scim_managed_user, but '@all' matches if any user is SCIM-managed."""
+        if not cls._is_all_pseudo_user(identifier):
+            return cls.is_scim_managed_user(params, identifier)
+        cls.ensure_enterprise(params)
+        if not params.enterprise:
+            return False
+        scim_nodes = cls._scim_node_ids(params)
+        if not scim_nodes:
+            return False
+        parents = cls._node_parents(params)
+        return any(
+            any(n in scim_nodes for n in cls._walk_ancestors(int(user['node_id']), parents))
+            for user in params.enterprise.get('users') or []
+            if user.get('node_id')
+        )
+
+    @classmethod
+    def identity_change_error(cls, params: KeeperParams, identifier: str) -> Optional[str]:
+        if not cls.is_scim_managed_identifier(params, identifier):
+            return None
+        if cls._is_all_pseudo_user(identifier):
             return (
-                f'User {email} is managed by an existing SCIM provider. '
+                '@all includes one or more users managed by an existing SCIM provider. '
                 'SailPoint may only change folder/record (and admin) entitlements; '
-                'node/team/role identity changes are not allowed.'
+                'node/team/role identity changes are not allowed for those users.'
             )
-        return None
+        return (
+            f'User {identifier} is managed by an existing SCIM provider. '
+            'SailPoint may only change folder/record (and admin) entitlements; '
+            'node/team/role identity changes are not allowed.'
+        )
diff --git a/keepercommander/service/commands/integrations/sailpoint/service.py b/keepercommander/service/commands/integrations/sailpoint/service.py
index a1010685b..6d5fd5ad4 100644
--- a/keepercommander/service/commands/integrations/sailpoint/service.py
+++ b/keepercommander/service/commands/integrations/sailpoint/service.py
@@ -25,14 +25,7 @@
 
 
 class SailPointService:
-    """
-    Entry point for SailPoint Service Mode.
-
-    Pending entitlements and SailPoint settings live on a dedicated vault
-    config record (``Commander Service Mode SailPoint Config``). Runtime
-    resolves that UID from ``SAILPOINT_RECORD`` (compose env) or
-    ``params.sailpoint_record_uid``. Docker config stays on ``COMMANDER_RECORD``.
-    """
+    """Entry point for SailPoint Service Mode; settings/pending entitlements live on the record pinned by SAILPOINT_RECORD."""
 
     PARAMS_ATTR = PARAMS_ATTR
 
@@ -50,6 +43,16 @@ def record_has_marker(cls, params: KeeperParams, record_uid: str) -> bool:
             for field in record.custom
         )
 
+    @classmethod
+    def _record_has_marker_safe(cls, params: KeeperParams, uid: str) -> bool:
+        """record_has_marker, but never raises; startup-only -- handle_command/after_command
+        must call record_has_marker directly so a check failure denies the request instead of silently skipping it."""
+        try:
+            return cls.record_has_marker(params, uid)
+        except Exception as e:
+            logger.warning(f'SailPoint: marker check failed for record {uid}: {e}')
+            return False
+
     @classmethod
     def record_uid(cls, params: Optional[KeeperParams] = None) -> Optional[str]:
         if params is not None:
@@ -68,22 +71,13 @@ def bind_params(cls, params: KeeperParams, record_uid: Optional[str] = None) ->
 
     @classmethod
     def maybe_enable(cls, params: KeeperParams, args) -> None:
-        """
-        Bind params and sanitize the Service Mode command allowlist when the
-        SailPoint config record has the integration marker.
-
-        Callers must gate on ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Bind params and sanitize the allowlist when the config record has the marker; callers must gate on SAILPOINT_RECORD first."""
         uid = cls.record_uid(params)
-        try:
-            if not cls.record_has_marker(params, uid):
-                logger.warning(
-                    f'{SAILPOINT_RECORD_ENV}={uid} is set but record is missing '
-                    f'{SAILPOINT_MARKER_FIELD}; SailPoint mode not enabled'
-                )
-                return
-        except Exception as e:
-            logger.warning(f'SailPoint marker check failed; mode not enabled: {e}')
+        if not cls._record_has_marker_safe(params, uid):
+            logger.warning(
+                f'{SAILPOINT_RECORD_ENV}={uid} is set but record is missing '
+                f'{SAILPOINT_MARKER_FIELD}; SailPoint mode not enabled'
+            )
             return
 
         cls.bind_params(params, uid)
@@ -98,25 +92,17 @@ def maybe_enable(cls, params: KeeperParams, args) -> None:
 
     @classmethod
     def start_background_services(cls) -> None:
-        """
-        Start the entitlement poller when SailPoint is enabled.
-
-        Callers must gate on ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Start the entitlement poller; callers must gate on SAILPOINT_RECORD before invoking this."""
         from ....core.globals import get_current_params
         params = get_current_params()
         if not params:
             logger.warning('SailPoint poller not started: Keeper params not loaded')
             return
         uid = cls.record_uid(params)
-        try:
-            if not cls.record_has_marker(params, uid):
-                logger.warning(
-                    f'SailPoint poller not started: record {uid} missing {SAILPOINT_MARKER_FIELD}'
-                )
-                return
-        except Exception as e:
-            logger.warning(f'SailPoint poller not started: marker check failed: {e}')
+        if not cls._record_has_marker_safe(params, uid):
+            logger.warning(
+                f'SailPoint poller not started: record {uid} missing {SAILPOINT_MARKER_FIELD}'
+            )
             return
         cls.bind_params(params, uid)
         try:
@@ -129,15 +115,12 @@ def start_background_services(cls) -> None:
     def handle_command(
         cls, params: KeeperParams, command: str
     ) -> Tuple[str, Optional[Tuple[Any, int]]]:
-        """
-        Prepare a SailPoint Service Mode command.
-
-        Returns ``(command_to_run, short_circuit)``. Callers must gate on
-        ``SAILPOINT_RECORD`` before invoking this.
-        """
+        """Prepare a SailPoint command, returning (command_to_run, short_circuit); callers must gate on SAILPOINT_RECORD first."""
         cls.bind_params(params)
         uid = cls.record_uid(params)
+        # Not swallowed here -- a check failure must deny the request, not silently skip gating.
         if not cls.record_has_marker(params, uid):
+            logger.debug(f'SailPoint: record {uid} has no active marker; skipping all command gating for this request')
             return command, None
         return SailPointCommandHook(uid).before_command(params, command)
 
@@ -146,6 +129,8 @@ def after_command(cls, params: KeeperParams, command: str, success: bool = True)
         """Callers must gate on ``SAILPOINT_RECORD`` before invoking this."""
         cls.bind_params(params)
         uid = cls.record_uid(params)
+        # Not swallowed here either -- callers turn a raised exception into an explicit 500.
         if not cls.record_has_marker(params, uid):
+            logger.debug(f'SailPoint: record {uid} has no active marker; skipping pending-entitlement queue for this request')
             return
         SailPointCommandHook(uid).after_command(params, command, success)
diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py
index f04a7152b..ae2e2dade 100644
--- a/keepercommander/service/util/command_util.py
+++ b/keepercommander/service/util/command_util.py
@@ -219,10 +219,15 @@ def blocked(error):
             if protected_command_error:
                 return blocked(protected_command_error)
 
-            sailpoint_enabled = bool((os.environ.get('SAILPOINT_RECORD') or '').strip())
+            sailpoint_uid = (os.environ.get('SAILPOINT_RECORD') or '').strip()
+            sailpoint_enabled = bool(sailpoint_uid)
 
-            with hide_from_record_cache(params, protected_uids), \
-                 hide_from_folder_cache(params, protected_folder_uids):
+            # Only SailPoint's own record is exempt from handle_command's guard; every other
+            # protected record (and the folder cache, which has no exemption) stays hidden throughout.
+            handle_command_uids = {uid: title for uid, title in protected_uids.items() if uid != sailpoint_uid}
+
+            with hide_from_folder_cache(params, protected_folder_uids), \
+                    hide_from_record_cache(params, handle_command_uids):
                 if sailpoint_enabled:
                     from ..commands.integrations.sailpoint.service import SailPointService
                     command, sailpoint_response = SailPointService.handle_command(params, command)
@@ -231,7 +236,8 @@ def blocked(error):
                         response = CommandExecutor.encrypt_response(response)
                         return response, status_code
 
-                return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command)
+                with hide_from_record_cache(params, protected_uids):
+                    return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command)
             response = return_value if return_value else printed_output
 
             # Debug logging with sanitization
@@ -247,7 +253,10 @@ def blocked(error):
 
             if status_code == 200 and sailpoint_enabled:
                 try:
-                    SailPointService.after_command(params, command, success=True)
+                    # Same rule as the pre-dispatch phase: every protected record except
+                    # SailPoint's own stays hidden here too.
+                    with hide_from_record_cache(params, handle_command_uids):
+                        SailPointService.after_command(params, command, success=True)
                 except Exception as e:
                     logger.error(f'SailPoint post-process failed: {sanitize_debug_data(str(e))}')
                     err = {
diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index 7d1b18c4b..9d3bbbced 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -16,6 +16,7 @@
 import contextlib
 import os
 from collections import UserDict
+from collections.abc import MutableMapping
 from typing import Dict, FrozenSet, Iterable, Optional, Set, Tuple
 
 # Each integration's own setup pins its config record's UID here, regardless of its title. Extend when a new integration gets an always-hidden record.
@@ -25,6 +26,7 @@
     'SLACK_RECORD': '<Slack config record>',
     'TEAMS_RECORD': '<Teams config record>',
     'GCHAT_RECORD': '<GChat config record>',
+    'SAILPOINT_RECORD': '<SailPoint config record>',
 }
 
 # uid-keyed caches resolve_single_record/load_pam_record fall back to when a UID isn't in record_cache.
@@ -70,6 +72,7 @@ def _protected_titles() -> Tuple[str, ...]:
     from ..commands.terraform_app_setup import TerraformSetupConstants
     from ..commands.integrations.slack_app_setup import SlackAppSetupCommand
     from ..commands.integrations.teams_app_setup import TeamsAppSetupCommand
+    from ..commands.integrations.sailpoint_app_setup import SailPointAppSetupCommand
     from ..docker.models import DockerSetupConstants, GChatConstants
     return (
         *SERVICE_CONFIG_RECORD_TITLES,
@@ -78,6 +81,7 @@ def _protected_titles() -> Tuple[str, ...]:
         GChatConstants.DEFAULT_RECORD_NAME,
         SlackAppSetupCommand().get_default_record_name(),
         TeamsAppSetupCommand().get_default_record_name(),
+        SailPointAppSetupCommand().get_default_record_name(),
     )
 
 
@@ -206,7 +210,7 @@ def hide_from_record_cache(params, protected_uids: Dict[str, str]):
     saved_entries = {}
     for attr in _GUARDED_CACHE_ATTRS:
         source = getattr(params, attr, None)
-        if not isinstance(source, dict):
+        if not isinstance(source, MutableMapping):
             continue
         original_caches[attr] = source
         saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
@@ -283,7 +287,7 @@ def hide_from_folder_cache(params, protected_folder_uids: Set[str]):
 
         for attr in _GUARDED_FOLDER_CACHE_ATTRS:
             source = getattr(params, attr, None)
-            if not isinstance(source, dict):
+            if not isinstance(source, MutableMapping):
                 continue
             original_caches[attr] = source
             saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py
index 9d8ea389b..bd5238d29 100644
--- a/keepercommander/service/util/verified_command.py
+++ b/keepercommander/service/util/verified_command.py
@@ -40,6 +40,11 @@ class Verifycommand:
         "The '--' argument separator is not permitted through Service Mode"
     )
 
+    # expand_cmd_args (commands/base.py) substitutes ${VARNAME} after this check runs but before execution -- ban it outright.
+    _ENV_VAR_EXPANSION_MSG = (
+        "The '${VARNAME}' environment-variable syntax is not permitted through Service Mode"
+    )
+
     # WARNING: everything below is a DENYLIST. Any command/flag that reads or
     # writes a host file and is NOT enumerated here is allowed by default.
     # Adding a new command with local file I/O? Add it here, or it silently
@@ -89,6 +94,7 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None):
 
         for validator in (
             Verifycommand.validate_service_mode_double_dash,
+            Verifycommand.validate_service_mode_env_var_expansion_command,
             Verifycommand.validate_service_mode_legacy_command,
             Verifycommand.validate_service_mode_pam_tunnel_command,
             Verifycommand.validate_service_mode_download_attachment_command,
@@ -137,6 +143,14 @@ def validate_service_mode_double_dash(command_tokens, request_temp_dir=None):
             return Verifycommand._DOUBLE_DASH_MSG
         return None
 
+    @staticmethod
+    def validate_service_mode_env_var_expansion_command(command_tokens, request_temp_dir=None):
+        """Block '${VARNAME}' anywhere in Service Mode input; error or None."""
+        from ...commands.base import parameter_pattern
+        if any(parameter_pattern.search(tok) for tok in command_tokens):
+            return Verifycommand._ENV_VAR_EXPANSION_MSG
+        return None
+
     @staticmethod
     def validate_service_mode_legacy_command(command_tokens, request_temp_dir=None):
         """Block legacy commands in Service Mode; error or None."""
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index de0736de8..c1ffdbf1b 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -247,33 +247,161 @@ def test_protected_record_check_runs_for_every_command(self):
             CommandExecutor.execute('whoami')
             mock_get_uids.assert_called_once()
 
-    def test_sailpoint_handling_runs_inside_the_record_cache_guard(self):
-        """SailPoint's own pre-processing (handle_command) can resolve/act on
-        records before cli.do_command ever runs -- it must run with the guard
-        already active, not before it, or a folder/recursive share under
-        SailPoint mode could reach the protected record before it's hidden."""
+    def test_sailpoint_handling_runs_before_the_record_cache_guard(self):
+        """handle_command needs its OWN record visible to read its marker/capability fields.
+        Safe because Layer B already blocks direct references, and _before_share only resolves exact cache keys."""
         params = _params_with_protected_and_normal_record()
         seen_during_handle_command = {}
+        seen_during_dispatch = {}
 
         def fake_handle_command(p, command):
             seen_during_handle_command['keys'] = set(p.record_cache.keys())
             return command, None
 
+        def fake_capture(p, command):
+            seen_during_dispatch['keys'] = set(p.record_cache.keys())
+            return 'ok', 'ok', ''
+
         with mock.patch(
             'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
-        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch(
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': PROTECTED_UID}), mock.patch(
             'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
             side_effect=fake_handle_command,
         ), mock.patch.object(
-            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
+            CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture,
         ):
             CommandExecutor.execute(f'get {NORMAL_UID}')
 
-        self.assertNotIn(PROTECTED_UID, seen_during_handle_command['keys'])
+        # Unhidden for SailPoint's own gate -- it needs to read its own config record.
+        self.assertIn(PROTECTED_UID, seen_during_handle_command['keys'])
         self.assertIn(NORMAL_UID, seen_during_handle_command['keys'])
-        # Restored after the whole guarded block exits, same as the non-SailPoint case.
+        # Hidden again for the actual dispatched command -- an admin still can't reach it.
+        self.assertNotIn(PROTECTED_UID, seen_during_dispatch['keys'])
+        self.assertIn(NORMAL_UID, seen_during_dispatch['keys'])
+        # Restored after the whole request completes.
         self.assertIn(PROTECTED_UID, params.record_cache)
 
+    def test_other_protected_records_stay_hidden_from_sailpoint_handle_command(self):
+        """Regression test: only SailPoint's own pinned UID is exempt from handle_command's guard;
+        every other protected record (PROTECTED_UID standing in for e.g. Docker's) must stay hidden."""
+        params = _params_with_protected_and_normal_record()
+        seen_during_handle_command = {}
+
+        def fake_handle_command(p, command):
+            seen_during_handle_command['keys'] = set(p.record_cache.keys())
+            return command, None
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-own-uid'}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
+            side_effect=fake_handle_command,
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', ''),
+        ):
+            CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertNotIn(PROTECTED_UID, seen_during_handle_command['keys'])
+        self.assertIn(NORMAL_UID, seen_during_handle_command['keys'])
+
+    def test_direct_reference_to_protected_record_never_reaches_sailpoint_handle_command(self):
+        """Layer B blocks a command that directly names the protected UID/title before
+        SailPoint's handle_command ever runs, regardless of the record being unhidden for it."""
+        params = _params_with_protected_and_normal_record()
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
+        ) as mock_handle_command, mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', ''),
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(f'get {PROTECTED_UID}')
+
+        self.assertEqual(status_code, 403)
+        mock_handle_command.assert_not_called()
+        mock_capture.assert_not_called()
+
+    def test_sailpoint_marker_check_failure_denies_the_request(self):
+        """A transient marker-read failure must deny the request, not silently skip SailPoint's gating."""
+        params = _params_with_protected_and_normal_record()
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.record_has_marker',
+            side_effect=RuntimeError('decrypt failed'),
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', ''),
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertEqual(status_code, 500)
+        mock_capture.assert_not_called()
+
+    def test_sailpoint_own_record_stays_hidden_during_dispatch_when_another_record_is_also_protected(self):
+        """Regression test: when a second protected record forces the outer (handle_command)
+        guard to actually wrap record_cache, the inner (dispatch) guard must still hide
+        SailPoint's own UID too, not silently no-op because the cache is no longer a plain dict."""
+        sailpoint_uid = 'SAILPOINT_OWN_UID'
+        params = params_module.KeeperParams()
+        params.service_mode = False
+        params.record_cache = {
+            sailpoint_uid: _record_cache_entry(sailpoint_uid, 'Commander Service Mode SailPoint Config'),
+            PROTECTED_UID: _record_cache_entry(PROTECTED_UID, PROTECTED_TITLE),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        seen_during_dispatch = {}
+
+        def fake_capture(p, command):
+            seen_during_dispatch['keys'] = set(p.record_cache.keys())
+            return 'ok', 'ok', ''
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': sailpoint_uid}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
+            side_effect=lambda p, command: (command, None),
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture,
+        ):
+            CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertNotIn(sailpoint_uid, seen_during_dispatch['keys'])
+        self.assertNotIn(PROTECTED_UID, seen_during_dispatch['keys'])
+        self.assertIn(NORMAL_UID, seen_during_dispatch['keys'])
+
+    def test_other_protected_records_stay_hidden_during_sailpoint_after_command(self):
+        """after_command follows the same rule as handle_command: SailPoint's own record stays
+        visible (it writes the pending-entitlement queue there), every other protected record stays hidden."""
+        sailpoint_uid = 'SAILPOINT_OWN_UID'
+        params = params_module.KeeperParams()
+        params.service_mode = False
+        params.record_cache = {
+            sailpoint_uid: _record_cache_entry(sailpoint_uid, 'Commander Service Mode SailPoint Config'),
+            PROTECTED_UID: _record_cache_entry(PROTECTED_UID, PROTECTED_TITLE),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        seen_during_after_command = {}
+
+        def fake_after_command(p, command, success=True):
+            seen_during_after_command['keys'] = set(p.record_cache.keys())
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': sailpoint_uid}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
+            side_effect=lambda p, command: (command, None),
+        ), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.after_command',
+            side_effect=fake_after_command,
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', ''),
+        ):
+            CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertIn(sailpoint_uid, seen_during_after_command['keys'])
+        self.assertNotIn(PROTECTED_UID, seen_during_after_command['keys'])
+        self.assertIn(NORMAL_UID, seen_during_after_command['keys'])
+
 
 class TestSyncDownExemptionCommandExecution(TestCase):
     """slack-app-setup --sync-down must keep working for its OWN config record while every
@@ -420,6 +548,7 @@ def test_normal_record_is_unaffected(self):
         self.assertEqual(status_code, 200)
         mock_capture.assert_called_once()
 
+
 class TestProtectedFolderCommandExecution(TestCase):
     """The shared folder holding a protected config record must be just as unreachable
     as the record itself -- ls/tree/rndir/mv/share-folder all resolve folders through the
@@ -517,6 +646,74 @@ def fake_capture(p, command):
         self.assertIn(self.PROTECTED_FOLDER_UID, params.root_folder.subfolders)
 
 
+class TestSailPointFolderProtectionCommandExecution(TestCase):
+    """Regression test: hide_from_folder_cache was imported and fed protected_folder_uids,
+    but never actually invoked as a context manager, so a protected record's folder kept
+    showing up in tree/ls even though the record itself was correctly hidden."""
+
+    FOLDER_UID = 'PROTECTED_FOLDER_UID'
+    RECORD_UID = 'PROTECTED_FOLDER_RECORD_UID'
+
+    def _params(self):
+        p = params_module.KeeperParams()
+        p.service_mode = False
+        p.record_cache = {
+            self.RECORD_UID: _record_cache_entry(self.RECORD_UID, 'Commander Service Mode SailPoint Config'),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        p.root_folder = RootFolderNode()
+        node = SharedFolderNode()
+        node.uid = self.FOLDER_UID
+        node.name = 'Commander Service Mode - SailPoint'
+        p.folder_cache = {self.FOLDER_UID: node}
+        p.root_folder.subfolders = [self.FOLDER_UID]
+        p.shared_folder_cache = {self.FOLDER_UID: {'name_unencrypted': node.name}}
+        p.subfolder_cache = {self.FOLDER_UID: {'type': 'shared_folder', 'shared_folder_uid': self.FOLDER_UID}}
+        p.subfolder_record_cache = {self.FOLDER_UID: {self.RECORD_UID}}
+        return p
+
+    def test_folder_hidden_during_actual_dispatch(self):
+        params = self._params()
+        seen = {}
+
+        def fake_capture(p, command):
+            seen['folder_keys'] = set(p.folder_cache.keys())
+            seen['subfolders'] = list(p.root_folder.subfolders)
+            return 'ok', 'ok', ''
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.object(CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture):
+            CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertNotIn(self.FOLDER_UID, seen['folder_keys'])
+        self.assertNotIn(self.FOLDER_UID, seen['subfolders'])
+        self.assertIn(self.FOLDER_UID, params.folder_cache)
+        self.assertIn(self.FOLDER_UID, params.root_folder.subfolders)
+
+    def test_folder_hidden_during_sailpoint_handle_command_too(self):
+        """Unlike the record, the folder has no reason to be visible to handle_command,
+        so it must stay hidden for the whole request, not just the final dispatch."""
+        params = self._params()
+        seen = {}
+
+        def fake_handle_command(p, command):
+            seen['folder_keys'] = set(p.folder_cache.keys())
+            return command, None
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command',
+            side_effect=fake_handle_command,
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
+        ):
+            CommandExecutor.execute(f'get {NORMAL_UID}')
+
+        self.assertNotIn(self.FOLDER_UID, seen['folder_keys'])
+
+
 class TestReservedAttachmentCommandExecution(TestCase):
     """A record with an arbitrary, non-default title must still be blocked if it carries
     one of Commander's own reserved config-file attachments (config.json/service_config.json)."""
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index 9fe2a6940..57b8b9fae 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -45,6 +45,10 @@ def test_contains_expected_titles_lowercased(self):
         self.assertIn('commander service mode docker config', titles)
         self.assertIn('commander service mode', titles)
 
+    def test_contains_sailpoint_title(self):
+        titles = get_protected_record_title_set()
+        self.assertIn('commander service mode sailpoint config', titles)
+
     def test_contains_terraform_slack_teams_gchat_titles(self):
         titles = get_protected_record_title_set()
         self.assertIn('commander service mode terraform config', titles)
@@ -108,6 +112,25 @@ def test_no_docker_env_var_falls_back_to_title_only(self):
             result = get_protected_record_uids(p)
         self.assertEqual(set(result.keys()), {'UID_CONFIG'})
 
+    def test_sailpoint_record_protected_by_uid_even_with_custom_title(self):
+        """--record-name can give the SailPoint config record a custom title; SAILPOINT_RECORD must still identify it."""
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'SAILPOINT_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom SailPoint Title'})
+            result = get_protected_record_uids(p)
+        self.assertIn(uid, result)
+
+    def test_sailpoint_env_uid_present_even_without_params(self):
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'SAILPOINT_RECORD': uid}):
+            self.assertIn(uid, get_protected_record_uids(None))
+
+    def test_sailpoint_default_title_protected_without_env_var(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            p = _params_with_records({'UID_SAILPOINT': 'Commander Service Mode SailPoint Config'})
+            result = get_protected_record_uids(p)
+        self.assertEqual(set(result.keys()), {'UID_SAILPOINT'})
+
     def test_malformed_env_uid_falls_back_to_title_matching(self):
         """A misconfigured pinning env var (not a real record UID) must not become a phantom protected token."""
         with mock.patch.dict(os.environ, {'TERRAFORM_RECORD': '1'}):
@@ -248,6 +271,24 @@ def test_nested_share_caches_are_guarded_too(self):
         self.assertIn('PROTECTED', p.nested_share_record_data)
         self.assertIs(type(p.nested_share_records), dict)
 
+    def test_nested_call_still_hides_its_own_uid_set(self):
+        """Regression test: the outer call replaces record_cache with a _GuardedRecordCache
+        (a UserDict, not a dict), so a naive isinstance(source, dict) check in the inner call
+        would skip wrapping entirely and silently no-op instead of hiding OUTER_ONLY too."""
+        p = _params_with_records({'OUTER_ONLY': 'x', 'BOTH': 'y', 'NORMAL': 'Other'})
+        with hide_from_record_cache(p, {'BOTH': 'y'}):
+            self.assertIn('OUTER_ONLY', p.record_cache)
+            with hide_from_record_cache(p, {'OUTER_ONLY': 'x', 'BOTH': 'y'}):
+                self.assertNotIn('OUTER_ONLY', p.record_cache)
+                self.assertNotIn('BOTH', p.record_cache)
+                self.assertIn('NORMAL', p.record_cache)
+            self.assertIn('OUTER_ONLY', p.record_cache)
+            self.assertNotIn('BOTH', p.record_cache)
+
+        self.assertIs(type(p.record_cache), dict)
+        self.assertIn('OUTER_ONLY', p.record_cache)
+        self.assertIn('BOTH', p.record_cache)
+
     def test_subfolder_record_cache_uid_is_stripped_for_the_duration(self):
         """_build_folder_json leaks the bare UID from a folder's set even when the record fails to load."""
         p = _params_with_records({'NORMAL': 'Other'})
@@ -428,6 +469,27 @@ def test_restores_even_if_block_raises(self):
             with hide_from_folder_cache(p, {'FOLDER1'}):
                 raise ValueError('boom')
         self.assertIn('FOLDER1', p.folder_cache)
+
+    def test_nested_call_still_hides_its_own_uid_set(self):
+        """Same nesting bug as record_cache's -- the outer call's _GuardedRecordCache must not
+        make the inner call skip wrapping and silently no-op for a folder only it hides."""
+        p = _params_with_folder(folder_uid='FOLDER1')
+        node2 = _folder_node('FOLDER2', 'Commander Service Mode - Terraform')
+        p.folder_cache['FOLDER2'] = node2
+        p.root_folder.subfolders.append('FOLDER2')
+        p.shared_folder_cache['FOLDER2'] = {'name_unencrypted': node2.name}
+        p.subfolder_cache['FOLDER2'] = {'type': 'shared_folder', 'shared_folder_uid': 'FOLDER2'}
+
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            self.assertIn('FOLDER2', p.folder_cache)
+            with hide_from_folder_cache(p, {'FOLDER1', 'FOLDER2'}):
+                self.assertNotIn('FOLDER1', p.folder_cache)
+                self.assertNotIn('FOLDER2', p.folder_cache)
+            self.assertIn('FOLDER2', p.folder_cache)
+            self.assertNotIn('FOLDER1', p.folder_cache)
+
+        self.assertIn('FOLDER1', p.folder_cache)
+        self.assertIn('FOLDER2', p.folder_cache)
         self.assertIn('FOLDER1', p.root_folder.subfolders)
 
     def test_reintroduction_during_block_is_blocked(self):
diff --git a/unit-tests/service/test_sailpoint_pending.py b/unit-tests/service/test_sailpoint_pending.py
index a1ad7fed0..d70d49da2 100644
--- a/unit-tests/service/test_sailpoint_pending.py
+++ b/unit-tests/service/test_sailpoint_pending.py
@@ -94,7 +94,7 @@ def test_parse_identity_mutation(self):
             'enterprise-user user@co.com --add-role Admin'
         )
         self.assertIsNotNone(parsed)
-        self.assertEqual(parsed.emails, ['user@co.com'])
+        self.assertEqual(parsed.identifiers, ['user@co.com'])
         self.assertTrue(parsed.has_role_change)
         self.assertFalse(parsed.has_team_change)
         self.assertFalse(parsed.has_node_change)
@@ -115,6 +115,22 @@ def test_parse_identity_mutation_ignores_non_mutating(self):
             )
         )
 
+    def test_parse_identity_mutation_accepts_numeric_id(self):
+        # enterprise-user accepts a numeric enterprise_user_id in place of an
+        # email; the SCIM guard must see it too, not silently no-op.
+        parsed = SailPointCommandParser.parse_identity_mutation(
+            'enterprise-user 1469016254185479 --add-role Admin'
+        )
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed.identifiers, ['1469016254185479'])
+        self.assertTrue(parsed.has_role_change)
+
+    def test_parse_identity_mutation_accepts_at_all(self):
+        parsed = SailPointCommandParser.parse_identity_mutation('eu @all --add-role Admin')
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed.identifiers, ['@all'])
+        self.assertTrue(parsed.has_role_change)
+
     def test_parse_share_record(self):
         parsed = SailPointCommandParser.parse_share('share-record -e user@co.com -w RECORD_UID')
         self.assertIsNotNone(parsed)
@@ -391,6 +407,70 @@ def _load(_params):
             self.assertIsNotNone(err)
             query.assert_called()
 
+    def test_find_user_by_numeric_id(self):
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'user@co.com', 'enterprise_user_id': 555, 'node_id': 10}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        user = SailPointScimGuard.find_user(params, '555')
+        self.assertIsNotNone(user)
+        self.assertEqual(user['username'], 'user@co.com')
+
+    def test_find_user_by_alias(self):
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'primary@co.com', 'enterprise_user_id': 555, 'node_id': 10}],
+            'user_aliases': [{'username': 'alias@co.com', 'enterprise_user_id': 555}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        user = SailPointScimGuard.find_user(params, 'alias@co.com')
+        self.assertIsNotNone(user)
+        self.assertEqual(user['username'], 'primary@co.com')
+
+    def test_identity_change_error_blocks_numeric_id_for_scim_user(self):
+        # The exact bypass this test guards against: the same user refused by
+        # email must also be refused when named by their enterprise user id.
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'user@co.com', 'enterprise_user_id': 555, 'node_id': 10}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        by_email = SailPointScimGuard.identity_change_error(params, 'user@co.com')
+        by_id = SailPointScimGuard.identity_change_error(params, '555')
+        self.assertIsNotNone(by_email)
+        self.assertIsNotNone(by_id)
+
+    def test_identity_change_error_at_all_blocks_when_any_scim_user_exists(self):
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [
+                {'username': 'plain@co.com', 'enterprise_user_id': 1, 'node_id': 1},
+                {'username': 'managed@co.com', 'enterprise_user_id': 2, 'node_id': 10},
+            ],
+            'nodes': [
+                {'node_id': 1, 'parent_id': None},
+                {'node_id': 10, 'parent_id': None, 'scim_id': 1},
+            ],
+            'scims': [],
+        }
+        err = SailPointScimGuard.identity_change_error(params, '@all')
+        self.assertIsNotNone(err)
+        self.assertIn('@all', err)
+
+    def test_identity_change_error_at_all_allows_when_no_scim_user(self):
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'plain@co.com', 'enterprise_user_id': 1, 'node_id': 1}],
+            'nodes': [{'node_id': 1, 'parent_id': None}],
+            'scims': [],
+        }
+        err = SailPointScimGuard.identity_change_error(params, '@all')
+        self.assertIsNone(err)
+
 
 class SailPointApplierTest(unittest.TestCase):
     def test_apply_nsf_folder_and_record_commands(self):
@@ -493,6 +573,72 @@ def test_default_record_name_and_env_key(self):
         self.assertEqual(cmd.get_default_folder_name(), 'Commander Service Mode - SailPoint')
 
 
+class SailPointMarkerFailureModeTest(unittest.TestCase):
+    """A marker read that raises must fail closed for per-request paths (deny/propagate);
+    startup paths may still safely degrade to 'not enabled'."""
+
+    def test_handle_command_propagates_marker_check_failure(self):
+        from keepercommander.service.commands.integrations.sailpoint.service import (
+            SailPointService,
+        )
+        from keepercommander.params import KeeperParams
+
+        params = KeeperParams()
+        with mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch.object(
+            SailPointService, 'record_has_marker', side_effect=RuntimeError('decrypt failed')
+        ):
+            with self.assertRaises(RuntimeError):
+                SailPointService.handle_command(params, 'get SOME_OTHER_RECORD_UID')
+
+    def test_after_command_propagates_marker_check_failure(self):
+        from keepercommander.service.commands.integrations.sailpoint.service import (
+            SailPointService,
+        )
+        from keepercommander.params import KeeperParams
+
+        params = KeeperParams()
+        with mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch.object(
+            SailPointService, 'record_has_marker', side_effect=RuntimeError('decrypt failed')
+        ):
+            with self.assertRaises(RuntimeError):
+                SailPointService.after_command(params, 'get SOME_OTHER_RECORD_UID', success=True)
+
+    def test_maybe_enable_degrades_to_not_enabled_on_marker_check_failure(self):
+        """Startup path -- nothing has executed yet, so degrading to 'not enabled' is safe."""
+        from keepercommander.service.commands.integrations.sailpoint.service import (
+            SailPointService,
+        )
+        from keepercommander.params import KeeperParams
+
+        params = KeeperParams()
+        args = mock.Mock(commands=None)
+        with mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch.object(
+            SailPointService, 'record_has_marker', side_effect=RuntimeError('decrypt failed')
+        ):
+            SailPointService.maybe_enable(params, args)
+        self.assertFalse(hasattr(params, SailPointService.PARAMS_ATTR))
+
+    def test_start_background_services_degrades_to_not_started_on_marker_check_failure(self):
+        from keepercommander.service.commands.integrations.sailpoint.service import (
+            SailPointService,
+        )
+        from keepercommander.service.commands.integrations.sailpoint.poller import (
+            SailPointEntitlementPoller,
+        )
+        from keepercommander.params import KeeperParams
+
+        params = KeeperParams()
+        with mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch(
+            'keepercommander.service.core.globals.get_current_params', return_value=params
+        ), mock.patch.object(
+            SailPointService, 'record_has_marker', side_effect=RuntimeError('decrypt failed')
+        ), mock.patch.object(
+            SailPointEntitlementPoller, 'start'
+        ) as mock_start:
+            SailPointService.start_background_services()
+        mock_start.assert_not_called()
+
+
 class SailPointShareTargetValidationTest(unittest.TestCase):
     def _params(self):
         params = mock.Mock()
@@ -868,6 +1014,130 @@ def test_before_command_injects_transfer_target(self):
         self.assertEqual(short[1], 403)
         self.assertIn('--delete', short[0]['error'])
 
+    def test_before_command_blocks_identity_mutation_by_numeric_id_for_scim_user(self):
+        from keepercommander.service.commands.integrations.sailpoint.command_hook import (
+            SailPointCommandHook,
+        )
+        from keepercommander.service.commands.integrations.sailpoint.config_fields import (
+            SailPointCapabilities,
+        )
+
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'victim@co.com', 'enterprise_user_id': 555, 'node_id': 10}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        caps = SailPointCapabilities(allow_roles=True, allow_teams=True)
+        hook = SailPointCommandHook('cfg-uid')
+
+        with mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.command_hook.read_capabilities',
+            return_value=caps,
+        ):
+            _, blocked_by_email = hook.before_command(
+                params, 'enterprise-user victim@co.com --add-role Admin'
+            )
+            _, blocked_by_id = hook.before_command(
+                params, 'enterprise-user 555 --add-role Admin'
+            )
+
+        self.assertIsNotNone(blocked_by_email)
+        self.assertEqual(blocked_by_email[1], 403)
+        self.assertIsNotNone(blocked_by_id)
+        self.assertEqual(blocked_by_id[1], 403)
+
+    def test_before_command_blocks_at_all_add_role_when_scim_user_exists(self):
+        from keepercommander.service.commands.integrations.sailpoint.command_hook import (
+            SailPointCommandHook,
+        )
+        from keepercommander.service.commands.integrations.sailpoint.config_fields import (
+            SailPointCapabilities,
+        )
+
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [
+                {'username': 'plain@co.com', 'enterprise_user_id': 1, 'node_id': 1},
+                {'username': 'managed@co.com', 'enterprise_user_id': 2, 'node_id': 10},
+            ],
+            'nodes': [
+                {'node_id': 1, 'parent_id': None},
+                {'node_id': 10, 'parent_id': None, 'scim_id': 1},
+            ],
+            'scims': [],
+        }
+        caps = SailPointCapabilities(allow_roles=True, allow_teams=True)
+        hook = SailPointCommandHook('cfg-uid')
+
+        with mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.command_hook.read_capabilities',
+            return_value=caps,
+        ):
+            _, short = hook.before_command(params, 'eu @all --add-role Admin')
+
+        self.assertIsNotNone(short)
+        self.assertEqual(short[1], 403)
+
+    def test_before_command_blocks_numeric_id_across_mutation_categories(self):
+        """Same numeric-id bypass, exercised for remove-role/add-team/remove-team/node."""
+        from keepercommander.service.commands.integrations.sailpoint.command_hook import (
+            SailPointCommandHook,
+        )
+        from keepercommander.service.commands.integrations.sailpoint.config_fields import (
+            SailPointCapabilities,
+        )
+
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'victim@co.com', 'enterprise_user_id': 555, 'node_id': 10}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        caps = SailPointCapabilities(allow_roles=True, allow_teams=True)
+        hook = SailPointCommandHook('cfg-uid')
+
+        for cmd in (
+            'enterprise-user 555 --remove-role Admin',
+            'enterprise-user 555 --add-team AWS',
+            'enterprise-user 555 --remove-team AWS',
+            'enterprise-user 555 --node OtherNode',
+        ):
+            with mock.patch(
+                'keepercommander.service.commands.integrations.sailpoint.command_hook.read_capabilities',
+                return_value=caps,
+            ):
+                _, short = hook.before_command(params, cmd)
+            self.assertIsNotNone(short, cmd)
+            self.assertEqual(short[1], 403, cmd)
+
+    def test_before_command_blocks_at_all_for_non_role_actions_too(self):
+        """Guard blocks '@all' even for actions Commander itself would silently no-op."""
+        from keepercommander.service.commands.integrations.sailpoint.command_hook import (
+            SailPointCommandHook,
+        )
+        from keepercommander.service.commands.integrations.sailpoint.config_fields import (
+            SailPointCapabilities,
+        )
+
+        params = mock.Mock()
+        params.enterprise = {
+            'users': [{'username': 'managed@co.com', 'enterprise_user_id': 2, 'node_id': 10}],
+            'nodes': [{'node_id': 10, 'parent_id': None, 'scim_id': 1}],
+            'scims': [],
+        }
+        caps = SailPointCapabilities(allow_roles=True, allow_teams=True)
+        hook = SailPointCommandHook('cfg-uid')
+
+        with mock.patch(
+            'keepercommander.service.commands.integrations.sailpoint.command_hook.read_capabilities',
+            return_value=caps,
+        ):
+            _, short = hook.before_command(params, 'eu @all --add-team AWS')
+
+        self.assertIsNotNone(short)
+        self.assertEqual(short[1], 403)
+
     def test_before_command_rejects_banned_commands_at_runtime(self):
         """Banned commands rejected even if present in stored config (in-place upgrade scenario)."""
         from keepercommander.service.commands.integrations.sailpoint.command_hook import (
diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py
index e1a94337a..7d95a787d 100644
--- a/unit-tests/service/test_verified_command.py
+++ b/unit-tests/service/test_verified_command.py
@@ -347,6 +347,27 @@ def test_double_dash_blocked_everywhere(self):
         self.assertIsNone(check(_tokens('pam tunnel edit uid')))
         self.assertIsNone(check(_tokens('get RECORD_UID')))
 
+    def test_env_var_expansion_blocked_everywhere(self):
+        """expand_cmd_args substitutes ${VARNAME} after this check runs, so it must be banned outright."""
+        check = Verifycommand.validate_service_mode_restrictions
+        ban = "'${VARNAME}'"
+        for cmd in (
+            "record-add --folder FOLDER_UID '${PAYLOAD}'",
+            "record-add --field=${PAYLOAD}",
+            "get ${last_record_uid}",
+            "share-folder ${LAST_FOLDER_UID} -e a@b.com",
+        ):
+            with self.subTest(cmd=cmd):
+                err = check(_tokens(cmd))
+                self.assertIsNotNone(err)
+                self.assertIn(ban, err)
+
+        # Unrelated commands without ${...} syntax remain unaffected.
+        self.assertIsNone(check(_tokens('get RECORD_UID')))
+        self.assertIsNone(check(_tokens('record-add --title=x login=user')))
+        # A single literal '$' or unmatched braces are not the ${VARNAME} pattern.
+        self.assertIsNone(check(_tokens('record-add --field=$5.00')))
+
     def test_temp_path_leaf_symlink_is_not_containment(self):
         """A symlink planted at the temp-dir leaf must not escape containment."""
         request_temp_dir = tempfile.mkdtemp()