Replace include_special_cased_deps flag with should_traverse_deps_predicate

src/python/pants/backend/project_info/dependencies.py

@@ -14,6 +14,7 @@
     TransitiveTargets,
     TransitiveTargetsRequest,
     UnexpandedTargets,
+    always_traverse,
 )
 from pants.option.option_types import BoolOption
 
@@ -43,7 +44,8 @@ async def dependencies(
 ) -> Dependencies:
     if dependencies_subsystem.transitive:
         transitive_targets = await Get(
-            TransitiveTargets, TransitiveTargetsRequest(addresses, include_special_cased_deps=True)
+            TransitiveTargets,
+            TransitiveTargetsRequest(addresses, should_traverse_deps_predicate=always_traverse),
         )
         targets = Targets(transitive_targets.dependencies)
     else:
@@ -55,7 +57,9 @@ async def dependencies(
         dependencies_per_target_root = await MultiGet(
             Get(
                 Targets,
-                DependenciesRequest(tgt.get(DependenciesField), include_special_cased_deps=True),
+                DependenciesRequest(
+                    tgt.get(DependenciesField), should_traverse_deps_predicate=always_traverse
+                ),
             )
             for tgt in target_roots
         )

src/python/pants/backend/project_info/dependents.py

@@ -10,7 +10,12 @@
 from pants.engine.console import Console
 from pants.engine.goal import Goal, GoalSubsystem, LineOriented
 from pants.engine.rules import Get, MultiGet, collect_rules, goal_rule, rule
-from pants.engine.target import AllUnexpandedTargets, Dependencies, DependenciesRequest
+from pants.engine.target import (
+    AllUnexpandedTargets,
+    Dependencies,
+    DependenciesRequest,
+    always_traverse,
+)
 from pants.option.option_types import BoolOption
 from pants.util.frozendict import FrozenDict
 from pants.util.logging import LogLevel
@@ -25,7 +30,12 @@ class AddressToDependents:
 @rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)
 async def map_addresses_to_dependents(all_targets: AllUnexpandedTargets) -> AddressToDependents:
     dependencies_per_target = await MultiGet(
-        Get(Addresses, DependenciesRequest(tgt.get(Dependencies), include_special_cased_deps=True))
+        Get(
+            Addresses,
+            DependenciesRequest(
+                tgt.get(Dependencies), should_traverse_deps_predicate=always_traverse
+            ),
+        )
         for tgt in all_targets
     )
 

src/python/pants/backend/project_info/filedeps.py

@@ -19,6 +19,7 @@
     TransitiveTargets,
     TransitiveTargetsRequest,
     UnexpandedTargets,
+    always_traverse,
 )
 from pants.option.option_types import BoolOption
 from pants.util.strutil import softwrap
@@ -72,7 +73,8 @@ async def file_deps(
     targets: Iterable[Target]
     if filedeps_subsystem.transitive:
         transitive_targets = await Get(
-            TransitiveTargets, TransitiveTargetsRequest(addresses, include_special_cased_deps=True)
+            TransitiveTargets,
+            TransitiveTargetsRequest(addresses, should_traverse_deps_predicate=always_traverse),
         )
         targets = transitive_targets.closure
     else:

src/python/pants/backend/project_info/paths.py

@@ -19,6 +19,7 @@
     Targets,
     TransitiveTargets,
     TransitiveTargetsRequest,
+    always_traverse,
 )
 from pants.option.option_types import StrOption
 
@@ -114,13 +115,16 @@ async def paths(console: Console, paths_subsystem: PathsSubsystem) -> PathsGoal:
     destination = to_tgts.expect_single()
 
     transitive_targets = await Get(
-        TransitiveTargets, TransitiveTargetsRequest([root.address], include_special_cased_deps=True)
+        TransitiveTargets,
+        TransitiveTargetsRequest([root.address], should_traverse_deps_predicate=always_traverse),
     )
 
     adjacent_targets_per_target = await MultiGet(
         Get(
             Targets,
-            DependenciesRequest(tgt.get(Dependencies), include_special_cased_deps=True),
+            DependenciesRequest(
+                tgt.get(Dependencies), should_traverse_deps_predicate=always_traverse
+            ),
         )
         for tgt in transitive_targets.closure
     )

src/python/pants/backend/project_info/peek.py

@@ -30,6 +30,7 @@
     Target,
     Targets,
     UnexpandedTargets,
+    always_traverse,
 )
 from pants.option.option_types import BoolOption
 from pants.util.strutil import softwrap
@@ -179,7 +180,9 @@ async def get_target_data(
     dependencies_per_target = await MultiGet(
         Get(
             Targets,
-            DependenciesRequest(tgt.get(Dependencies), include_special_cased_deps=True),
+            DependenciesRequest(
+                tgt.get(Dependencies), should_traverse_deps_predicate=always_traverse
+            ),
         )
         for tgt in sorted_targets
     )

src/python/pants/backend/visibility/lint.py

@@ -18,6 +18,7 @@
     DependenciesRuleApplication,
     DependenciesRuleApplicationRequest,
     FieldSet,
+    always_traverse,
 )
 from pants.util.logging import LogLevel
 
@@ -39,7 +40,12 @@ async def check_visibility_rule_violations(
     request: EnforceVisibilityRules.Batch, platform: Platform, run_id: RunId
 ) -> LintResult:
     all_dependencies = await MultiGet(
-        Get(Addresses, DependenciesRequest(field_set.dependencies, include_special_cased_deps=True))
+        Get(
+            Addresses,
+            DependenciesRequest(
+                field_set.dependencies, should_traverse_deps_predicate=always_traverse
+            ),
+        )
         for field_set in request.elements
     )
     all_dependencies_rule_action = await MultiGet(

src/python/pants/engine/internals/graph.py

@@ -47,6 +47,7 @@
     Dependencies,
     DependenciesRequest,
     ExplicitlyProvidedDependencies,
+    ExplicitlyProvidedDependenciesRequest,
     Field,
     FieldDefaultFactoryRequest,
     FieldDefaultFactoryResult,
@@ -700,7 +701,7 @@ async def transitive_dependency_mapping(request: _DependencyMappingRequest) -> _
                     Targets,
                     DependenciesRequest(
                         tgt.get(Dependencies),
-                        include_special_cased_deps=request.tt_request.include_special_cased_deps,
+                        should_traverse_deps_predicate=request.tt_request.should_traverse_deps_predicate,
                     ),
                 )
                 for tgt in queued
@@ -711,7 +712,7 @@ async def transitive_dependency_mapping(request: _DependencyMappingRequest) -> _
                     UnexpandedTargets,
                     DependenciesRequest(
                         tgt.get(Dependencies),
-                        include_special_cased_deps=request.tt_request.include_special_cased_deps,
+                        should_traverse_deps_predicate=request.tt_request.should_traverse_deps_predicate,
                     ),
                 )
                 for tgt in queued
@@ -787,7 +788,8 @@ async def coarsened_targets(
         _DependencyMapping,
         _DependencyMappingRequest(
             TransitiveTargetsRequest(
-                request.roots, include_special_cased_deps=request.include_special_cased_deps
+                request.roots,
+                should_traverse_deps_predicate=request.should_traverse_deps_predicate,
             ),
             expanded_targets=request.expanded_targets,
         ),
@@ -1205,8 +1207,24 @@ def __init__(
 
 
 @rule
-async def determine_explicitly_provided_dependencies(
+async def convert_dependencies_request_to_explicitly_provided_dependencies_request(
     request: DependenciesRequest,
+) -> ExplicitlyProvidedDependenciesRequest:
+    """This rule discards any deps predicate from DependenciesRequest.
+
+    Calculating ExplicitlyProvidedDependencies does not use any deps traversal predicates as it is
+    meant to list all explicit deps from the given field. By stripping the predicate from the
+    request, we ensure that the cache key for ExplicitlyProvidedDependencies calculation does not
+    include the predicate increasing the cache-hit rate.
+    """
+    # TODO: Maybe require Get(ExplicitlyProvidedDependencies, ExplicitlyProvidedDependenciesRequest)
+    #       and deprecate Get(ExplicitlyProvidedDependencies, DependenciesRequest) via this rule.
+    return ExplicitlyProvidedDependenciesRequest(request.field)
+
+
+@rule
+async def determine_explicitly_provided_dependencies(
+    request: ExplicitlyProvidedDependenciesRequest,
     union_membership: UnionMembership,
     registered_target_types: RegisteredTargetTypes,
     subproject_roots: SubprojectRoots,
@@ -1300,6 +1318,13 @@ async def resolve_dependencies(
         WrappedTargetRequest(request.field.address, description_of_origin="<infallible>"),
     )
     tgt = wrapped_tgt.target
+
+    # This predicate allows the dep graph to ignore dependencies of selected targets
+    # including any explicit deps and any inferred deps.
+    # For example, to avoid traversing the deps of package targets.
+    if not request.should_traverse_deps_predicate(tgt, request.field):
+        return Addresses([])
+
     try:
         explicitly_provided = await Get(
             ExplicitlyProvidedDependencies, DependenciesRequest, request
@@ -1378,33 +1403,34 @@ async def resolve_dependencies(
     # If the target has `SpecialCasedDependencies`, such as the `archive` target having
     # `files` and `packages` fields, then we possibly include those too. We don't want to always
     # include those dependencies because they should often be excluded from the result due to
-    # being handled elsewhere in the calling code.
-    special_cased: tuple[Address, ...] = ()
-    if request.include_special_cased_deps:
-        # Unlike normal, we don't use `tgt.get()` because there may be >1 subclass of
-        # SpecialCasedDependencies.
-        special_cased_fields = tuple(
-            field
-            for field in tgt.field_values.values()
-            if isinstance(field, SpecialCasedDependencies)
-        )
-        # We can't use the normal `Get(Addresses, UnparsedAddressInputs)` due to a graph cycle.
-        special_cased = await MultiGet(
-            Get(
-                Address,
-                AddressInput,
-                AddressInput.parse(
-                    addr,
-                    relative_to=tgt.address.spec_path,
-                    subproject_roots=subproject_roots,
-                    description_of_origin=(
-                        f"the `{special_cased_field.alias}` field from the target {tgt.address}"
-                    ),
+    # being handled elsewhere in the calling code. So, we only include fields based on
+    # the should_traverse_deps_predicate.
+
+    # Unlike normal, we don't use `tgt.get()` because there may be >1 subclass of
+    # SpecialCasedDependencies.
+    special_cased_fields = tuple(
+        field
+        for field in tgt.field_values.values()
+        if isinstance(field, SpecialCasedDependencies)
+        and request.should_traverse_deps_predicate(tgt, field)
+    )
+    # We can't use the normal `Get(Addresses, UnparsedAddressInputs)` due to a graph cycle.
+    special_cased = await MultiGet(
+        Get(
+            Address,
+            AddressInput,
+            AddressInput.parse(
+                addr,
+                relative_to=tgt.address.spec_path,
+                subproject_roots=subproject_roots,
+                description_of_origin=(
+                    f"the `{special_cased_field.alias}` field from the target {tgt.address}"
                 ),
-            )
-            for special_cased_field in special_cased_fields
-            for addr in special_cased_field.to_unparsed_address_inputs().values
+            ),
         )
+        for special_cased_field in special_cased_fields
+        for addr in special_cased_field.to_unparsed_address_inputs().values
+    )
 
     excluded = explicitly_provided_ignores.union(
         *itertools.chain(deps.exclude for deps in inferred)

src/python/pants/engine/internals/graph_test.py

@@ -65,6 +65,7 @@
     Targets,
     TransitiveTargets,
     TransitiveTargetsRequest,
+    always_traverse,
 )
 from pants.engine.unions import UnionMembership, UnionRule
 from pants.source.filespec import Filespec
@@ -314,7 +315,8 @@ def get_target(name: str) -> Target:
     assert direct_deps == Targets()
 
     direct_deps = transitive_targets_rule_runner.request(
-        Targets, [DependenciesRequest(root[Dependencies], include_special_cased_deps=True)]
+        Targets,
+        [DependenciesRequest(root[Dependencies], should_traverse_deps_predicate=always_traverse)],
     )
     assert direct_deps == Targets([d1, d2, d3])
 
@@ -327,7 +329,11 @@ def get_target(name: str) -> Target:
 
     transitive_targets = transitive_targets_rule_runner.request(
         TransitiveTargets,
-        [TransitiveTargetsRequest([root.address, d2.address], include_special_cased_deps=True)],
+        [
+            TransitiveTargetsRequest(
+                [root.address, d2.address], should_traverse_deps_predicate=always_traverse
+            )
+        ],
     )
     assert transitive_targets.roots == (root, d2)
     assert transitive_targets.dependencies == FrozenOrderedSet([d1, d2, d3, t2, t1])
@@ -367,6 +373,68 @@ def test_transitive_targets_tolerates_generated_target_cycles(
     ]
 
 
+def test_transitive_targets_with_should_traverse_deps_predicate(
+    transitive_targets_rule_runner: RuleRunner,
+) -> None:
+    transitive_targets_rule_runner.write_files(
+        {
+            "BUILD": dedent(
+                """\
+                target(name='t1')
+                target(name='t2', dependencies=[':t1'])
+                target(name='t3')
+                target(name='d1', dependencies=[':t1'])
+                target(name='d2', dependencies=[':t2'])
+                target(name='d3')
+                target(name='skipped', dependencies=[':t3'])
+                target(name='d4', tags=['skip_deps'], dependencies=[':skipped'])
+                target(name='root', dependencies=[':d1', ':d2', ':d3', ':d4'])
+                """
+            ),
+        }
+    )
+
+    def get_target(name: str) -> Target:
+        return transitive_targets_rule_runner.get_target(Address("", target_name=name))
+
+    t1 = get_target("t1")
+    t2 = get_target("t2")
+    t3 = get_target("t3")
+    d1 = get_target("d1")
+    d2 = get_target("d2")
+    d3 = get_target("d3")
+    skipped = get_target("skipped")
+    d4 = get_target("d4")
+    root = get_target("root")
+
+    direct_deps = transitive_targets_rule_runner.request(
+        Targets, [DependenciesRequest(root[Dependencies])]
+    )
+    assert direct_deps == Targets([d1, d2, d3, d4])
+
+    transitive_targets = transitive_targets_rule_runner.request(
+        TransitiveTargets,
+        [
+            TransitiveTargetsRequest(
+                [root.address, d2.address],
+                should_traverse_deps_predicate=lambda tgt, fld: "skip_deps"
+                not in (tgt[Tags].value or []),
+            )
+        ],
+    )
+    assert transitive_targets.roots == (root, d2)
+    # NB: `//:d2` is both a target root and a dependency of `//:root`.
+    assert transitive_targets.dependencies == FrozenOrderedSet([d1, d2, d3, d4, t2, t1])
+    assert transitive_targets.closure == FrozenOrderedSet([root, d2, d1, d3, d4, t2, t1])
+    # `//:d4` depends on `//:skipped` which depends on `//:t3`.
+    # Nothing else depends on `//:skipped` or `//:t3`, so they should not
+    # be present in the list of transitive deps thanks to `should_traverse_deps_predicate`.
+    assert skipped not in transitive_targets.dependencies
+    assert t3 not in transitive_targets.dependencies
+    assert skipped not in transitive_targets.closure
+    assert t3 not in transitive_targets.closure
+
+
 def test_coarsened_targets(transitive_targets_rule_runner: RuleRunner) -> None:
     """CoarsenedTargets should "coarsen" a cycle into a single CoarsenedTarget instance."""
     transitive_targets_rule_runner.write_files(