diff --git a/datadog_sync/commands/shared/options.py b/datadog_sync/commands/shared/options.py
index df944fcc..a519b6ec 100644
--- a/datadog_sync/commands/shared/options.py
+++ b/datadog_sync/commands/shared/options.py
@@ -499,6 +499,21 @@ def click_config_file_provider(ctx: Context, opts: CustomOptionClass, value: Non
         help="Skip resource if resource connection fails.",
         cls=CustomOptionClass,
     ),
+    option(
+        "--drop-unresolvable-principals",
+        required=False,
+        is_flag=True,
+        default=False,
+        show_default=True,
+        help="For restriction policies and restricted_roles lists: drop principal/role "
+        "references that are absent from BOTH destination and source state (permanently "
+        "gone, e.g. deleted before the org's first import) instead of skipping the whole "
+        "resource. If dropping empties a binding/list that had entries at the source, the "
+        "resource connection fails normally. --skip-failed-resource-connections may suppress "
+        "that failure and continue syncing; an ERROR log and risk metric explicitly warn that "
+        "the destination resource may be unrestricted. Off by default.",
+        cls=CustomOptionClass,
+    ),
     option(
         "--cleanup",
         default="False",
diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py
index 88867594..8e6ed486 100644
--- a/datadog_sync/utils/base_resource.py
+++ b/datadog_sync/utils/base_resource.py
@@ -141,6 +141,13 @@ def build_excluded_attributes(self) -> None:
                 self.excluded_attributes[i] = "root" + "".join(["['{}']".format(v) for v in attr.split(".")])
 
 
+@dataclass(frozen=True)
+class ResourceConnectionResult:
+    """Outcome metadata from resolving a resource's cross-resource connections."""
+
+    empty_binding_escalation: bool = False
+
+
 class BaseResource(abc.ABC):
     resource_type: str
     resource_config: ResourceConfig
@@ -437,6 +444,191 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
         return failed_connections
 
+    def _resolve_or_drop(self, plain_id: str, resource_to_connect: str) -> Tuple[Optional[str], bool]:
+        """Resolve plain_id (no "type:" prefix) against destination state.
+
+        Returns (resolved_destination_id, is_permanently_stale):
+          - Present in state.destination[resource_to_connect]: returns
+            (destination_id, False) -- success path, caller keeps/remaps this entry.
+          - Absent from destination: calls ensure_resource_loaded() (for
+            --minimize-reads correctness), then rechecks destination because the lazy load
+            may have populated its mapping, and only then checks source.
+              - Present in source ("not yet synced"): returns (None, False) -- caller
+                must treat this as today's hard-fail (add plain_id to failed_connections);
+                NOT a drop. This is the legitimate "retry on a later sync" case.
+              - Absent from source (permanently gone -- deleted before this org's
+                first-ever import, or never existed): returns (None, True) -- caller
+                drops this entry ONLY if self.config.drop_unresolvable_principals is True;
+                if the flag is False, caller must treat this identically to the "not yet
+                synced" case (today's unchanged hard-fail).
+
+        Callers own: parsing "type:id" composites to plain_id before calling, reassembling
+        "type:id" on success, deciding what "the list" means for their shape
+        (binding.principals vs. flat restricted_roles), and the "list is now empty"
+        access-elevation check.
+
+        state.destination/state.source are defaultdict(dict), so indexing an unknown
+        resource_to_connect returns {} rather than raising -- plain `in` membership checks
+        are safe and no KeyError guard is needed. An exception raised by
+        ensure_resource_loaded (e.g. a storage error) propagates uncaught, matching every
+        other ensure_resource_loaded call site.
+        """
+        destination = self.config.state.destination[resource_to_connect]
+        if plain_id in destination:
+            return destination[plain_id]["id"], False
+
+        self.config.state.ensure_resource_loaded(resource_to_connect, plain_id)
+        destination = self.config.state.destination[resource_to_connect]
+        if plain_id in destination:
+            return destination[plain_id]["id"], False
+
+        if plain_id in self.config.state.source[resource_to_connect]:
+            return None, False
+        return None, True
+
+    # Composite "type:id" principal type prefixes -> the resource_to_connect they map to.
+    _PRINCIPAL_TYPE_MAP: ClassVar[Dict[str, str]] = {"user": "users", "role": "roles", "team": "teams"}
+
+    def _filter_stale_binding_principals(
+        self, _id: str, bindings: Optional[List[Dict]]
+    ) -> Tuple[Dict[str, List[str]], bool]:
+        """Resolve/drop/hard-fail composite "type:id" principals across restriction-policy
+        bindings (shared by restriction_policies / monitors / synthetics_tests).
+
+        For each principal in each binding: resolve against destination (remap in place),
+        else consult source. Source-present ("not yet synced") or flag-off -> hard-fail
+        (add to failed_connections). Source-absent ("permanently gone") AND
+        --drop-unresolvable-principals -> drop it, WARN, and count it.
+
+        Rebuilds each binding's "principals" as a NEW list (never del/pop/index-assign
+        during iteration) to avoid the enumerate/index-shift skip bug. Returns
+        (failed_connections_dict, empty_binding_risk) where empty_binding_risk is True if
+        any binding whose source list was non-empty became empty after dropping.
+        """
+        failed_connections_dict: Dict[str, List[str]] = defaultdict(list)
+        empty_binding_risk = False
+        for binding in bindings or []:
+            principals = binding.get("principals")
+            if not principals:
+                continue
+            had_principals = len(principals) > 0
+            kept: List[str] = []
+            for policy_id in principals:
+                parts = policy_id.split(":", 1)
+                if len(parts) != 2:
+                    kept.append(policy_id)
+                    continue
+                _type, plain_id = parts
+                resource_to_connect = self._PRINCIPAL_TYPE_MAP.get(_type)
+                if resource_to_connect is None:
+                    # org: (already remapped by pre_resource_action_hook) or any other
+                    # non-user/role/team principal -> pass through untouched.
+                    kept.append(policy_id)
+                    continue
+                resolved, stale = self._resolve_or_drop(plain_id, resource_to_connect)
+                if resolved is not None:
+                    kept.append(f"{_type}:{resolved}")
+                elif stale and self.config.drop_unresolvable_principals:
+                    self.config.logger.warning(
+                        f"dropping stale principal '{policy_id}': absent from source and "
+                        "destination (likely deleted before the org's first sync)",
+                        resource_type=self.resource_type,
+                        _id=_id,
+                    )
+                    if self.config.counter is not None:
+                        self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id)
+                else:
+                    # Source-present ("not yet synced", retry later) or flag off:
+                    # unchanged hard-fail semantics -- keep the original id and record it.
+                    kept.append(policy_id)
+                    failed_connections_dict[resource_to_connect].append(plain_id)
+            binding["principals"] = kept
+            if had_principals and not kept:
+                empty_binding_risk = True
+        return failed_connections_dict, empty_binding_risk
+
+    def _filter_stale_flat_roles(self, _id: str, container: Optional[Dict], key: str) -> Tuple[List[str], bool]:
+        """Resolve/drop/hard-fail a flat list of role ids at container[key]
+        (restricted_roles / options.restricted_roles / metadata.restricted_roles).
+
+        Same three-way logic as _filter_stale_binding_principals but for plain role ids
+        (no "type:" prefix). Rebuilds the list as a NEW list. Returns
+        (failed_role_ids, empty_list_risk) where empty_list_risk is True if a non-empty
+        source list became empty after dropping.
+        """
+        failed: List[str] = []
+        if not container:
+            return failed, False
+        roles = container.get(key)
+        if not roles:
+            return failed, False
+        had_roles = len(roles) > 0
+        kept: List[str] = []
+        for role in roles:
+            plain_id = str(role)
+            resolved, stale = self._resolve_or_drop(plain_id, "roles")
+            if resolved is not None:
+                kept.append(type(role)(resolved))
+            elif stale and self.config.drop_unresolvable_principals:
+                self.config.logger.warning(
+                    f"dropping stale role '{plain_id}' from {key}: absent from source and "
+                    "destination (likely deleted before the org's first sync)",
+                    resource_type=self.resource_type,
+                    _id=_id,
+                )
+                if self.config.counter is not None:
+                    self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id)
+            else:
+                kept.append(role)
+                failed.append(plain_id)
+        container[key] = kept
+        return failed, (had_roles and not kept)
+
+    def _raise_connection_error_if_any(
+        self, _id: str, failed_connections_dict: Dict[str, List[str]], empty_binding_risk: bool = False
+    ) -> bool:
+        """Shared terminal step for the drop-aware connect_resources overrides.
+
+        Mirrors the base connect_resources raise/skip behavior: raise unless
+        --skip-failed-resource-connections is set. empty_binding_risk is threaded onto the
+        exception so _apply_resource_cb can tag skipped-resource metrics. When
+        --skip-failed-resource-connections suppresses the exception, returns True for an
+        empty-binding risk so the apply path can tag the successful action and include it in
+        the end-of-run escalation summary. Otherwise returns False.
+        """
+        if not failed_connections_dict and not empty_binding_risk:
+            return False
+        e = ResourceConnectionError(
+            failed_connections_dict=failed_connections_dict, empty_binding_risk=empty_binding_risk
+        )
+        if empty_binding_risk:
+            if self.config.skip_failed_resource_connections:
+                self.config.logger.error(
+                    "access-elevation risk: a binding/list whose source had principals became "
+                    "empty after dropping unresolvable references; "
+                    "--skip-failed-resource-connections is enabled, continuing sync. "
+                    "DESTINATION RESOURCE MAY BE UNRESTRICTED",
+                    resource_type=self.resource_type,
+                    _id=_id,
+                )
+            else:
+                self.config.logger.error(
+                    "access-elevation risk: a binding/list whose source had principals became "
+                    "empty after dropping unresolvable references; refusing to sync",
+                    resource_type=self.resource_type,
+                    _id=_id,
+                )
+        if not self.config.skip_failed_resource_connections:
+            self.config.logger.info(
+                f"skipping resource: {str(e)}",
+                _id=_id,
+                resource_type=self.resource_type,
+            )
+            raise e
+        else:
+            self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type)
+            return empty_binding_risk
+
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         """Extract dependency IDs referenced at r_obj[key] for resource_to_connect.
 
@@ -451,9 +643,9 @@ def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) ->
             return [str(v) for v in r_obj[key]]
         return [str(r_obj[key])]
 
-    def connect_resources(self, _id: str, resource: Dict) -> None:
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
         if not self.resource_config.resource_connections:
-            return
+            return ResourceConnectionResult()
 
         failed_connections_dict = defaultdict(list)
         for resource_to_connect, v in self.resource_config.resource_connections.items():
@@ -475,6 +667,8 @@ def connect_resources(self, _id: str, resource: Dict) -> None:
             else:
                 self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type)
 
+        return ResourceConnectionResult()
+
     def filter(self, resource: Dict) -> bool:
         if not self.config.filters or self.resource_type not in self.config.filters:
             return True
diff --git a/datadog_sync/utils/configuration.py b/datadog_sync/utils/configuration.py
index f91e9591..75589760 100644
--- a/datadog_sync/utils/configuration.py
+++ b/datadog_sync/utils/configuration.py
@@ -8,7 +8,7 @@
 import logging
 import sys
 import time
-from typing import Any, Optional, Union, Dict, List
+from typing import Any, Optional, TYPE_CHECKING, Union, Dict, List
 
 import click
 
@@ -47,6 +47,9 @@
 from datadog_sync.utils.state import State
 from datadog_sync.utils.storage.storage_types import StorageType
 
+if TYPE_CHECKING:
+    from datadog_sync.utils.workers import Counter
+
 
 @dataclass
 class Configuration(object):
@@ -77,6 +80,16 @@ class Configuration(object):
     allow_self_lockout: bool
     datadog_host_override: Optional[str] = None
     emit_json: bool = False
+    # Opt-in: drop principal/role references that are absent from BOTH destination and
+    # source state (permanently gone -- e.g. deleted before this org's first-ever import)
+    # instead of hard-failing the whole resource. When False (default) behavior is
+    # byte-for-byte unchanged. The shared BaseResource drop-aware helpers consume this
+    # flag; model-specific connection wiring is intentionally separate.
+    drop_unresolvable_principals: bool = False
+    # Set at runtime by Workers.__init__ so resource models (which only hold self.config)
+    # can record dropped stale principals into the apply-run Counter from inside
+    # connect_id/connect_resources. None until a Workers is constructed for the command.
+    counter: Optional["Counter"] = None
     # Opt-in refresh of state.destination from storage before apply_resources
     # dispatches workers. Motivating case: an external orchestrator runs
     # sync-cli once per resource type as separate processes reading a shared
@@ -407,6 +420,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration:
     # Additional settings
     force_missing_dependencies = kwargs.get("force_missing_dependencies") or False
     skip_failed_resource_connections = kwargs.get("skip_failed_resource_connections")
+    drop_unresolvable_principals = kwargs.get("drop_unresolvable_principals") or False
     refresh_destination_state_before_apply = kwargs.get("refresh_destination_state_before_apply") or False
     max_workers = kwargs.get("max_workers")
     max_workers_per_type_raw = kwargs.get("max_workers_per_type")
@@ -660,6 +674,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration:
         filter_operator=filter_operator,
         force_missing_dependencies=force_missing_dependencies,
         skip_failed_resource_connections=skip_failed_resource_connections,
+        drop_unresolvable_principals=drop_unresolvable_principals,
         refresh_destination_state_before_apply=refresh_destination_state_before_apply,
         max_workers=max_workers,
         max_workers_per_type=max_workers_per_type,
diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py
index 800d2dbb..aa550b78 100644
--- a/datadog_sync/utils/resource_utils.py
+++ b/datadog_sync/utils/resource_utils.py
@@ -78,8 +78,13 @@ def __init__(self, _id: str, _type: str):
 
 
 class ResourceConnectionError(Exception):
-    def __init__(self, failed_connections_dict):
+    def __init__(self, failed_connections_dict, empty_binding_risk: bool = False):
         super(ResourceConnectionError, self).__init__(f"Failed to connect resource. {dict(failed_connections_dict)}")
+        # empty_binding_risk flags the access-elevation case: a restriction-policy
+        # binding (or a restricted_roles list) whose source-side list was non-empty
+        # but became empty after dropping unresolvable principals. Read by
+        # ResourcesHandler._apply_resource_cb to add a distinguishing metric tag.
+        self.empty_binding_risk = empty_binding_risk
 
 
 class CustomClientHTTPError(Exception):
diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py
index 752b7392..ad6bde84 100644
--- a/datadog_sync/utils/resources_handler.py
+++ b/datadog_sync/utils/resources_handler.py
@@ -78,6 +78,48 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None:
         ids = counter.skipped_missing_deps_by_type[rt]
         if ids:
             _chunked_emit(rt, "skipped for missing dependencies", ids)
+    # --drop-unresolvable-principals aggregate signals. Both dicts stay empty when the flag
+    # is off, so these lines never appear in the default path. getattr guards Counters
+    # constructed before these buckets existed.
+    for rt in sorted(getattr(counter, "stale_principals_dropped_by_type", {}).keys()):
+        ids = counter.stale_principals_dropped_by_type[rt]
+        if ids:
+            _chunked_emit(rt, "dropped stale principals", ids)
+    for rt in sorted(getattr(counter, "empty_binding_risk_by_type", {}).keys()):
+        ids = counter.empty_binding_risk_by_type[rt]
+        if ids:
+            total = len(ids)
+            for start in range(0, total, _SUMMARY_ID_CHUNK):
+                chunk = ids[start:start + _SUMMARY_ID_CHUNK]
+                end = min(start + _SUMMARY_ID_CHUNK, total)
+                logger.error(
+                    "sync summary: %s skipped %d resource(s) for empty-binding "
+                    "access-elevation risk [%d-%d of %d]: %s",
+                    rt,
+                    total,
+                    start + 1,
+                    end,
+                    total,
+                    ", ".join(chunk),
+                )
+    for rt in sorted(getattr(counter, "empty_binding_escalation_by_type", {}).keys()):
+        ids = counter.empty_binding_escalation_by_type[rt]
+        if ids:
+            total = len(ids)
+            for start in range(0, total, _SUMMARY_ID_CHUNK):
+                chunk = ids[start:start + _SUMMARY_ID_CHUNK]
+                end = min(start + _SUMMARY_ID_CHUNK, total)
+                logger.error(
+                    "sync summary: %s synced %d resource(s) after an empty-binding "
+                    "connection failure was suppressed [%d-%d of %d]: %s. "
+                    "DESTINATION RESOURCE MAY BE UNRESTRICTED",
+                    rt,
+                    total,
+                    start + 1,
+                    end,
+                    total,
+                    ", ".join(chunk),
+                )
 
 
 def _list_time_filter_passes(r_class, config, resource) -> bool:
@@ -451,7 +493,12 @@ async def _apply_resource_cb(self, q_item: List) -> None:
 
             # Run hooks
             await r_class._pre_resource_action_hook(_id, resource)
-            r_class.connect_resources(_id, resource)
+            connection_result = r_class.connect_resources(_id, resource)
+            empty_binding_escalation = connection_result.empty_binding_escalation
+
+            success_metric_tags = []
+            if empty_binding_escalation:
+                success_metric_tags.append("risk:empty_restriction_policy")
 
             prep_resource(r_class.resource_config, resource)
             if _id in self.config.state.destination[resource_type]:
@@ -463,16 +510,26 @@ async def _apply_resource_cb(self, q_item: List) -> None:
 
                 self.config.logger.debug(f"Running update for {resource_type} with {_id}")
                 await r_class._update_resource(_id, resource)
+                if empty_binding_escalation:
+                    self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id)
                 await r_class._send_action_metrics(
-                    Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:update"]
+                    Command.SYNC.value,
+                    _id,
+                    Status.SUCCESS.value,
+                    tags=["action_sub_type:update", *success_metric_tags],
                 )
                 self.config.logger.debug(f"Finished update for {resource_type} with {_id}")
                 self._emit(resource_type, _id, "sync", "success", "update")
             else:
                 self.config.logger.debug(f"Running create for {resource_type} with id: {_id}")
                 await r_class._create_resource(_id, resource)
+                if empty_binding_escalation:
+                    self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id)
                 await r_class._send_action_metrics(
-                    Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:create"]
+                    Command.SYNC.value,
+                    _id,
+                    Status.SUCCESS.value,
+                    tags=["action_sub_type:create", *success_metric_tags],
                 )
                 self.config.logger.debug(f"finished create for {resource_type} with id: {_id}")
                 self._emit(resource_type, _id, "sync", "success", "create")
@@ -502,9 +559,15 @@ async def _apply_resource_cb(self, q_item: List) -> None:
             )
             _reason, _fc = self._sanitize_reason(e)
             self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc)
-            await r_class._send_action_metrics(
-                Command.SYNC.value, _id, Status.SKIPPED.value, tags=["reason:connection_error"]
-            )
+            # Distinguish the access-elevation case (a restriction-policy binding /
+            # restricted_roles list that emptied out after dropping stale principals) with
+            # a dedicated metric tag and a separate counter bucket for the end-of-run
+            # summary. getattr guards ResourceConnectionError instances without the attr.
+            extra_tags = ["reason:connection_error"]
+            if getattr(e, "empty_binding_risk", False):
+                extra_tags.append("risk:empty_restriction_policy")
+                self.worker.counter.record_empty_binding_risk(resource_type=resource_type, _id=_id)
+            await r_class._send_action_metrics(Command.SYNC.value, _id, Status.SKIPPED.value, tags=extra_tags)
         except Exception as e:
             # Track the source id in the counter's failed-ids bucket so
             # apply_resources can emit a targeted per-type summary at the
diff --git a/datadog_sync/utils/workers.py b/datadog_sync/utils/workers.py
index f67dea66..35704899 100644
--- a/datadog_sync/utils/workers.py
+++ b/datadog_sync/utils/workers.py
@@ -23,6 +23,10 @@ def __init__(self, config: Configuration) -> None:
         self.workers: List[Task] = []
         self.work_queue: Queue = Queue()
         self.counter: Counter = Counter()
+        # Back-reference so resource models (which only hold self.config) can record
+        # dropped stale principals into the counter from inside connect_id/connect_resources.
+        # Last-writer-wins is fine: only one Workers drives a given command's apply pass.
+        config.counter = self.counter
         self.pbar: Optional[tqdm] = None
         self._running_workers_count: int = 0
         self._loop: AbstractEventLoop = get_event_loop()
@@ -119,6 +123,13 @@ class Counter:
     # actionable if the roles-sync run logged that X, Y, Z failed.
     failed_ids_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list))
     skipped_missing_deps_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list))
+    # Per-resource-type source IDs whose stale (source-absent) principal/role
+    # references were dropped from an access-control list, and resource IDs skipped
+    # because dropping left an empty binding/list (access-elevation risk). Populated
+    # only when --drop-unresolvable-principals is set. Surfaced by _emit_apply_summary.
+    stale_principals_dropped_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list))
+    empty_binding_risk_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list))
+    empty_binding_escalation_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list))
 
     def __str__(self):
         return (
@@ -129,6 +140,9 @@ def reset_counter(self) -> None:
         self.successes = self.failure = self.skipped = self.filtered = 0
         self.failed_ids_by_type = defaultdict(list)
         self.skipped_missing_deps_by_type = defaultdict(list)
+        self.stale_principals_dropped_by_type = defaultdict(list)
+        self.empty_binding_risk_by_type = defaultdict(list)
+        self.empty_binding_escalation_by_type = defaultdict(list)
 
     def increment_success(self) -> None:
         self.successes += 1
@@ -141,11 +155,32 @@ def increment_failure(self, resource_type: Optional[str] = None, _id: Optional[s
         if resource_type is not None and _id is not None:
             self.failed_ids_by_type[resource_type].append(str(_id))
 
-    def increment_skipped(self, resource_type: Optional[str] = None, _id: Optional[str] = None,
-                          missing_deps: bool = False) -> None:
+    def increment_skipped(
+        self, resource_type: Optional[str] = None, _id: Optional[str] = None, missing_deps: bool = False
+    ) -> None:
         self.skipped += 1
         if missing_deps and resource_type is not None and _id is not None:
             self.skipped_missing_deps_by_type[resource_type].append(str(_id))
 
     def increment_filtered(self) -> None:
         self.filtered += 1
+
+    # Two dedicated methods rather than the single-multi-purpose-method convention used by
+    # increment_skipped/increment_failure: the "dropped a stale principal but kept syncing"
+    # and "skipped for empty-binding access-elevation risk" cases are semantically distinct
+    # enough (one is a non-fatal drop, the other a hard-fail security signal) to warrant
+    # separate names. This deliberate deviation is called out here so it isn't flagged as
+    # an inconsistency.
+    def record_stale_principal_dropped(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None:
+        if resource_type is not None and _id is not None:
+            self.stale_principals_dropped_by_type[resource_type].append(str(_id))
+
+    def record_empty_binding_risk(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None:
+        if resource_type is not None and _id is not None:
+            self.empty_binding_risk_by_type[resource_type].append(str(_id))
+
+    def record_empty_binding_escalation(
+        self, resource_type: Optional[str] = None, _id: Optional[str] = None
+    ) -> None:
+        if resource_type is not None and _id is not None:
+            self.empty_binding_escalation_by_type[resource_type].append(str(_id))
diff --git a/tests/unit/test_apply_resource_connection_error.py b/tests/unit/test_apply_resource_connection_error.py
index 535c7b2d..2ecef0b2 100644
--- a/tests/unit/test_apply_resource_connection_error.py
+++ b/tests/unit/test_apply_resource_connection_error.py
@@ -14,6 +14,7 @@
 import asyncio
 from unittest.mock import AsyncMock, MagicMock
 
+from datadog_sync.utils.base_resource import ResourceConfig, ResourceConnectionResult
 from datadog_sync.utils.resource_utils import ResourceConnectionError
 from datadog_sync.utils.resources_handler import ResourcesHandler
 
@@ -68,3 +69,65 @@ def test_resource_connection_error_still_increments_skipped(mock_config):
     emit_args, emit_kwargs = handler._emit.call_args
     assert emit_args[:4] == ("dashboards", "dash-1", "sync", "skipped")
     assert "reason" in emit_kwargs
+
+
+def _metric_tags(r_class):
+    """Extract the tags passed to the (async) _send_action_metrics call."""
+    _args, kwargs = r_class._send_action_metrics.call_args
+    return kwargs.get("tags", [])
+
+
+def test_empty_binding_risk_adds_metric_tag_and_records(mock_config):
+    """empty_binding_risk=True must add the risk tag AND bump the dedicated counter."""
+    exc = ResourceConnectionError({"roles": ["role-gone"]}, empty_binding_risk=True)
+
+    handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:dash-1", exc)
+
+    r_class = mock_config.resources["restriction_policies"]
+    tags = _metric_tags(r_class)
+    assert "reason:connection_error" in tags
+    assert "risk:empty_restriction_policy" in tags
+    handler.worker.counter.record_empty_binding_risk.assert_called_once_with(
+        resource_type="restriction_policies", _id="dashboard:dash-1"
+    )
+
+
+def test_ordinary_connection_error_has_no_risk_tag(mock_config):
+    """Without empty_binding_risk the risk tag is absent and the counter is untouched."""
+    exc = ResourceConnectionError({"roles": ["role-not-yet-synced"]})
+
+    handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:xyz", exc)
+
+    r_class = mock_config.resources["restriction_policies"]
+    tags = _metric_tags(r_class)
+    assert "reason:connection_error" in tags
+    assert "risk:empty_restriction_policy" not in tags
+    handler.worker.counter.record_empty_binding_risk.assert_not_called()
+
+
+def test_successful_update_after_suppressed_empty_binding_risk_is_tagged_and_recorded(mock_config):
+    resource_type = "restriction_policies"
+    _id = "dashboard:dash-1"
+    r_class = MagicMock()
+    r_class.resource_config = ResourceConfig(base_path="", skip_resource_mapping=True)
+    r_class.connect_resources.return_value = ResourceConnectionResult(empty_binding_escalation=True)
+    r_class._pre_resource_action_hook = AsyncMock()
+    r_class._update_resource = AsyncMock()
+    r_class._send_action_metrics = AsyncMock()
+    mock_config.resources = {resource_type: r_class}
+    mock_config.state.source[resource_type][_id] = {"id": _id, "bindings": []}
+    mock_config.state.destination[resource_type][_id] = {"id": _id, "bindings": [{"relation": "editor"}]}
+
+    handler = ResourcesHandler(mock_config)
+    handler.worker = MagicMock()
+    handler.worker.counter = MagicMock()
+    handler.sorter = MagicMock()
+    handler._emit = MagicMock()
+
+    asyncio.run(handler._apply_resource_cb([resource_type, _id]))
+
+    assert "risk:empty_restriction_policy" in _metric_tags(r_class)
+    handler.worker.counter.record_empty_binding_escalation.assert_called_once_with(
+        resource_type=resource_type, _id=_id
+    )
+    handler.worker.counter.record_empty_binding_risk.assert_not_called()
diff --git a/tests/unit/test_apply_summary_handler.py b/tests/unit/test_apply_summary_handler.py
index 45ea3b7e..1290e58f 100644
--- a/tests/unit/test_apply_summary_handler.py
+++ b/tests/unit/test_apply_summary_handler.py
@@ -36,6 +36,19 @@ def _rendered_warnings(logger):
     return out
 
 
+def _rendered_errors(logger):
+    out = []
+    for call in logger.error.call_args_list:
+        args = call.args
+        if not args:
+            continue
+        try:
+            out.append(args[0] % args[1:])
+        except Exception:
+            out.append(str(args))
+    return out
+
+
 def test_summary_emits_failed_ids_by_type():
     counter = Counter()
     counter.increment_failure(resource_type="roles", _id="uuid-abc")
@@ -123,6 +136,55 @@ def test_summary_uses_join_not_repr_for_greppability():
     assert "sync summary" in joined
 
 
+def test_summary_emits_stale_principals_dropped_at_warning():
+    counter = Counter()
+    counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="role:role-uuid-1")
+
+    logger = _make_logger()
+    _emit_apply_summary(logger, counter)
+
+    joined = "\n".join(_rendered_warnings(logger))
+    assert "restriction_policies" in joined
+    assert "role:role-uuid-1" in joined
+    assert "dropped" in joined and "stale principal" in joined
+
+
+def test_summary_emits_empty_binding_risk_at_error():
+    counter = Counter()
+    counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1")
+
+    logger = _make_logger()
+    _emit_apply_summary(logger, counter)
+
+    joined_err = "\n".join(_rendered_errors(logger))
+    assert "restriction_policies" in joined_err
+    assert "dashboard:dash-1" in joined_err
+    assert "empty-binding" in joined_err and "access-elevation" in joined_err
+
+
+def test_summary_emits_applied_empty_binding_escalation_at_error():
+    counter = Counter()
+    counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1")
+
+    logger = _make_logger()
+    _emit_apply_summary(logger, counter)
+
+    joined_err = "\n".join(_rendered_errors(logger))
+    assert "restriction_policies" in joined_err
+    assert "dashboard:dash-1" in joined_err
+    assert "connection failure was suppressed" in joined_err
+    assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in joined_err
+
+
+def test_summary_silent_for_new_buckets_when_empty():
+    counter = Counter()
+    logger = _make_logger()
+    _emit_apply_summary(logger, counter)
+    # No stale/empty-binding activity -> neither WARNING nor ERROR lines emitted.
+    assert logger.warning.call_count == 0
+    assert logger.error.call_count == 0
+
+
 def test_chunk_constant_is_reasonable():
     assert 20 <= _SUMMARY_ID_CHUNK <= 200
 
diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py
new file mode 100644
index 00000000..e3bf0446
--- /dev/null
+++ b/tests/unit/test_base_resource.py
@@ -0,0 +1,160 @@
+# Unless explicitly stated otherwise all files in this repository are licensed
+# under the 3-clause BSD style license (see LICENSE).
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2019 Datadog, Inc.
+
+"""Unit tests for BaseResource._resolve_or_drop.
+
+_resolve_or_drop is the shared three-way resolver behind --drop-unresolvable-principals:
+it decides whether a plain dependency id resolves against destination state, is a
+"not yet synced" source-present miss, or is a permanently-gone (source-absent) stale
+reference. It reads only self.config, so we exercise it as an unbound method with a
+stub self rather than instantiating the abstract BaseResource.
+"""
+
+from collections import defaultdict
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from datadog_sync.utils.base_resource import BaseResource, ResourceConnectionResult
+
+
+def _make_config():
+    config = MagicMock()
+    state = MagicMock()
+    state.source = defaultdict(dict)
+    state.destination = defaultdict(dict)
+    state.ensure_resource_loaded = MagicMock()
+    config.state = state
+    return config
+
+
+def _call(config, plain_id, resource_to_connect):
+    stub = SimpleNamespace(config=config)
+    return BaseResource._resolve_or_drop(stub, plain_id, resource_to_connect)
+
+
+def test_connect_resources_returns_named_default_result_when_no_connections():
+    stub = SimpleNamespace(resource_config=SimpleNamespace(resource_connections=None))
+
+    result = BaseResource.connect_resources(stub, "resource-id", {})
+
+    assert result == ResourceConnectionResult()
+    assert result.empty_binding_escalation is False
+
+
+def test_resolve_or_drop_destination_present_returns_dest_id_not_stale():
+    config = _make_config()
+    config.state.destination["roles"]["src-role"] = {"id": "dst-role"}
+
+    resolved, stale = _call(config, "src-role", "roles")
+
+    assert resolved == "dst-role"
+    assert stale is False
+    # No need to consult source when destination already has it.
+    config.state.ensure_resource_loaded.assert_not_called()
+
+
+def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_both_states():
+    config = _make_config()
+
+    def load_role(resource_type, plain_id):
+        config.state.source[resource_type][plain_id] = {"id": plain_id}
+        config.state.destination[resource_type][plain_id] = {"id": "dst-role"}
+
+    config.state.ensure_resource_loaded.side_effect = load_role
+
+    resolved, stale = _call(config, "src-role", "roles")
+
+    assert resolved == "dst-role"
+    assert stale is False
+    config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role")
+
+
+def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_destination_only():
+    config = _make_config()
+
+    def load_role(resource_type, plain_id):
+        config.state.destination[resource_type][plain_id] = {"id": "dst-role"}
+
+    config.state.ensure_resource_loaded.side_effect = load_role
+
+    resolved, stale = _call(config, "src-role", "roles")
+
+    assert resolved == "dst-role"
+    assert stale is False
+    config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role")
+
+
+def test_resolve_or_drop_source_present_not_synced_returns_none_not_stale():
+    config = _make_config()
+    config.state.source["roles"]["src-role"] = {"id": "src-role"}
+
+    resolved, stale = _call(config, "src-role", "roles")
+
+    assert resolved is None
+    assert stale is False  # legitimate "not yet synced" -> caller hard-fails, does NOT drop
+    config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role")
+
+
+def test_resolve_or_drop_absent_from_both_is_permanently_stale():
+    config = _make_config()
+
+    resolved, stale = _call(config, "ghost-role", "roles")
+
+    assert resolved is None
+    assert stale is True  # absent from destination AND source -> permanently gone
+    config.state.ensure_resource_loaded.assert_called_once_with("roles", "ghost-role")
+
+
+def test_resolve_or_drop_unknown_resource_type_does_not_raise():
+    # state.destination/source are defaultdict(dict); an unknown type yields {} not KeyError.
+    config = _make_config()
+
+    resolved, stale = _call(config, "whatever", "never_seen_type")
+
+    assert resolved is None
+    assert stale is True
+
+
+def test_resolve_or_drop_propagates_ensure_resource_loaded_exception():
+    config = _make_config()
+    config.state.ensure_resource_loaded.side_effect = RuntimeError("storage down")
+
+    # Not in destination -> ensure_resource_loaded is called -> its error must propagate,
+    # matching every other ensure_resource_loaded call site (none catch it).
+    with pytest.raises(RuntimeError, match="storage down"):
+        _call(config, "src-role", "roles")
+
+
+def test_empty_binding_risk_continues_with_explicit_escalation_warning_when_connection_failures_are_skipped():
+    config = _make_config()
+    config.skip_failed_resource_connections = True
+    config.logger = MagicMock()
+    stub = SimpleNamespace(config=config, resource_type="restriction_policies")
+
+    escalation_risk = BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True)
+
+    assert escalation_risk is True
+    message = config.logger.error.call_args.args[0]
+    assert "continuing sync" in message
+    assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in message
+    assert "refusing to sync" not in message
+
+
+def test_empty_binding_risk_refuses_to_sync_when_connection_failures_are_not_skipped():
+    from datadog_sync.utils.resource_utils import ResourceConnectionError
+
+    config = _make_config()
+    config.skip_failed_resource_connections = False
+    config.logger = MagicMock()
+    stub = SimpleNamespace(config=config, resource_type="restriction_policies")
+
+    with pytest.raises(ResourceConnectionError):
+        BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True)
+
+    message = config.logger.error.call_args.args[0]
+    assert "refusing to sync" in message
+    assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" not in message
diff --git a/tests/unit/test_drop_unresolvable_principals_cli.py b/tests/unit/test_drop_unresolvable_principals_cli.py
new file mode 100644
index 00000000..1dc11f69
--- /dev/null
+++ b/tests/unit/test_drop_unresolvable_principals_cli.py
@@ -0,0 +1,43 @@
+# Unless explicitly stated otherwise all files in this repository are licensed
+# under the 3-clause BSD style license (see LICENSE).
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2019 Datadog, Inc.
+
+"""CLI round-trip test for --drop-unresolvable-principals.
+
+Mirrors test_force_missing_dependencies_cli.py. The flag lives in _diffs_options,
+which is attached to sync, diffs, and migrate (the three commands that run
+connect_resources). exit_code == 2 is click's "no such option" usage error, so
+exit_code != 2 proves the option is recognized by each command.
+"""
+
+import pytest
+from click.testing import CliRunner
+
+from datadog_sync.cli import cli
+
+
+@pytest.fixture
+def runner():
+    return CliRunner(mix_stderr=False)
+
+
+def test_sync_accepts_drop_unresolvable_principals(runner):
+    result = runner.invoke(cli, ["sync", "--drop-unresolvable-principals", "--validate=false"])
+    assert result.exit_code != 2
+
+
+def test_diffs_accepts_drop_unresolvable_principals(runner):
+    result = runner.invoke(cli, ["diffs", "--drop-unresolvable-principals"])
+    assert result.exit_code != 2
+
+
+def test_migrate_accepts_drop_unresolvable_principals(runner):
+    result = runner.invoke(cli, ["migrate", "--drop-unresolvable-principals", "--validate=false"])
+    assert result.exit_code != 2
+
+
+def test_import_rejects_drop_unresolvable_principals(runner):
+    # The flag is NOT attached to import (import does not run connect_resources).
+    result = runner.invoke(cli, ["import", "--drop-unresolvable-principals"])
+    assert result.exit_code == 2
diff --git a/tests/unit/test_resource_utils.py b/tests/unit/test_resource_utils.py
index c2780356..ca243115 100644
--- a/tests/unit/test_resource_utils.py
+++ b/tests/unit/test_resource_utils.py
@@ -7,7 +7,19 @@
 from unittest.mock import MagicMock, call
 
 from datadog_sync import models
-from datadog_sync.utils.resource_utils import find_attr
+from datadog_sync.utils.resource_utils import find_attr, ResourceConnectionError
+
+
+def test_resource_connection_error_empty_binding_risk_defaults_false():
+    exc = ResourceConnectionError({"roles": ["r1"]})
+    assert exc.empty_binding_risk is False
+
+
+def test_resource_connection_error_empty_binding_risk_settable():
+    exc = ResourceConnectionError({"roles": ["r1"]}, empty_binding_risk=True)
+    assert exc.empty_binding_risk is True
+    # Message formatting is unchanged by the new kwarg.
+    assert "Failed to connect resource." in str(exc)
 
 
 @pytest.fixture(scope="class")
diff --git a/tests/unit/test_workers.py b/tests/unit/test_workers.py
new file mode 100644
index 00000000..bf923ac4
--- /dev/null
+++ b/tests/unit/test_workers.py
@@ -0,0 +1,72 @@
+# Unless explicitly stated otherwise all files in this repository are licensed
+# under the 3-clause BSD style license (see LICENSE).
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2019 Datadog, Inc.
+
+"""Unit tests for the Counter buckets added for --drop-unresolvable-principals.
+
+Guards two things: (1) the new fields default empty and record correctly, and
+(2) reset_counter() zeroes them so repeated import/sync invocations in one process
+don't double-count.
+"""
+
+from datadog_sync.utils.workers import Counter
+
+
+def test_new_counter_fields_default_empty():
+    counter = Counter()
+    assert counter.stale_principals_dropped_by_type == {}
+    assert counter.empty_binding_risk_by_type == {}
+    assert counter.empty_binding_escalation_by_type == {}
+
+
+def test_record_stale_principal_dropped_appends_by_type():
+    counter = Counter()
+    counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:abc")
+    counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:def")
+    counter.record_stale_principal_dropped(resource_type="monitors", _id="mon-1")
+
+    assert counter.stale_principals_dropped_by_type["restriction_policies"] == [
+        "dashboard:abc",
+        "dashboard:def",
+    ]
+    assert counter.stale_principals_dropped_by_type["monitors"] == ["mon-1"]
+
+
+def test_record_empty_binding_risk_appends_by_type():
+    counter = Counter()
+    counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1")
+
+    assert counter.empty_binding_risk_by_type["restriction_policies"] == ["dashboard:dash-1"]
+
+
+def test_record_empty_binding_escalation_appends_by_type():
+    counter = Counter()
+    counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1")
+
+    assert counter.empty_binding_escalation_by_type["restriction_policies"] == ["dashboard:dash-1"]
+
+
+def test_record_methods_ignore_none_args():
+    counter = Counter()
+    counter.record_stale_principal_dropped(resource_type=None, _id="x")
+    counter.record_stale_principal_dropped(resource_type="roles", _id=None)
+    counter.record_empty_binding_risk(resource_type=None, _id=None)
+    counter.record_empty_binding_escalation(resource_type=None, _id=None)
+
+    assert counter.stale_principals_dropped_by_type == {}
+    assert counter.empty_binding_risk_by_type == {}
+    assert counter.empty_binding_escalation_by_type == {}
+
+
+def test_reset_counter_clears_new_fields():
+    counter = Counter()
+    counter.record_stale_principal_dropped(resource_type="roles", _id="r1")
+    counter.record_empty_binding_risk(resource_type="roles", _id="r2")
+    counter.record_empty_binding_escalation(resource_type="roles", _id="r3")
+
+    counter.reset_counter()
+
+    assert counter.stale_principals_dropped_by_type == {}
+    assert counter.empty_binding_risk_by_type == {}
+    assert counter.empty_binding_escalation_by_type == {}