diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py
index 07930cb6..a875a936 100644
--- a/datadog_sync/utils/resources_handler.py
+++ b/datadog_sync/utils/resources_handler.py
@@ -38,7 +38,7 @@ def __init__(self, config: Configuration) -> None:
         self.sorter: Optional[TopologicalSorter] = None
         self.cleanup_sorter: Optional[TopologicalSorter] = None
         self.worker: Optional[Workers] = None
-        self._dependency_graph = Optional[Dict[Tuple[str, str], List[Tuple[str, str]]]]
+        self._dependency_graph: Optional[Dict[Tuple[str, str], Set[Tuple[str, str]]]] = None
 
     @staticmethod
     def _sanitize_reason(err: Exception) -> str:
@@ -109,7 +109,7 @@ async def reset(self) -> None:
 
     async def apply_resources(self) -> Tuple[int, int]:
         # Build dependency graph and missing resources
-        self._dependency_graph, missing = self.get_dependency_graph()
+        self._dependency_graph, missing, filtered_count = self.get_dependency_graph()
 
         # Import resources that are missing but needed for resource connections
         if self.config.force_missing_dependencies and missing:
@@ -206,6 +206,7 @@ async def apply_resources(self) -> Tuple[int, int]:
             )
         else:
             await self.worker.schedule_workers(additional_coros=[self.run_sorter()])
+        self.worker.counter.filtered = filtered_count
         self.config.logger.info(f"finished syncing resource items: {self.worker.counter}.")
 
         self.config.state.dump_state()
@@ -216,14 +217,6 @@ async def _apply_resource_cb(self, q_item: List) -> None:
 
         try:
             r_class = self.config.resources[resource_type]
-
-            # Filter BEFORE deepcopy to avoid unnecessary memory allocation.
-            # Safe because filter() only reads the resource dict, never mutates it.
-            if not r_class.filter(self.config.state.source[resource_type][_id]):
-                self.worker.counter.increment_filtered()
-                self._emit(resource_type, _id, "sync", "filtered")
-                return
-
             resource = deepcopy(self.config.state.source[resource_type][_id])
 
             if not r_class.resource_config.concurrent:
@@ -283,7 +276,7 @@ async def _apply_resource_cb(self, q_item: List) -> None:
                 r_class.resource_config.async_lock.release()
 
     async def diffs(self) -> None:
-        self._dependency_graph, _ = self.get_dependency_graph()
+        self._dependency_graph, _, _ = self.get_dependency_graph()
 
         # Run pre-apply hooks
         resource_types = set(i[0] for i in self._dependency_graph.keys())
@@ -318,10 +311,6 @@ async def _diffs_worker_cb(self, q_item: List) -> None:
             else:
                 resource = self.config.state.source[resource_type][_id]
 
-                if not r_class.filter(resource):
-                    self._emit(resource_type, _id, "sync", "filtered")
-                    return
-
                 try:
                     await r_class._pre_resource_action_hook(_id, resource)
                 except SkipResource as e:
@@ -542,22 +531,41 @@ async def run_cleanup_sorter(self):
 
             await asyncio.sleep(0)
 
-    def get_dependency_graph(self) -> Tuple[Dict[Tuple[str, str], List[Tuple[str, str]]], Set[Tuple[str, str]]]:
+    def get_dependency_graph(
+        self,
+    ) -> Tuple[Dict[Tuple[str, str], Set[Tuple[str, str]]], Set[Tuple[str, str]], int]:
         """Build the dependency graph for all resources.
 
         Returns:
-            Tuple[Dict[Tuple[str, str], List[Tuple[str, str]]], Set[Tuple[str, str]]]: Returns
-            a tuple of the dependency graph and missing resources.
+            Tuple of (dependency_graph, missing_resources, filtered_count).
         """
         dependency_graph = {}
         missing_resources = set()
+        filtered_out = set()
+
+        for (resource_type, _id), resource in self.config.state.get_all_resources(self.config.resources_arg).items():
+            r_class = self.config.resources[resource_type]
+            if not r_class.filter(resource):
+                filtered_out.add((resource_type, _id))
+                continue
 
-        for resource_type, _id in self.config.state.get_all_resources(self.config.resources_arg).keys():
             deps, missing = self._resource_connections(resource_type, _id)
             dependency_graph[(resource_type, _id)] = deps
             missing_resources.update(missing)
 
-        return dependency_graph, missing_resources
+        # Emit filtered outcomes so --json consumers see which resources were excluded.
+        for resource_type, _id in filtered_out:
+            self._emit(resource_type, _id, "sync", "filtered")
+
+        # Remove dependency references to filtered-out resources only.
+        # Cross-type deps on resource types outside resources_arg must be
+        # preserved as phantom nodes — TopologicalSorter yields them first,
+        # ensuring dependencies are synced before dependents.
+        if filtered_out:
+            for key in dependency_graph:
+                dependency_graph[key] = dependency_graph[key] - filtered_out
+
+        return dependency_graph, missing_resources, len(filtered_out)
 
     def _resource_connections(self, resource_type: str, _id: str) -> Tuple[Set[Tuple[str, str]], Set[Tuple[str, str]]]:
         """Returns the failed connections and missing resources for a given resource.
diff --git a/datadog_sync/utils/workers.py b/datadog_sync/utils/workers.py
index 50e0e224..7cd66cb3 100644
--- a/datadog_sync/utils/workers.py
+++ b/datadog_sync/utils/workers.py
@@ -117,7 +117,7 @@ def __str__(self):
         )
 
     def reset_counter(self) -> None:
-        self.successes = self.failure = self.skipped = 0
+        self.successes = self.failure = self.skipped = self.filtered = 0
 
     def increment_success(self) -> None:
         self.successes += 1
diff --git a/tests/unit/test_dependency_graph.py b/tests/unit/test_dependency_graph.py
new file mode 100644
index 00000000..f79e06bb
--- /dev/null
+++ b/tests/unit/test_dependency_graph.py
@@ -0,0 +1,408 @@
+# 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.
+
+import pytest
+from unittest.mock import patch
+
+from datadog_sync.utils.resources_handler import ResourcesHandler
+from datadog_sync.utils.filter import process_filters
+
+
+@pytest.fixture
+def graph_test(config):
+    """Provides (handler, config) with full state isolation.
+
+    Saves config.filters, config.filter_operator, config.resources_arg,
+    and all source state before each test; restores after.
+    """
+    saved_filters = config.filters
+    saved_operator = config.filter_operator
+    saved_resources_arg = config.resources_arg[:]
+    saved_source = {}
+    for rt in config.resources:
+        saved_source[rt] = dict(config.state.source[rt])
+
+    handler = ResourcesHandler(config)
+
+    yield handler, config
+
+    config.filters = saved_filters
+    config.filter_operator = saved_operator
+    config.resources_arg = saved_resources_arg
+    for rt in config.resources:
+        config.state.source[rt].clear()
+        config.state.source[rt].update(saved_source.get(rt, {}))
+
+
+def setup_state(config, source_resources, resources_arg=None):
+    """Populate state.source and optionally set resources_arg.
+
+    Clears ALL resource types in source state first.
+    """
+    for rt in config.resources:
+        config.state.source[rt].clear()
+    for rt, resources in source_resources.items():
+        for _id, resource in resources.items():
+            config.state.source[rt][_id] = resource
+    if resources_arg is not None:
+        config.resources_arg = resources_arg
+    else:
+        config.resources_arg = list(source_resources.keys())
+
+
+def setup_filters(config, filter_strings, operator="OR"):
+    """Set filters on config from filter string list."""
+    config.filters = process_filters(filter_strings)
+    config.filter_operator = operator
+
+
+# ---------------------------------------------------------------------------
+# GREEN tests — must pass before AND after the change
+# ---------------------------------------------------------------------------
+
+
+def test_no_filters_all_resources_in_graph(graph_test):
+    handler, config = graph_test
+    config.filters = {}
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+                "dash-3": {"id": "dash-3"},
+            },
+            "monitors": {
+                "mon-1": {"id": "mon-1", "name": "Monitor 1"},
+                "mon-2": {"id": "mon-2", "name": "Monitor 2"},
+            },
+        },
+        resources_arg=["dashboards", "monitors"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert len(graph) == 5
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") in graph
+    assert ("dashboards", "dash-3") in graph
+    assert ("monitors", "mon-1") in graph
+    assert ("monitors", "mon-2") in graph
+
+
+def test_filter_matching_all_resources_same_as_no_filter(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=.*"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+                "dash-3": {"id": "dash-3"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert len(graph) == 3
+
+
+def test_resources_with_no_connections_have_empty_deps(graph_test):
+    handler, config = graph_test
+    config.filters = {}
+    setup_state(
+        config,
+        {
+            "monitors": {
+                "mon-1": {"id": "mon-1", "name": "Monitor 1"},
+                "mon-2": {"id": "mon-2", "name": "Monitor 2"},
+            },
+        },
+        resources_arg=["monitors"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert len(graph) == 2
+    assert graph[("monitors", "mon-1")] == set()
+    assert graph[("monitors", "mon-2")] == set()
+
+
+def test_missing_dependencies_detected(graph_test):
+    handler, config = graph_test
+    config.filters = {}
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1", "widgets": [{"definition": {"alert_id": "mon-99"}}]},
+            },
+        },
+        resources_arg=["dashboards", "monitors"],
+    )
+
+    _, missing, _ = handler.get_dependency_graph()
+
+    assert ("monitors", "mon-99") in missing
+
+
+def test_empty_state_returns_empty_graph(graph_test):
+    handler, config = graph_test
+    config.filters = {}
+    setup_state(config, {}, resources_arg=["dashboards"])
+
+    graph, missing, _ = handler.get_dependency_graph()
+
+    assert graph == {}
+    assert missing == set()
+
+
+def test_cross_type_dep_preserved_when_both_sides_in_graph(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1", "widgets": [{"definition": {"alert_id": "mon-1"}}]},
+            },
+            "monitors": {
+                "mon-1": {"id": "mon-1", "name": "Monitor 1"},
+            },
+        },
+        resources_arg=["dashboards", "monitors"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("monitors", "mon-1") in graph
+    assert ("monitors", "mon-1") in graph[("dashboards", "dash-1")]
+
+
+def test_cross_type_phantom_deps_preserved(graph_test):
+    """Deps on out-of-scope resource types are preserved as phantom nodes
+    so TopologicalSorter yields them first, ensuring correct ordering."""
+    handler, config = graph_test
+    config.filters = {}
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1", "widgets": [{"definition": {"alert_id": "mon-1"}}]},
+            },
+            "monitors": {
+                "mon-1": {"id": "mon-1", "name": "Monitor 1"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("monitors", "mon-1") in graph[("dashboards", "dash-1")]
+
+
+# ---------------------------------------------------------------------------
+# RED tests — fail before the change, pass after
+# ---------------------------------------------------------------------------
+
+
+def test_filter_excludes_resources_from_graph(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+                "dash-3": {"id": "dash-3"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") not in graph
+    assert ("dashboards", "dash-3") not in graph
+
+
+def test_filtered_out_deps_stripped_from_graph_values(graph_test):
+    """Deps pointing to resources excluded by --filter are stripped,
+    preventing phantom nodes for resources we don't want to process."""
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") not in graph
+    # dash-2 was filtered out, so any dep referencing it should be stripped
+    for deps in graph.values():
+        assert ("dashboards", "dash-2") not in deps
+
+
+def test_filtered_resources_deps_not_computed(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    with patch.object(
+        ResourcesHandler,
+        "_resource_connections",
+        wraps=handler._resource_connections,
+    ) as mock_rc:
+        graph, _, _ = handler.get_dependency_graph()
+
+    call_args = [c[0] for c in mock_rc.call_args_list]
+    assert ("dashboards", "dash-1") in call_args
+    assert ("dashboards", "dash-2") not in call_args
+
+
+def test_filter_on_one_type_doesnt_affect_other_types(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+            },
+            "monitors": {
+                "mon-1": {"id": "mon-1", "name": "Monitor 1"},
+                "mon-2": {"id": "mon-2", "name": "Monitor 2"},
+            },
+        },
+        resources_arg=["dashboards", "monitors"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") not in graph
+    assert ("monitors", "mon-1") in graph
+    assert ("monitors", "mon-2") in graph
+
+
+def test_or_filter_operator_includes_any_match(graph_test):
+    handler, config = graph_test
+    setup_filters(
+        config,
+        [
+            "Type=dashboards;Name=id;Value=^dash-1$",
+            "Type=dashboards;Name=id;Value=^dash-3$",
+        ],
+        operator="OR",
+    )
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+                "dash-3": {"id": "dash-3"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") not in graph
+    assert ("dashboards", "dash-3") in graph
+
+
+def test_and_filter_operator_requires_all_match(graph_test):
+    handler, config = graph_test
+    setup_filters(
+        config,
+        [
+            "Type=dashboards;Name=id;Value=^dash-1$",
+            "Type=dashboards;Name=title;Value=^My Dashboard$",
+        ],
+        operator="AND",
+    )
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1", "title": "My Dashboard"},
+                "dash-2": {"id": "dash-2", "title": "My Dashboard"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert ("dashboards", "dash-1") in graph
+    assert ("dashboards", "dash-2") not in graph
+
+
+def test_all_resources_filtered_returns_empty_graph(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^nonexistent$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1"},
+                "dash-2": {"id": "dash-2"},
+            },
+        },
+        resources_arg=["dashboards"],
+    )
+
+    graph, _, _ = handler.get_dependency_graph()
+
+    assert graph == {}
+
+
+def test_missing_deps_only_from_filtered_resources(graph_test):
+    handler, config = graph_test
+    setup_filters(config, ["Type=dashboards;Name=id;Value=^dash-1$"])
+    setup_state(
+        config,
+        {
+            "dashboards": {
+                "dash-1": {"id": "dash-1", "widgets": [{"definition": {"alert_id": "mon-99"}}]},
+                "dash-2": {"id": "dash-2", "widgets": [{"definition": {"alert_id": "mon-88"}}]},
+            },
+        },
+        resources_arg=["dashboards", "monitors"],
+    )
+
+    _, missing, _ = handler.get_dependency_graph()
+
+    assert ("monitors", "mon-99") in missing
+    assert ("monitors", "mon-88") not in missing
diff --git a/tests/unit/test_report_e2e.py b/tests/unit/test_report_e2e.py
index 625a5c55..6fdeafd6 100644
--- a/tests/unit/test_report_e2e.py
+++ b/tests/unit/test_report_e2e.py
@@ -356,7 +356,7 @@ def test_update_coexists_with_create(self, runner):
 
 
 class TestFilteredOutcome:
-    """Test that --filter excludes resources and emits filtered status."""
+    """Test that --filter emits filtered outcomes for excluded resources."""
 
     def test_filtered_resources_emitted(self, runner):
         _setup_source_dashboards()
@@ -378,9 +378,10 @@ def test_filtered_resources_emitted(self, runner):
         outcomes = _parse_outcomes(ret.output)
         filtered = [o for o in outcomes if o["status"] == "filtered"]
         non_filtered = [o for o in outcomes if o["status"] != "filtered"]
-        # def-456 and ghi-789 should be filtered out, only abc-123 passes
+        # abc-123 passes the filter; def-456 and ghi-789 are filtered out
+        # and emitted as "filtered" outcomes.
         assert len(non_filtered) == 1, f"Expected exactly 1 non-filtered, got {non_filtered}"
-        assert len(filtered) == 2, f"Expected exactly 2 filtered outcomes, got {filtered}"
+        assert len(filtered) == 2, f"Expected 2 filtered outcomes, got {filtered}"
 
 
 class TestDeleteOutcome: