# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import dataclasses
import functools
import itertools
import json
import logging
import os.path
from collections import defaultdict
from collections.abc import Coroutine, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import PurePath
from typing import Any, DefaultDict, NamedTuple, Type, TypeVar, cast

from pants.base.deprecated import warn_or_error
from pants.base.specs import AncestorGlobSpec, RawSpecsWithoutFileOwners, RecursiveGlobSpec
from pants.build_graph.address import BuildFileAddressRequest, ResolveError
from pants.engine.addresses import Address, Addresses, AddressInput, UnparsedAddressInputs
from pants.engine.collection import Collection
from pants.engine.environment import ChosenLocalEnvironmentName, EnvironmentName
from pants.engine.fs import EMPTY_SNAPSHOT, GlobMatchErrorBehavior, PathGlobs
from pants.engine.internals import native_engine
from pants.engine.internals.build_files import (
    AddressFamilyDir,
    BuildFileOptions,
    ensure_address_family,
    find_build_file,
    find_target_adaptor,
    maybe_resolve_address,
    parse_address_family,
    resolve_address,
)
from pants.engine.internals.mapper import AddressFamilies, SpecsFilter
from pants.engine.internals.native_engine import AddressParseException
from pants.engine.internals.parametrize import Parametrize, _TargetParametrization
from pants.engine.internals.parametrize import (  # noqa: F401
    _TargetParametrizations as _TargetParametrizations,
)
from pants.engine.internals.parametrize import (  # noqa: F401
    _TargetParametrizationsRequest as _TargetParametrizationsRequest,
)
from pants.engine.internals.synthetic_targets import (
    SyntheticTargetsSpecPathsRequest,
    get_synthetic_targets_spec_paths,
)
from pants.engine.internals.target_adaptor import SourceBlocks, TargetAdaptor, TargetAdaptorRequest
from pants.engine.intrinsics import digest_to_snapshot, path_globs_to_paths
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import (
    AllTargets,
    AllUnexpandedTargets,
    CoarsenedTarget,
    CoarsenedTargets,
    CoarsenedTargetsRequest,
    Dependencies,
    DependenciesRequest,
    DepsTraversalBehavior,
    ExplicitlyProvidedDependencies,
    ExplicitlyProvidedDependenciesRequest,
    Field,
    FieldDefaultFactoryRequest,
    FieldDefaultFactoryResult,
    FieldDefaults,
    FieldSetsPerTarget,
    FieldSetsPerTargetRequest,
    FilteredTargets,
    GeneratedTargets,
    GenerateSourcesRequest,
    GenerateTargetsRequest,
    HydratedSources,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
    InvalidFieldException,
    MultipleSourcesField,
    OverridesField,
    RegisteredTargetTypes,
    SourcesField,
    SourcesPaths,
    SourcesPathsRequest,
    SpecialCasedDependencies,
    Target,
    TargetFilesGenerator,
    TargetFilesGeneratorSettings,
    TargetFilesGeneratorSettingsRequest,
    TargetGenerator,
    Targets,
    TargetTypesToGenerateTargetsRequests,
    TransitivelyExcludeDependencies,
    TransitivelyExcludeDependenciesRequest,
    TransitiveTargets,
    TransitiveTargetsRequest,
    UnexpandedTargets,
    UnrecognizedTargetTypeException,
    ValidatedDependencies,
    ValidateDependenciesRequest,
    WrappedTarget,
    WrappedTargetRequest,
    _generate_file_level_targets,
    generate_sources,
    generate_targets,
)
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.bootstrap_options import UnmatchedBuildFileGlobs
from pants.option.global_options import GlobalOptions
from pants.util.docutil import bin_name, doc_url
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.memo import memoized
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import bullet_list, pluralize, softwrap
from pants.vcs.hunk import TextBlocks

logger = logging.getLogger(__name__)


# -----------------------------------------------------------------------------------------------
# Address -> Target(s)
# -----------------------------------------------------------------------------------------------


@rule(_masked_types=[EnvironmentName])
async def filter_targets(targets: Targets, specs_filter: SpecsFilter) -> FilteredTargets:
    return FilteredTargets(tgt for tgt in targets if specs_filter.matches(tgt))


@rule
async def target_types_to_generate_targets_requests(
    union_membership: UnionMembership,
) -> TargetTypesToGenerateTargetsRequests:
    return TargetTypesToGenerateTargetsRequests(
        {
            request_cls.generate_from: request_cls  # type: ignore[misc]
            for request_cls in union_membership.get(GenerateTargetsRequest)
        }
    )


@memoized
def warn_deprecated_target_type(tgt_type: type[Target]) -> None:
    assert tgt_type.deprecated_alias_removal_version is not None
    warn_or_error(
        removal_version=tgt_type.deprecated_alias_removal_version,
        entity=f"the target name {tgt_type.deprecated_alias}",
        hint=(
            f"Instead, use `{tgt_type.alias}`, which behaves the same. Run `{bin_name()} "
            "update-build-files` to automatically fix your BUILD files."
        ),
    )


@memoized
def warn_deprecated_field_type(field_type: type[Field]) -> None:
    assert field_type.deprecated_alias_removal_version is not None
    warn_or_error(
        removal_version=field_type.deprecated_alias_removal_version,
        entity=f"the field name {field_type.deprecated_alias}",
        hint=(
            f"Instead, use `{field_type.alias}`, which behaves the same. Run `{bin_name()} "
            "update-build-files` to automatically fix your BUILD files."
        ),
    )


@dataclass(frozen=True)
class _AdaptorAndType:
    adaptor: TargetAdaptor
    target_type: type[Target]


@dataclass(frozen=True)
class _RequestAdaptorAndType:
    address: Address
    description_of_origin: str


@rule
async def _determine_target_adaptor_and_type(
    req: _RequestAdaptorAndType, registered_target_types: RegisteredTargetTypes
) -> _AdaptorAndType:
    target_adaptor = await find_target_adaptor(
        TargetAdaptorRequest(req.address, description_of_origin=req.description_of_origin)
    )
    target_type = registered_target_types.aliases_to_types.get(target_adaptor.type_alias, None)
    if target_type is None:
        raise UnrecognizedTargetTypeException(
            target_adaptor.type_alias,
            registered_target_types,
            req.address,
            target_adaptor.description_of_origin,
        )
    if (
        target_type.deprecated_alias is not None
        and target_type.deprecated_alias == target_adaptor.type_alias
        and not req.address.is_generated_target
    ):
        warn_deprecated_target_type(target_type)
    return _AdaptorAndType(target_adaptor, target_type)


@dataclass(frozen=True)
class _TargetGeneratorOverridesRequest:
    target_generator: TargetGenerator


@dataclass(frozen=True)
class ResolvedTargetGeneratorRequests:
    requests: tuple[GenerateTargetsRequest, ...] = tuple()


@dataclass(frozen=True)
class ResolveTargetGeneratorRequests:
    address: Address
    description_of_origin: str = dataclasses.field(hash=False, compare=False)


@dataclass(frozen=True)
class ResolveAllTargetGeneratorRequests:
    description_of_origin: str = dataclasses.field(hash=False, compare=False)
    of_type: type[TargetGenerator] | None = None


async def _parametrized_target_generators_with_templates(
    address: Address,
    target_adaptor: TargetAdaptor,
    target_type: type[TargetGenerator],
    generator_fields: dict[str, Any],
    union_membership: UnionMembership,
) -> list[tuple[TargetGenerator, Mapping[str, Any]]]:
    # Pre-load field values from defaults for the target type being generated.
    if hasattr(target_type, "generated_target_cls"):
        family = (
            await parse_address_family(**implicitly(AddressFamilyDir(address.spec_path)))
        ).ensure()
        template_fields = dict(family.defaults.get(target_type.generated_target_cls.alias, {}))
    else:
        template_fields = {}

    # Split out the `propagated_fields` before construction.
    copied_fields = (
        *target_type.copied_fields,
        *target_type._find_copied_plugin_fields(union_membership),
    )
    moved_fields = (
        *target_type.moved_fields,
        *target_type._find_moved_plugin_fields(union_membership),
    )
    for field_type in copied_fields:
        for alias in (field_type.deprecated_alias, field_type.alias):
            if alias is None:
                continue
            # Any deprecated field use will be checked on the generator target.
            field_value = generator_fields.get(alias, None)
            if field_value is not None:
                template_fields[alias] = field_value
    for field_type in moved_fields:
        # We must check for deprecated field usage here before passing the value to the generator.
        if field_type.deprecated_alias is not None:
            field_value = generator_fields.pop(field_type.deprecated_alias, None)
            if field_value is not None:
                warn_deprecated_field_type(field_type)
                template_fields[field_type.deprecated_alias] = field_value
        field_value = generator_fields.pop(field_type.alias, None)
        if field_value is not None:
            template_fields[field_type.alias] = field_value

    # Move parametrize groups over to `template_fields` in order to expand them.
    parametrize_group_field_names = [
        name
        for name, field in generator_fields.items()
        if isinstance(field, Parametrize) and field.is_group
    ]
    for field_name in parametrize_group_field_names:
        template_fields[field_name] = generator_fields.pop(field_name)

    field_type_aliases = target_type._get_field_aliases_to_field_types(
        target_type.class_field_types(union_membership)
    ).keys()
    generator_fields_parametrized = {
        name
        for name, field in generator_fields.items()
        if isinstance(field, Parametrize) and name in field_type_aliases
    }
    if generator_fields_parametrized:
        noun = pluralize(len(generator_fields_parametrized), "field", include_count=False)
        generator_fields_parametrized_text = ", ".join(
            repr(f) for f in generator_fields_parametrized
        )
        raise InvalidFieldException(
            f"Only fields which will be moved to generated targets may be parametrized, "
            f"so target generator {address} (with type {target_type.alias}) cannot "
            f"parametrize the {generator_fields_parametrized_text} {noun}."
        )
    return [
        (
            _create_target(
                address,
                target_type,
                target_adaptor,
                generator_fields,
                union_membership,
                name_explicitly_set=target_adaptor.name is not None,
            ),
            template,
        )
        for address, template in Parametrize.expand(address, template_fields)
    ]


async def _target_generator_overrides(
    target_generator: TargetGenerator, unmatched_build_file_globs: UnmatchedBuildFileGlobs
) -> dict[str, dict[str, Any]]:
    address = target_generator.address
    if target_generator.has_field(OverridesField):
        overrides_field = target_generator[OverridesField]
        overrides_flattened = overrides_field.flatten()
    else:
        overrides_flattened = {}
    if isinstance(target_generator, TargetFilesGenerator):
        override_globs = OverridesField.to_path_globs(
            address, overrides_flattened, unmatched_build_file_globs
        )
        override_paths = await concurrently(
            path_globs_to_paths(path_globs) for path_globs in override_globs
        )
        return OverridesField.flatten_paths(
            address, zip(override_paths, override_globs, overrides_flattened.values())
        )
    return overrides_flattened


@rule
async def resolve_generator_target_requests(
    req: ResolveTargetGeneratorRequests,
    union_membership: UnionMembership,
    target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests,
    unmatched_build_file_globs: UnmatchedBuildFileGlobs,
) -> ResolvedTargetGeneratorRequests:
    adaptor_and_type = await _determine_target_adaptor_and_type(
        _RequestAdaptorAndType(req.address, req.description_of_origin), **implicitly()
    )
    target_adaptor = adaptor_and_type.adaptor
    target_type = adaptor_and_type.target_type
    if not issubclass(target_type, TargetGenerator):
        return ResolvedTargetGeneratorRequests()

    generate_request = target_types_to_generate_requests.request_for(target_type)
    if not generate_request:
        return ResolvedTargetGeneratorRequests()
    generator_fields = dict(target_adaptor.kwargs)
    generators = await _parametrized_target_generators_with_templates(
        req.address,
        target_adaptor,
        target_type,
        generator_fields,
        union_membership,
    )
    base_generator = _create_target(
        req.address,
        target_type,
        target_adaptor,
        generator_fields,
        union_membership,
    )
    overrides = await _target_generator_overrides(base_generator, unmatched_build_file_globs)
    return ResolvedTargetGeneratorRequests(
        requests=tuple(
            generate_request(
                generator,
                template_address=generator.address,
                template=template,
                overrides={
                    name: dict(Parametrize.expand(generator.address, override))
                    for name, override in overrides.items()
                },
            )
            for generator, template in generators
        )
    )


@rule(_masked_types=[EnvironmentName])
async def address_families_from_raw_specs_without_file_owners(
    specs: RawSpecsWithoutFileOwners,
    build_file_options: BuildFileOptions,
) -> AddressFamilies:
    if not (specs.dir_literals or specs.dir_globs or specs.recursive_globs or specs.ancestor_globs):
        return AddressFamilies()
    # Resolve all globs.
    build_file_globs, validation_globs = specs.to_build_file_path_globs_tuple(
        build_patterns=build_file_options.patterns,
        build_ignore_patterns=build_file_options.ignores,
    )
    build_file_paths, _ = await concurrently(
        path_globs_to_paths(build_file_globs),
        path_globs_to_paths(validation_globs),
    )
    dirnames = set(
        await get_synthetic_targets_spec_paths(
            SyntheticTargetsSpecPathsRequest(tuple(specs.glob_specs())), **implicitly()
        )
    )
    dirnames.update(os.path.dirname(f) for f in build_file_paths.files)
    return AddressFamilies(
        await concurrently(
            ensure_address_family(**implicitly(AddressFamilyDir(d))) for d in dirnames
        )
    )


@rule
async def resolve_all_generator_target_requests(
    req: ResolveAllTargetGeneratorRequests,
) -> ResolvedTargetGeneratorRequests:
    address_families = await address_families_from_raw_specs_without_file_owners(
        RawSpecsWithoutFileOwners(
            recursive_globs=(RecursiveGlobSpec(""),),
            description_of_origin="the `ResolveAllTargetGeneratorRequests` rule",
        ),
        **implicitly(),
    )
    results = await concurrently(
        resolve_generator_target_requests(
            ResolveTargetGeneratorRequests(address, req.description_of_origin), **implicitly()
        )
        for family in address_families
        for address, target_adaptor in family.addresses_to_target_adaptors.items()
        if not req.of_type or target_adaptor.type_alias == req.of_type.alias
    )
    return ResolvedTargetGeneratorRequests(
        tuple(itertools.chain.from_iterable(result.requests for result in results))
    )


@rule
async def resolve_target_parametrizations(
    request: _TargetParametrizationsRequest,
    union_membership: UnionMembership,
    env_name: EnvironmentName,
) -> _TargetParametrizations:
    address = request.address
    adaptor_and_type = await _determine_target_adaptor_and_type(
        _RequestAdaptorAndType(request.address, request.description_of_origin), **implicitly()
    )
    target_adaptor = adaptor_and_type.adaptor
    target_type = adaptor_and_type.target_type

    parametrizations: list[_TargetParametrization] = []
    requests: ResolvedTargetGeneratorRequests | None = None
    if issubclass(target_type, TargetGenerator):
        requests = await resolve_generator_target_requests(
            ResolveTargetGeneratorRequests(address, request.description_of_origin), **implicitly()
        )
    if requests and requests.requests:
        all_generated = await concurrently(
            generate_targets(
                **implicitly({generate_request: GenerateTargetsRequest, env_name: EnvironmentName})
            )
            for generate_request in requests.requests
        )
        parametrizations.extend(
            _TargetParametrization(generate_request.generator, generated_batch)
            for generated_batch, generate_request in zip(all_generated, requests.requests)
        )
    else:
        parametrizations.append(
            _target_parametrizations(address, target_adaptor, target_type, union_membership)
        )

    return _TargetParametrizations(parametrizations)


@rule(_masked_types=[EnvironmentName])
async def resolve_target(
    request: WrappedTargetRequest,
    target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests,
    local_environment_name: ChosenLocalEnvironmentName,
) -> WrappedTarget:
    address = request.address
    base_address = address.maybe_convert_to_target_generator()
    parametrizations = await resolve_target_parametrizations(
        **implicitly(
            {
                _TargetParametrizationsRequest(
                    base_address, description_of_origin=request.description_of_origin
                ): _TargetParametrizationsRequest,
                local_environment_name.val: EnvironmentName,
            }
        )
    )
    target = parametrizations.get(address, target_types_to_generate_requests)
    if target is None:
        raise ResolveError(
            softwrap(
                f"""
                The address `{address}` from {request.description_of_origin} was not generated by
                the target `{base_address}`. Did you mean one of these addresses?

                {bullet_list(str(t.address) for t in parametrizations.all)}
                """
            )
        )
    return WrappedTarget(target)


@rule(_masked_types=[EnvironmentName])
async def resolve_unexpanded_targets(addresses: Addresses) -> UnexpandedTargets:
    wrapped_targets = await concurrently(
        resolve_target(
            WrappedTargetRequest(
                a,
                # Idiomatic rules should not be manually constructing `Addresses`. Instead, they
                # should use `UnparsedAddressInputs` or `Specs` rules.
                #
                # It is technically more correct for us to require callers of
                # `Addresses -> UnexpandedTargets` to specify a `description_of_origin`. But in
                # practice, this dramatically increases boilerplate, and it should never be
                # necessary.
                #
                # Note that this contrasts with an individual `Address`, which often is unverified
                # because it can come from the rule `AddressInput -> Address`, which only verifies
                # that it has legal syntax and does not check the address exists.
                description_of_origin="<infallible>",
            ),
            **implicitly(),
        )
        for a in addresses
    )
    return UnexpandedTargets(wrapped_target.target for wrapped_target in wrapped_targets)


_TargetType = TypeVar("_TargetType", bound=Target)


def _create_target(
    address: Address,
    target_type: type[_TargetType],
    target_adaptor: TargetAdaptor,
    field_values: Mapping[str, Any],
    union_membership: UnionMembership,
    name_explicitly_set: bool | None = None,
) -> _TargetType:
    target = target_type(
        field_values,
        address,
        name_explicitly_set=(
            target_adaptor.name_explicitly_set
            if name_explicitly_set is None
            else name_explicitly_set
        ),
        union_membership=union_membership,
        description_of_origin=target_adaptor.description_of_origin,
        origin_sources_blocks=target_adaptor.origin_sources_blocks,
    )
    # Check for any deprecated field usage.
    for field_type in target.field_types:
        if field_type.deprecated_alias is not None and field_type.deprecated_alias in field_values:
            warn_deprecated_field_type(field_type)

    return target


def _target_parametrizations(
    address: Address,
    target_adaptor: TargetAdaptor,
    target_type: type[Target],
    union_membership: UnionMembership,
) -> _TargetParametrization:
    expanded_parametrizations = tuple(Parametrize.expand(address, target_adaptor.kwargs))
    first_address, first_kwargs = expanded_parametrizations[0]
    if first_address is not address:
        # The target was parametrized, and so the original Target does not exist.
        generated = FrozenDict(
            (
                parameterized_address,
                _create_target(
                    parameterized_address,
                    target_type,
                    target_adaptor,
                    parameterized_fields,
                    union_membership,
                ),
            )
            for parameterized_address, parameterized_fields in expanded_parametrizations
        )
        return _TargetParametrization(None, generated)
    else:
        # The target was not parametrized.
        target = _create_target(
            address,
            target_type,
            target_adaptor,
            target_adaptor.kwargs,
            union_membership,
        )
        return _TargetParametrization(target, FrozenDict())


@dataclass(frozen=True)
class WrappedTargetForBootstrap:
    """Used to avoid a rule graph cycle when evaluating bootstrap targets.

    This does not work with target generation and parametrization. It also ignores any unrecognized
    fields in the target, to accommodate plugin fields which are not yet registered during
    bootstrapping.

    This should only be used by bootstrapping code.
    """

    val: Target


@rule
async def resolve_target_for_bootstrapping(
    request: WrappedTargetRequest,
    union_membership: UnionMembership,
) -> WrappedTargetForBootstrap:
    adaptor_and_type = await _determine_target_adaptor_and_type(
        _RequestAdaptorAndType(
            request.address,
            description_of_origin=request.description_of_origin,
        ),
        **implicitly(),
    )
    target_adaptor = adaptor_and_type.adaptor
    target_type = adaptor_and_type.target_type
    target = target_type(
        target_adaptor.kwargs,
        request.address,
        name_explicitly_set=target_adaptor.name_explicitly_set,
        union_membership=union_membership,
        ignore_unrecognized_fields=True,
        description_of_origin=target_adaptor.description_of_origin,
        origin_sources_blocks=target_adaptor.origin_sources_blocks,
    )
    return WrappedTargetForBootstrap(target)


@rule(_masked_types=[EnvironmentName])
async def resolve_targets(
    targets: UnexpandedTargets,
    target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests,
    local_environment_name: ChosenLocalEnvironmentName,
) -> Targets:
    # Replace all generating targets with what they generate. Otherwise, keep them. If a target
    # generator does not generate any targets, keep the target generator.
    # TODO: This method does not preserve the order of inputs.
    expanded_targets: OrderedSet[Target] = OrderedSet()
    generator_targets = []
    parametrizations_gets = []
    for tgt in targets:
        if (
            target_types_to_generate_requests.is_generator(tgt)
            and not tgt.address.is_generated_target
        ):
            generator_targets.append(tgt)
            parametrizations_gets.append(
                resolve_target_parametrizations(
                    **implicitly(
                        {
                            _TargetParametrizationsRequest(
                                tgt.address.maybe_convert_to_target_generator(),
                                # Idiomatic rules should not be manually creating `UnexpandedTargets`, so
                                # we can be confident that the targets actually exist and the addresses
                                # are already legitimate.
                                description_of_origin="<infallible>",
                            ): _TargetParametrizationsRequest,
                            local_environment_name.val: EnvironmentName,
                        }
                    )
                )
            )
        else:
            expanded_targets.add(tgt)

    all_generated_targets = await concurrently(parametrizations_gets)
    expanded_targets.update(
        tgt
        for generator, parametrizations in zip(generator_targets, all_generated_targets)
        for tgt in parametrizations.generated_or_generator(generator.address)
    )
    return Targets(expanded_targets)


@rule(desc="Find all targets in the project", level=LogLevel.DEBUG, _masked_types=[EnvironmentName])
async def find_all_targets() -> AllTargets:
    tgts = await resolve_targets(
        **implicitly(
            RawSpecsWithoutFileOwners(
                recursive_globs=(RecursiveGlobSpec(""),),
                description_of_origin="the `AllTargets` rule",
            )
        )
    )
    return AllTargets(tgts)


@rule(
    desc="Find all (unexpanded) targets in the project",
    level=LogLevel.DEBUG,
    _masked_types=[EnvironmentName],
)
async def find_all_unexpanded_targets() -> AllUnexpandedTargets:
    tgts = await resolve_unexpanded_targets(
        **implicitly(
            RawSpecsWithoutFileOwners(
                recursive_globs=(RecursiveGlobSpec(""),),
                description_of_origin="the `AllTargets` rule",
            )
        )
    )
    return AllUnexpandedTargets(tgts)


# -----------------------------------------------------------------------------------------------
# TransitiveTargets
# -----------------------------------------------------------------------------------------------


class CycleException(Exception):
    def __init__(self, subject: Address, path: tuple[Address, ...]) -> None:
        path_string = "\n".join((f"-> {a}" if a == subject else f"   {a}") for a in path)
        super().__init__(
            f"The dependency graph contained a cycle:\n{path_string}\n\nTo fix this, first verify "
            "if your code has an actual import cycle. If it does, you likely need to re-architect "
            "your code to avoid the cycle.\n\nIf there is no cycle in your code, then you may need "
            "to use more granular targets. Split up the problematic targets into smaller targets "
            "with more granular `sources` fields so that you can adjust the `dependencies` fields "
            "to avoid introducing a cycle.\n\nAlternatively, use Python dependency inference "
            "(`--python-infer-imports`), rather than explicit `dependencies`. Pants will infer "
            "dependencies on specific files, rather than entire targets. This extra precision "
            "means that you will only have cycles if your code actually does have cycles in it."
        )
        self.subject = subject
        self.path = path


def _detect_cycles(
    roots: tuple[Address, ...], dependency_mapping: dict[Address, tuple[Address, ...]]
) -> None:
    path_stack: OrderedSet[Address] = OrderedSet()
    visited: set[Address] = set()

    def maybe_report_cycle(address: Address) -> None:
        # NB: File-level dependencies are cycle tolerant.
        if address.is_file_target or address not in path_stack:
            return

        # The path of the cycle is shorter than the entire path to the cycle: if the suffix of
        # the path representing the cycle contains a file dep, it is ignored.
        in_cycle = False
        for path_address in path_stack:
            if in_cycle and path_address.is_file_target:
                # There is a file address inside the cycle: do not report it.
                return
            elif in_cycle:
                # Not a file address.
                continue
            else:
                # We're entering the suffix of the path that contains the cycle if we've reached
                # the address in question.
                in_cycle = path_address == address
        # If we did not break out early, it's because there were no file addresses in the cycle.
        raise CycleException(address, (*path_stack, address))

    def visit(address: Address):
        if address in visited:
            maybe_report_cycle(address)
            return
        path_stack.add(address)
        visited.add(address)

        for dep_address in dependency_mapping[address]:
            visit(dep_address)

        path_stack.remove(address)

    for root in roots:
        visit(root)
        if path_stack:
            raise AssertionError(
                f"The stack of visited nodes should have been empty at the end of recursion, "
                f"but it still contained: {path_stack}"
            )


@dataclass(frozen=True)
class _DependencyMappingRequest:
    tt_request: TransitiveTargetsRequest
    expanded_targets: bool


@dataclass(frozen=True)
class _DependencyMapping:
    mapping: FrozenDict[Address, tuple[Address, ...]]
    visited: FrozenOrderedSet[Target]
    roots_as_targets: Collection[Target]


@rule
async def transitive_dependency_mapping(request: _DependencyMappingRequest) -> _DependencyMapping:
    """This uses iteration, rather than recursion, so that we can tolerate dependency cycles.

    Unlike a traditional BFS algorithm, we batch each round of traversals via `MultiGet` for
    improved performance / concurrency.
    """
    roots_as_targets = await resolve_unexpanded_targets(Addresses(request.tt_request.roots))
    visited: OrderedSet[Target] = OrderedSet()
    queued = FrozenOrderedSet(roots_as_targets)
    dependency_mapping: dict[Address, tuple[Address, ...]] = {}
    while queued:
        direct_dependencies: tuple[Collection[Target], ...]
        if request.expanded_targets:
            direct_dependencies = await concurrently(  # noqa: PNT30: this is inherently sequential
                resolve_targets(
                    **implicitly(
                        DependenciesRequest(
                            tgt.get(Dependencies),
                            should_traverse_deps_predicate=request.tt_request.should_traverse_deps_predicate,
                        )
                    )
                )
                for tgt in queued
            )
        else:
            direct_dependencies = await concurrently(  # noqa: PNT30: this is inherently sequential
                resolve_unexpanded_targets(
                    **implicitly(
                        DependenciesRequest(
                            tgt.get(Dependencies),
                            should_traverse_deps_predicate=request.tt_request.should_traverse_deps_predicate,
                        )
                    )
                )
                for tgt in queued
            )

        dependency_mapping.update(
            zip(
                (t.address for t in queued),
                (tuple(t.address for t in deps) for deps in direct_dependencies),
            )
        )

        queued = FrozenOrderedSet(itertools.chain.from_iterable(direct_dependencies)).difference(
            visited
        )
        visited.update(queued)

    # NB: We use `roots_as_targets` to get the root addresses, rather than `request.roots`. This
    # is because expanding from the `Addresses` -> `Targets` may have resulted in generated
    # targets being used, so we need to use `roots_as_targets` to have this expansion.
    # TODO(#12871): Fix this to not be based on generated targets.
    _detect_cycles(tuple(t.address for t in roots_as_targets), dependency_mapping)
    return _DependencyMapping(
        FrozenDict(dependency_mapping), FrozenOrderedSet(visited), roots_as_targets
    )


class SubprojectRoots(Collection[str]):
