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

from __future__ import annotations

import dataclasses
import json
import logging
import os
import shlex
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import PurePath
from textwrap import dedent  # noqa: PNT20
from typing import TypeVar

import packaging.specifiers
import packaging.version
from packaging.requirements import Requirement

from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import (
    Executable,
    MainSpecification,
    PexCompletePlatformsField,
    PexLayout,
    PythonRequirementFindLinksField,
    PythonRequirementsField,
)
from pants.backend.python.util_rules import pex_cli, pex_requirements
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex_cli import PexCliProcess, PexPEX, maybe_log_pex_stderr
from pants.backend.python.util_rules.pex_environment import (
    CompletePexEnvironment,
    PexEnvironment,
    PexSubsystem,
    PythonExecutable,
)
from pants.backend.python.util_rules.pex_requirements import (
    EntireLockfile,
    LoadedLockfile,
    LoadedLockfileRequest,
    Lockfile,
    Resolve,
    ResolvePexConfigRequest,
    determine_resolve_pex_config,
    get_lockfile_for_resolve,
    load_lockfile,
    validate_metadata,
)
from pants.backend.python.util_rules.pex_requirements import (
    PexRequirements as PexRequirements,  # Explicit re-export.
)
from pants.build_graph.address import Address
from pants.core.environments.target_types import EnvironmentTarget
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
from pants.core.util_rules.stripped_source_files import rules as stripped_source_rules
from pants.core.util_rules.system_binaries import BashBinary
from pants.engine.addresses import UnparsedAddressInputs
from pants.engine.collection import Collection, DeduplicatedCollection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.environment import EnvironmentName
from pants.engine.fs import EMPTY_DIGEST, AddPrefix, CreateDigest, Digest, FileContent, MergeDigests
from pants.engine.internals.graph import (
    hydrate_sources,
    resolve_targets,
    resolve_unparsed_address_inputs,
)
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.internals.native_engine import Snapshot
from pants.engine.intrinsics import add_prefix, create_digest, digest_to_snapshot, merge_digests
from pants.engine.process import (
    Process,
    ProcessCacheScope,
    ProcessResult,
    fallible_to_exec_result_or_raise,
)
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import HydrateSourcesRequest, SourcesField, TransitiveTargetsRequest
from pants.engine.unions import UnionMembership, union
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import bullet_list, pluralize, softwrap

logger = logging.getLogger(__name__)


@union(in_scope_types=[EnvironmentName])
@dataclass(frozen=True)
class PythonProvider:
    """Union which should have 0 or 1 implementations registered which provide Python.

    Subclasses should provide a rule from their subclass type to `PythonExecutable`.
    """

    interpreter_constraints: InterpreterConstraints


@rule(polymorphic=True)
async def get_python_executable(
    provider: PythonProvider, env_name: EnvironmentName
) -> PythonExecutable:
    raise NotImplementedError()


class PexPlatforms(DeduplicatedCollection[str]):
    sort_input = True

    def generate_pex_arg_list(self) -> list[str]:
        args = []
        for platform in self:
            args.extend(["--platform", platform])
        return args


class CompletePlatforms(DeduplicatedCollection[str]):
    sort_input = True

    def __init__(self, iterable: Iterable[str] = (), *, digest: Digest = EMPTY_DIGEST):
        super().__init__(iterable)
        self._digest = digest

    @classmethod
    def from_snapshot(cls, snapshot: Snapshot) -> CompletePlatforms:
        return cls(snapshot.files, digest=snapshot.digest)

    @property
    def digest(self) -> Digest:
        return self._digest

    def generate_pex_arg_list(self) -> Iterator[str]:
        for path in self:
            yield "--complete-platform"
            yield path


@rule
async def digest_complete_platform_addresses(
    addresses: UnparsedAddressInputs,
) -> CompletePlatforms:
    original_file_targets = await resolve_targets(**implicitly(addresses))
    original_files_sources = await concurrently(
        hydrate_sources(
            HydrateSourcesRequest(
                tgt.get(SourcesField),
                for_sources_types=(
                    FileSourceField,
                    ResourceSourceField,
                ),
                enable_codegen=True,
            ),
            **implicitly(),
        )
        for tgt in original_file_targets
    )
    snapshot = await digest_to_snapshot(
        **implicitly(MergeDigests(sources.snapshot.digest for sources in original_files_sources))
    )
    return CompletePlatforms.from_snapshot(snapshot)


@rule
async def digest_complete_platforms(
    complete_platforms: PexCompletePlatformsField,
) -> CompletePlatforms:
    return await digest_complete_platform_addresses(complete_platforms.to_unparsed_address_inputs())


@dataclass(frozen=True)
class PexRequest(EngineAwareParameter):
    output_filename: str
    internal_only: bool
    layout: PexLayout
    python: PythonExecutable | None
    requirements: PexRequirements | EntireLockfile
    interpreter_constraints: InterpreterConstraints
    platforms: PexPlatforms
    complete_platforms: CompletePlatforms
    sources: Digest | None
    additional_inputs: Digest
    main: MainSpecification | None
    inject_args: tuple[str, ...]
    inject_env: FrozenDict[str, str]
    additional_args: tuple[str, ...]
    pex_path: tuple[Pex, ...]
    description: str | None = dataclasses.field(compare=False)
    cache_scope: ProcessCacheScope
    scie_output_files: Iterable[str] | None = None
    scie_output_directories: Iterable[str] | None = None

    def __init__(
        self,
        *,
        output_filename: str,
        scie_output_files: Iterable[str] | None = None,
        scie_output_directories: Iterable[str] | None = None,
        internal_only: bool,
        layout: PexLayout | None = None,
        python: PythonExecutable | None = None,
        requirements: PexRequirements | EntireLockfile = PexRequirements(),
        interpreter_constraints=InterpreterConstraints(),
        platforms=PexPlatforms(),
        complete_platforms=CompletePlatforms(),
        sources: Digest | None = None,
        additional_inputs: Digest | None = None,
        main: MainSpecification | None = None,
        inject_args: Iterable[str] = (),
        inject_env: Mapping[str, str] = FrozenDict(),
        additional_args: Iterable[str] = (),
        pex_path: Iterable[Pex] = (),
        description: str | None = None,
        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
    ) -> None:
        """A request to create a PEX from its inputs.

        :param output_filename: The name of the built Pex file, which typically should end in
            `.pex`.
        :param internal_only: Whether we ever materialize the Pex and distribute it directly
            to end users, such as with the `binary` goal. Typically, instead, the user never
            directly uses the Pex, e.g. with `lint` and `test`. If True, we will use a Pex setting
            that results in faster build time but compatibility with fewer interpreters at runtime.
        :param layout: The filesystem layout to create the PEX with.
        :param python: A particular PythonExecutable to use, which must match any relevant
            interpreter_constraints.
        :param requirements: The requirements that the PEX should contain.
        :param interpreter_constraints: Any constraints on which Python versions may be used.
        :param platforms: Which abbreviated platforms should be supported. Setting this value will
            cause interpreter constraints to not be used at PEX build time because platforms already
            constrain the valid Python versions, e.g. by including `cp36m` in the platform string.
            Unfortunately this also causes interpreter constraints to not be embedded in the built
            PEX for use at runtime which can lead to problems.
            See: https://github.com/pantsbuild/pants/issues/13904.
        :param complete_platforms: Which complete platforms should be supported. Setting this value
            will cause interpreter constraints to not be used at PEX build time because complete
            platforms completely constrain the valid Python versions. Unfortunately this also causes
            interpreter constraints to not be embedded in the built PEX for use at runtime which can
            lead to problems. See: https://github.com/pantsbuild/pants/issues/13904.
        :param sources: Any source files that should be included in the Pex.
        :param additional_inputs: Any inputs that are not source files and should not be included
            directly in the Pex, but should be present in the environment when building the Pex.
        :param main: The main for the built Pex, equivalent to Pex's `-e` or '-c' flag. If
            left off, the Pex will open up as a REPL.
        :param inject_args: Command line arguments to freeze in to the PEX.
        :param inject_env: Environment variables to freeze in to the PEX.
        :param additional_args: Any additional Pex flags.
        :param pex_path: Pex files to add to the PEX_PATH.
        :param description: A human-readable description to render in the dynamic UI when building
            the Pex.
        :param cache_scope: The cache scope for the underlying pex cli invocation process.
        :param scie_output_files If we are also building native executable scies for the PEX, their filenames
        :param scie_output_directories If we are also building native executable scies for the PEX, using a directory layout, their dirs.
        """
        object.__setattr__(self, "output_filename", output_filename)
        object.__setattr__(self, "scie_output_files", scie_output_files)
        object.__setattr__(self, "scie_output_directories", scie_output_directories)
        object.__setattr__(self, "internal_only", internal_only)
        # Use any explicitly requested layout, or Packed for internal PEXes (which is a much
        # friendlier layout for the CAS than Zipapp.)
        object.__setattr__(
            self, "layout", layout or (PexLayout.PACKED if internal_only else PexLayout.ZIPAPP)
        )
        object.__setattr__(self, "python", python)
        object.__setattr__(self, "requirements", requirements)
        object.__setattr__(self, "interpreter_constraints", interpreter_constraints)
        object.__setattr__(self, "platforms", platforms)
        object.__setattr__(self, "complete_platforms", complete_platforms)
        object.__setattr__(self, "sources", sources)
        object.__setattr__(self, "additional_inputs", additional_inputs or EMPTY_DIGEST)
        object.__setattr__(self, "main", main)
        object.__setattr__(self, "inject_args", tuple(inject_args))
        object.__setattr__(self, "inject_env", FrozenDict(inject_env))
        object.__setattr__(self, "additional_args", tuple(additional_args))
        object.__setattr__(self, "pex_path", tuple(pex_path))
        object.__setattr__(self, "description", description)
        object.__setattr__(self, "cache_scope", cache_scope)

        self.__post_init__()

    def __post_init__(self):
        if self.internal_only and self.platforms:
            raise ValueError(
                softwrap(
                    f"""
                    Internal only PEXes can only constrain interpreters with interpreter_constraints.
                    Given platform constraints {self.platforms} for internal only pex request:
                    {self}.
                    """
                )
            )
        if self.internal_only and self.complete_platforms:
            raise ValueError(
                softwrap(
                    f"""
                    Internal only PEXes can only constrain interpreters with interpreter_constraints.
                    Given complete_platform constraints {self.complete_platforms} for internal only
                    pex request: {self}.
                    """
                )
            )
        if self.python and self.platforms:
            raise ValueError(
                softwrap(
                    f"""
                    Only one of platforms or a specific interpreter may be set. Got
                    both {self.platforms} and {self.python}.
                    """
                )
            )
        if self.python and self.complete_platforms:
            raise ValueError(
                softwrap(
                    f"""
                    Only one of complete_platforms or a specific interpreter may be set. Got
                    both {self.complete_platforms} and {self.python}.
                    """
                )
            )
        if self.python and self.interpreter_constraints:
            raise ValueError(
                softwrap(
                    f"""
                    Only one of interpreter_constraints or a specific interpreter may be set. Got
                    both {self.interpreter_constraints} and {self.python}.
                    """
                )
            )

    def debug_hint(self) -> str:
        return self.output_filename


@dataclass(frozen=True)
class OptionalPexRequest:
    maybe_pex_request: PexRequest | None


@dataclass(frozen=True)
class Pex:
    """Wrapper for a digest containing a pex file created with some filename."""

    digest: Digest
    name: str
    python: PythonExecutable | None


@dataclass(frozen=True)
class OptionalPex:
    maybe_pex: Pex | None


@rule(desc="Find Python interpreter for constraints", level=LogLevel.DEBUG)
async def find_interpreter(
    interpreter_constraints: InterpreterConstraints,
    pex_subsystem: PexSubsystem,
    env_target: EnvironmentTarget,
    union_membership: UnionMembership,
) -> PythonExecutable:
    python_providers = union_membership.get(PythonProvider)
    if len(python_providers) > 1:
        raise ValueError(
            softwrap(
                f"""
                Too many Python provider plugins were registered. We expected 0 or 1, but found
                {len(python_providers)}. Providers were:

                {bullet_list(repr(provider.__class__) for provider in python_providers)}
                """
            )
        )
    if python_providers:
        python_provider = next(iter(python_providers))
        python = await get_python_executable(
            **implicitly({python_provider(interpreter_constraints): PythonProvider})
        )
        return python

    formatted_constraints = " OR ".join(str(constraint) for constraint in interpreter_constraints)
    result = await fallible_to_exec_result_or_raise(
        **implicitly(
            PexCliProcess(
                description=f"Find interpreter for constraints: {formatted_constraints}",
                subcommand=(),
                # Here, we run the Pex CLI with no requirements, which just selects an interpreter.
                # Normally, this would start an isolated repl. By passing `--`, we force the repl to
                # instead act as an interpreter (the selected one) and tell us about itself. The upshot
                # is we run the Pex interpreter selection logic unperturbed but without resolving any
                # distributions.
                extra_args=(
                    *interpreter_constraints.generate_pex_arg_list(),
                    "--",
                    "-c",
                    # N.B.: The following code snippet must be compatible with Python 2.7 and
                    # Python 3.5+.
                    #
                    # When hashing, we pick 8192 for efficiency of reads and fingerprint updates
                    # (writes) since it's a common OS buffer size and an even multiple of the
                    # hash block size.
                    dedent(
                        """\
                    import hashlib, os, sys

                    python = os.path.realpath(sys.executable)
                    print(python)

                    hasher = hashlib.sha256()
                    with open(python, "rb") as fp:
                      for chunk in iter(lambda: fp.read(8192), b""):
                          hasher.update(chunk)
                    print(hasher.hexdigest())
                    """
                    ),
                ),
                level=LogLevel.DEBUG,
                cache_scope=env_target.executable_search_path_cache_scope(),
            )
        )
    )
    path, fingerprint = result.stdout.decode().strip().splitlines()

    maybe_log_pex_stderr(result.stderr, pex_subsystem.verbosity)

    return PythonExecutable(path=path, fingerprint=fingerprint)


@dataclass(frozen=True)
class BuildPexResult:
    result: ProcessResult
    pex_filename: str
    digest: Digest
    python: PythonExecutable | None

    def create_pex(self) -> Pex:
        return Pex(digest=self.digest, name=self.pex_filename, python=self.python)


@dataclass
class _BuildPexPythonSetup:
    python: PythonExecutable | None
    argv: list[str]


@rule
async def _determine_pex_python_and_platforms(request: PexRequest) -> _BuildPexPythonSetup:
    # NB: If `--platform` is specified, this signals that the PEX should not be built locally.
    # `--interpreter-constraint` only makes sense in the context of building locally. These two
    # flags are mutually exclusive. See https://github.com/pex-tool/pex/issues/957.
    if request.platforms or request.complete_platforms:
        # Note that this means that this is not an internal-only pex.
        # TODO(#9560): consider validating that these platforms are valid with the interpreter
        #  constraints.
        return _BuildPexPythonSetup(
            None,
            [
                *request.platforms.generate_pex_arg_list(),
                *request.complete_platforms.generate_pex_arg_list(),
            ],
        )

    if request.python:
        python = request.python
    else:
        python = await find_interpreter(request.interpreter_constraints, **implicitly())

    if request.python or request.internal_only:
        # Sometimes we want to build and run with a specific interpreter (either because request
        # demanded it, or because it's an internal-only PEX). We will have already validated that
        # there were no platforms.
        return _BuildPexPythonSetup(python, ["--python", python.path])

    else:
        # Otherwise, we don't want to force compatibility with a particular interpreter (as in, the
        # resulting PEX should follow the ICs), but we _do_ want to tell PEX about at least one
        # interpreter that is compatible, to ensure that an interpreter installed/managed by
        # provider backends are visible (in the extreme case, a machine may have no Python
        # interpreters installed at all, and just rely on Pants' provider backends to install them,
        # and thus pex searching $PATH will find nothing).
        return _BuildPexPythonSetup(
            python,
            [
                *request.interpreter_constraints.generate_pex_arg_list(),
                "--python-path",
                python.path,
            ],
        )


@dataclass
class _BuildPexRequirementsSetup:
    digests: list[Digest]
    argv: list[str]
    concurrency_available: int


@dataclass(frozen=True)
class PexRequirementsInfo:
    req_strings: tuple[str, ...]
    find_links: tuple[str, ...]


@rule
async def get_req_strings(pex_reqs: PexRequirements) -> PexRequirementsInfo:
    addrs: list[Address] = []
    specs: list[str] = []
    req_strings: list[str] = []
    find_links: set[str] = set()
    for req_str_or_addr in pex_reqs.req_strings_or_addrs:
        if isinstance(req_str_or_addr, Address):
            addrs.append(req_str_or_addr)
        else:
            assert isinstance(req_str_or_addr, str)
            # Require a `//` prefix, to distinguish address specs from
            # local or VCS requirements.
            if req_str_or_addr.startswith(os.path.sep * 2):
                specs.append(req_str_or_addr)
            else:
                req_strings.append(req_str_or_addr)
    if specs:
        addrs_from_specs = await resolve_unparsed_address_inputs(
            UnparsedAddressInputs(
                specs,
                owning_address=None,
                description_of_origin=pex_reqs.description_of_origin,
            ),
            **implicitly(),
        )
        addrs.extend(addrs_from_specs)
    if addrs:
        transitive_targets = await transitive_targets_get(
            TransitiveTargetsRequest(addrs), **implicitly()
        )
        req_strings.extend(
            PexRequirements.req_strings_from_requirement_fields(
                tgt[PythonRequirementsField]
                for tgt in transitive_targets.closure
                if tgt.has_field(PythonRequirementsField)
            )
        )
        find_links.update(
            find_links
            for tgt in transitive_targets.closure
            if tgt.has_field(PythonRequirementFindLinksField)
            for find_links in tgt[PythonRequirementFindLinksField].value or ()
        )
    return PexRequirementsInfo(tuple(sorted(req_strings)), tuple(sorted(find_links)))


async def _get_entire_lockfile_and_requirements(
    requirements: EntireLockfile | PexRequirements,
) -> tuple[LoadedLockfile | None, tuple[str, ...]]:
    lockfile: Lockfile | None = None
    complete_req_strings: tuple[str, ...] = tuple()
    # TODO: This is clunky, but can be simplified once we get rid of old-style tool
    #  lockfiles, because we can unify EntireLockfile and Resolve.
    if isinstance(requirements, EntireLockfile):
        complete_req_strings = requirements.complete_req_strings or tuple()
        lockfile = requirements.lockfile
    elif (
        isinstance(requirements.from_superset, Resolve)
        and requirements.from_superset.use_entire_lockfile
    ):
        lockfile = await get_lockfile_for_resolve(requirements.from_superset, **implicitly())
    if not lockfile:
        return None, complete_req_strings
    loaded_lockfile = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())
    return loaded_lockfile, complete_req_strings


@rule
async def _setup_pex_requirements(
    request: PexRequest, python_setup: PythonSetup
) -> _BuildPexRequirementsSetup:
    resolve_name: str | None
    if isinstance(request.requirements, EntireLockfile):
        resolve_name = request.requirements.lockfile.resolve_name
    elif isinstance(request.requirements.from_superset, Resolve):
        resolve_name = request.requirements.from_superset.name
    else:
        # This implies that, currently, per-resolve options are only configurable for resolves.
        # However, if no resolve is specified, we will still load options that apply to every
        # resolve, like `[python-repos].indexes`.
        resolve_name = None
    resolve_config = await determine_resolve_pex_config(
        ResolvePexConfigRequest(resolve_name), **implicitly()
    )

    pex_lock_resolver_args = list(resolve_config.pex_args())
    pip_resolver_args = [*resolve_config.pex_args(), "--resolver-version", "pip-2020-resolver"]

    loaded_lockfile, complete_req_strings = await _get_entire_lockfile_and_requirements(
        request.requirements
    )
    if loaded_lockfile:
        argv = (
            ["--lock", loaded_lockfile.lockfile_path, *pex_lock_resolver_args]
            if loaded_lockfile.is_pex_native
            # We use pip to resolve a requirements.txt pseudo-lockfile, possibly with hashes.
            else [
                "--requirement",
                loaded_lockfile.lockfile_path,
                "--no-transitive",
                *pip_resolver_args,
            ]
        )
        if loaded_lockfile.metadata and complete_req_strings:
            validate_metadata(
                loaded_lockfile.metadata,
                request.interpreter_constraints,
                loaded_lockfile.original_lockfile,
                complete_req_strings,
                # We're using the entire lockfile, so there is no Pex subsetting operation we
                # can delegate requirement validation to.  So we do our naive string-matching
                # validation.
                validate_consumed_req_strings=True,
                python_setup=python_setup,
                resolve_config=resolve_config,
            )

        return _BuildPexRequirementsSetup(
            [loaded_lockfile.lockfile_digest], argv, loaded_lockfile.requirement_estimate
        )

    assert isinstance(request.requirements, PexRequirements)
    reqs_info = await get_req_strings(request.requirements)

    # TODO: This is not the best heuristic for available concurrency, since the
    # requirements almost certainly have transitive deps which also need building, but it
    # is better than using something hardcoded.
    concurrency_available = len(reqs_info.req_strings)

    if isinstance(request.requirements.from_superset, Pex):
        repository_pex = request.requirements.from_superset
        return _BuildPexRequirementsSetup(
            [repository_pex.digest],
            [*reqs_info.req_strings, "--pex-repository", repository_pex.name],
            concurrency_available,
        )

    elif isinstance(request.requirements.from_superset, Resolve):
        lockfile = await get_lockfile_for_resolve(
            request.requirements.from_superset, **implicitly()
        )
        loaded_lockfile = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())

        # NB: This is also validated in the constructor.
        assert loaded_lockfile.is_pex_native
        if not reqs_info.req_strings:
            return _BuildPexRequirementsSetup([], [], concurrency_available)

        if loaded_lockfile.metadata:
            validate_metadata(
                loaded_lockfile.metadata,
                request.interpreter_constraints,
                loaded_lockfile.original_lockfile,
                consumed_req_strings=reqs_info.req_strings,
                # Don't validate user requirements when subsetting a resolve, as Pex's
                # validation during the subsetting is far more precise than our naive string
                # comparison. For example, if a lockfile was generated with `foo==1.2.3`
                # and we want to resolve `foo>=1.0.0` or just `foo` out of it, Pex will do
                # so successfully, while our naive validation would fail.
                validate_consumed_req_strings=False,
                python_setup=python_setup,
                resolve_config=resolve_config,
            )

        return _BuildPexRequirementsSetup(
            [loaded_lockfile.lockfile_digest],
            [
                *reqs_info.req_strings,
                "--lock",
                loaded_lockfile.lockfile_path,
                *pex_lock_resolver_args,
            ],
            concurrency_available,
        )

    # We use pip to perform a normal resolve.
    digests = []
    argv = [
        *reqs_info.req_strings,
        *pip_resolver_args,
        *(f"--find-links={find_links}" for find_links in reqs_info.find_links),
    ]
    if request.requirements.constraints_strings:
        constraints_file = "__constraints.txt"
        constraints_content = "\n".join(request.requirements.constraints_strings)
        digests.append(
            await create_digest(
                CreateDigest([FileContent(constraints_file, constraints_content.encode())])
            )
        )
        argv.extend(["--constraints", constraints_file])
    return _BuildPexRequirementsSetup(digests, argv, concurrency_available=concurrency_available)


@rule(level=LogLevel.DEBUG)
async def build_pex(
    request: PexRequest, python_setup: PythonSetup, pex_subsystem: PexSubsystem
) -> BuildPexResult:
    """Returns a PEX with the given settings."""

    if not request.python and not request.interpreter_constraints:
        # Blank ICs in the request means that the caller wants us to use the ICs configured
        # for the resolve (falling back to the global ICs).
        resolve_name = ""
        if isinstance(request.requirements, PexRequirements) and isinstance(
            request.requirements.from_superset, Resolve
        ):
            resolve_name = request.requirements.from_superset.name
        elif isinstance(request.requirements, EntireLockfile):
            resolve_name = request.requirements.lockfile.resolve_name

        if resolve_name:
            request = dataclasses.replace(
                request,
                interpreter_constraints=InterpreterConstraints(
                    python_setup.resolves_to_interpreter_constraints.get(
                        resolve_name,
                        python_setup.interpreter_constraints,
                    )
                ),
            )

    source_dir_name = "source_files"

    pex_python_setup_req = _determine_pex_python_and_platforms(request)
    requirements_setup_req = _setup_pex_requirements(**implicitly({request: PexRequest}))
    sources_digest_as_subdir_req = add_prefix(
        AddPrefix(request.sources or EMPTY_DIGEST, source_dir_name)
    )
    if isinstance(request.requirements, PexRequirements):
        (
            pex_python_setup,
            requirements_setup,
            sources_digest_as_subdir,
            req_info,
        ) = await concurrently(
            pex_python_setup_req,
            requirements_setup_req,
            sources_digest_as_subdir_req,
            get_req_strings(request.requirements),
        )
        req_strings = req_info.req_strings
    else:
        pex_python_setup, requirements_setup, sources_digest_as_subdir = await concurrently(
            pex_python_setup_req,
            requirements_setup_req,
            sources_digest_as_subdir_req,
        )
        req_strings = ()

    argv = [
        "--output-file",
        request.output_filename,
        *request.additional_args,
    ]

    argv.extend(pex_python_setup.argv)

    if request.main is not None:
        argv.extend(request.main.iter_pex_args())
        if isinstance(request.main, Executable):
            # Unlike other MainSpecifiecation types (that can pass spec as-is to pex),
            # Executable must be an actual path relative to the sandbox.
            # request.main.spec is a python source file including its spec_path.
            # To make it relative to the sandbox, we strip the source root
            # and add the source_dir_name (sources get prefixed with that below).
            stripped = await strip_file_name(StrippedFileNameRequest(request.main.spec))
            argv.append(os.path.join(source_dir_name, stripped.value))

    argv.extend(
        f"--inject-args={shlex.quote(injected_arg)}" for injected_arg in request.inject_args
    )
    argv.extend(f"--inject-env={k}={v}" for k, v in sorted(request.inject_env.items()))

    # TODO(John Sirois): Right now any request requirements will shadow corresponding pex path
    #  requirements, which could lead to problems. Support shading python binaries.
    #  See: https://github.com/pantsbuild/pants/issues/9206
    if request.pex_path:
        argv.extend(["--pex-path", ":".join(pex.name for pex in request.pex_path)])

    if request.internal_only:
        # An internal-only runs on a single machine, and pre-installing wheels is wasted work in
        # that case (see https://github.com/pex-tool/pex/issues/2292#issuecomment-1854582647 for
        # analysis).
        argv.append("--no-pre-install-wheels")

    argv.append(f"--sources-directory={source_dir_name}")

    # Include any additional arguments and input digests required by the requirements.
    argv.extend(requirements_setup.argv)

    merged_digest = await merge_digests(
        MergeDigests(
            (
                request.complete_platforms.digest,
                sources_digest_as_subdir,
