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

from __future__ import annotations

import enum
import logging
import os
import re
import tempfile
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from pathlib import Path
from typing import Any, TypeVar, assert_never, cast

from pants.base.build_environment import (
    get_buildroot,
    get_default_pants_config_file,
    get_pants_cachedir,
    pants_version,
)
from pants.base.exceptions import BuildConfigurationError
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.option.custom_types import memory_size
from pants.option.errors import OptionsError
from pants.option.option_types import (
    BoolOption,
    DictOption,
    DirOption,
    EnumOption,
    FloatOption,
    IntOption,
    MemorySizeOption,
    StrListOption,
    StrOption,
)
from pants.option.option_value_container import OptionValueContainer
from pants.option.options import Options
from pants.util.docutil import bin_name, doc_url
from pants.util.logging import LogLevel
from pants.util.osutil import CPU_COUNT
from pants.util.strutil import fmt_memory_size, softwrap
from pants.version import VERSION

logger = logging.getLogger(__name__)


# The time that leases are acquired for in the local store. Configured on the Python side
# in order to ease interaction with the StoreGCService, which needs to be aware of its value.
LOCAL_STORE_LEASE_TIME_SECS = 2 * 60 * 60


MEGABYTES = 1_000_000
GIGABYTES = 1_000 * MEGABYTES


_G = TypeVar("_G", bound="_GlobMatchErrorBehaviorOptionBase")


class RemoteProvider(Enum):
    """Which remote provider to use."""

    reapi = "reapi"
    experimental_file = "experimental-file"
    experimental_github_actions_cache = "experimental-github-actions-cache"

    def _supports_execution(self) -> bool:
        return self is RemoteProvider.reapi

    def _supported_schemes(self) -> list[str]:
        if self is RemoteProvider.reapi:
            return ["grpc", "grpcs"]
        elif self is RemoteProvider.experimental_file:
            return ["file"]
        elif self is RemoteProvider.experimental_github_actions_cache:
            return ["http", "https"]

        assert_never(self)

    def _matches_scheme(self, addr: str) -> bool:
        return any(addr.startswith(f"{scheme}://") for scheme in self._supported_schemes())

    def _human_readable_schemes(self) -> str:
        return ", ".join(f"`{s}://`" for s in sorted(self._supported_schemes()))

    def validate_address(self, addr: str, address_source: str, provider_source: str) -> None:
        if self._matches_scheme(addr):
            # All good! The scheme matches this provider.
            return

        other_providers = [
            provider for provider in RemoteProvider if provider._matches_scheme(addr)
        ]
        if other_providers:
            rendered = ", ".join(f"`{p.value}`" for p in other_providers)
            provider_did_you_mean = (
                f"to use a provider that does support this scheme ({rendered}) or "
            )
        else:
            provider_did_you_mean = ""

        schemes_did_you_mean = (
            f"to use a scheme that is supported by this provider ({self._human_readable_schemes()})"
        )

        raise OptionsError(
            softwrap(
                f"""
                Value `{addr}` from {address_source} is invalid: it doesn't have a scheme that is
                supported by provider `{self.value}` from {provider_source}.

                Did you mean {provider_did_you_mean}{schemes_did_you_mean}?
                """
            )
        )

    def validate_execution_supported(self, provider_source: str, execution_implied_by: str) -> None:
        if self._supports_execution():
            # All good! Execution is supported by this provider.
            return

        supported_execution_providers = ", ".join(
            f"`{provider.value}`" for provider in RemoteProvider if provider._supports_execution()
        )
        raise OptionsError(
            softwrap(
                f"""
                Value `{self.value}` from {provider_source} is invalid: it does not support remote
                execution, but remote execution is required due to {execution_implied_by}.

                Either disable remote execution, or use a provider that does support remote
                execution: {supported_execution_providers}
                """
            )
        )

    @staticmethod
    def provider_help() -> Callable[[object], str]:
        def provider_list_item(p: RemoteProvider) -> str:
            if p is RemoteProvider.reapi:
                description = "a server using the Remote Execution API (https://github.com/bazelbuild/remote-apis)"
            elif p is RemoteProvider.experimental_github_actions_cache:
                description = "the GitHub Actions caching service"
            elif p is RemoteProvider.experimental_file:
                description = "a directory mapped on the current machine"
            else:
                assert_never(p)

            return f"- `{p.value}`: {description} (supported schemes for URIs: {p._human_readable_schemes()})"

        def renderer(_: object) -> str:
            list_items = "\n\n".join(provider_list_item(p) for p in RemoteProvider)
            return softwrap(
                f"""
                The type of provider to use, if using a remote cache and/or remote execution, See
                {doc_url("docs/using-pants/remote-caching-and-execution")} for details.

                Each provider supports different `remote_store_address` and (optional)
                `remote_execution_address` URIs.

                Supported values:

                {list_items}
                """
            )

        return renderer


@dataclass(frozen=True)
class _GlobMatchErrorBehaviorOptionBase:
    """This class exists to have dedicated types per global option of the `GlobMatchErrorBehavior`
    so we can extract the relevant option in a rule to limit the scope of downstream rules to avoid
    depending on the entire global options data."""

    error_behavior: GlobMatchErrorBehavior

    @classmethod
    def ignore(cls: type[_G]) -> _G:
        return cls(GlobMatchErrorBehavior.ignore)

    @classmethod
    def warn(cls: type[_G]) -> _G:
        return cls(GlobMatchErrorBehavior.warn)

    @classmethod
    def error(cls: type[_G]) -> _G:
        return cls(GlobMatchErrorBehavior.error)


class UnmatchedBuildFileGlobs(_GlobMatchErrorBehaviorOptionBase):
    """What to do when globs do not match in BUILD files."""


class UnmatchedCliGlobs(_GlobMatchErrorBehaviorOptionBase):
    """What to do when globs do not match in CLI args."""


class OwnersNotFoundBehavior(_GlobMatchErrorBehaviorOptionBase):
    """What to do when a file argument cannot be mapped to an owning target."""


@enum.unique
class RemoteCacheWarningsBehavior(Enum):
    ignore = "ignore"
    first_only = "first_only"
    backoff = "backoff"
    always = "always"


@enum.unique
class CacheContentBehavior(Enum):
    fetch = "fetch"
    validate = "validate"
    defer = "defer"


@enum.unique
class AuthPluginState(Enum):
    OK = "ok"
    UNAVAILABLE = "unavailable"


@dataclass(frozen=True)
class AuthPluginResult:
    """The return type for a function specified via `[GLOBAL].remote_auth_plugin`.

    The returned `store_headers` and `execution_headers` will replace whatever headers Pants would
    have used normally, e.g. what is set with `[GLOBAL].remote_store_headers`. This allows you to control
    the merge strategy if your plugin sets conflicting headers. Usually, you will want to preserve
    the `initial_store_headers` and `initial_execution_headers` passed to the plugin.

    If set, the returned `instance_name` will override `[GLOBAL].remote_instance_name`,
    `store_address` will override `[GLOBAL].remote_store_address`, and `execution_address` will
    override ``[GLOBAL].remote_execution_address``. The addresses are interpreted and validated in
    the same manner as the corresponding option.
    """

    state: AuthPluginState
    store_headers: dict[str, str]
    execution_headers: dict[str, str]
    provider: RemoteProvider = RemoteProvider.reapi
    store_address: str | None = None
    execution_address: str | None = None
    instance_name: str | None = None
    expiration: datetime | None = None
    plugin_name: str | None = None

    def __post_init__(self) -> None:
        name = self.plugin_name or ""
        plugin_context = f"in `AuthPluginResult` returned from `[GLOBAL].remote_auth_plugin` {name}"

        if self.store_address:
            self.provider.validate_address(
                self.store_address,
                address_source=f"`store_address` {plugin_context}",
                provider_source="`provider` in same result",
            )

        if self.execution_address:
            self.provider.validate_execution_supported(
                provider_source=f"`provider` {plugin_context}",
                execution_implied_by="`execution_address` in same result",
            )
            self.provider.validate_address(
                self.execution_address,
                address_source=f"`execution_address` {plugin_context}",
                provider_source="`provider` in same result",
            )

    @property
    def is_available(self) -> bool:
        return self.state == AuthPluginState.OK


@dataclass(frozen=True)
class DynamicRemoteOptions:
    """Options related to remote execution of processes which are computed dynamically."""

    provider: RemoteProvider
    execution: bool
    cache_read: bool
    cache_write: bool
    instance_name: str | None
    store_address: str | None
    execution_address: str | None
    store_headers: dict[str, str]
    execution_headers: dict[str, str]
    parallelism: int
    store_rpc_concurrency: int
    store_batch_load_enabled: bool
    cache_rpc_concurrency: int
    execution_rpc_concurrency: int

    def _validate_store_addr(self) -> None:
        if self.store_address:
            return
        if self.cache_read:
            raise OptionsError(
                softwrap(
                    """
                    The `[GLOBAL].remote_cache_read` option requires also setting the
                    `[GLOBAL].remote_store_address` option in order to work properly.
                    """
                )
            )
        if self.cache_write:
            raise OptionsError(
                softwrap(
                    """
                    The `[GLOBAL].remote_cache_write` option requires also setting the
                    `[GLOBAL].remote_store_address` option in order to work properly.
                    """
                )
            )

    def _validate_exec_addr(self) -> None:
        if not self.execution:
            return
        if not self.execution_address:
            raise OptionsError(
                softwrap(
                    """
                    The `[GLOBAL].remote_execution` option requires also setting the
                    `[GLOBAL].remote_execution_address` option in order to work properly.
                    """
                )
            )
        if not self.store_address:
            raise OptionsError(
                softwrap(
                    """
                    The `[GLOBAL].remote_execution_address` option requires also setting the
                    `[GLOBAL].remote_store_address` option. Often these have the same value.
                    """
                )
            )

    def __post_init__(self) -> None:
        self._validate_store_addr()
        self._validate_exec_addr()

    @classmethod
    def disabled(cls) -> DynamicRemoteOptions:
        return cls(
            provider=DEFAULT_EXECUTION_OPTIONS.remote_provider,
            execution=False,
            cache_read=False,
            cache_write=False,
            instance_name=None,
            store_address=None,
            execution_address=None,
            store_headers={},
            execution_headers={},
            parallelism=DEFAULT_EXECUTION_OPTIONS.process_execution_remote_parallelism,
            store_rpc_concurrency=DEFAULT_EXECUTION_OPTIONS.remote_store_rpc_concurrency,
            store_batch_load_enabled=DEFAULT_EXECUTION_OPTIONS.remote_store_batch_load_enabled,
            cache_rpc_concurrency=DEFAULT_EXECUTION_OPTIONS.remote_cache_rpc_concurrency,
            execution_rpc_concurrency=DEFAULT_EXECUTION_OPTIONS.remote_execution_rpc_concurrency,
        )

    @classmethod
    def _use_oauth_token(cls, bootstrap_options: OptionValueContainer) -> DynamicRemoteOptions:
        oauth_token = bootstrap_options.remote_oauth_bearer_token

        if set(oauth_token).intersection({"\n", "\r"}):
            raise OptionsError(
                "OAuth bearer token from `remote_oauth_bearer_token` option must not contain multiple lines."
            )

        token_header = {"authorization": f"Bearer {oauth_token}"}
        provider = cast(RemoteProvider, bootstrap_options.remote_provider)
        execution = cast(bool, bootstrap_options.remote_execution)
        cache_read = cast(bool, bootstrap_options.remote_cache_read)
        cache_write = cast(bool, bootstrap_options.remote_cache_write)
        store_address = cast("str | None", bootstrap_options.remote_store_address)
        execution_address = cast("str | None", bootstrap_options.remote_execution_address)
        instance_name = cast("str | None", bootstrap_options.remote_instance_name)
        execution_headers = cast("dict[str, str]", bootstrap_options.remote_execution_headers)
        store_headers = cast("dict[str, str]", bootstrap_options.remote_store_headers)
        parallelism = cast(int, bootstrap_options.process_execution_remote_parallelism)
        store_rpc_concurrency = cast(int, bootstrap_options.remote_store_rpc_concurrency)
        store_batch_load_enabled = cast(bool, bootstrap_options.remote_store_batch_load_enabled)
        cache_rpc_concurrency = cast(int, bootstrap_options.remote_cache_rpc_concurrency)
        execution_rpc_concurrency = cast(int, bootstrap_options.remote_execution_rpc_concurrency)
        execution_headers.update(token_header)
        store_headers.update(token_header)
        return cls(
            provider=provider,
            execution=execution,
            cache_read=cache_read,
            cache_write=cache_write,
            instance_name=instance_name,
            store_address=cls._normalize_address(store_address),
            execution_address=cls._normalize_address(execution_address),
            store_headers=store_headers,
            execution_headers=execution_headers,
            parallelism=parallelism,
            store_rpc_concurrency=store_rpc_concurrency,
            store_batch_load_enabled=store_batch_load_enabled,
            cache_rpc_concurrency=cache_rpc_concurrency,
            execution_rpc_concurrency=execution_rpc_concurrency,
        )

    @classmethod
    def from_options(
        cls,
        full_options: Options,
        env: CompleteEnvironmentVars,
        prior_result: AuthPluginResult | None = None,
        remote_auth_plugin_func: Callable | None = None,
    ) -> tuple[DynamicRemoteOptions, AuthPluginResult | None]:
        global_options = full_options.for_global_scope()
        execution = cast(bool, global_options.remote_execution)
        cache_read = cast(bool, global_options.remote_cache_read)
        cache_write = cast(bool, global_options.remote_cache_write)
        if not (execution or cache_read or cache_write):
            return cls.disabled(), None

        sources = {
            str(remote_auth_plugin_func): bool(remote_auth_plugin_func),
            "[GLOBAL].remote_oauth_bearer_token": bool(global_options.remote_oauth_bearer_token),
        }
        enabled_sources = [name for name, enabled in sources.items() if enabled]
        if len(enabled_sources) > 1:
            rendered = ", ".join(f"`{name}`" for name in enabled_sources)
            raise OptionsError(
                softwrap(
                    f"""
                    Multiple options are set that provide auth information: {rendered}.
                    This is not supported. Only one of those should be set.
                    """
                )
            )
        if global_options.remote_oauth_bearer_token:
            return cls._use_oauth_token(global_options), None
        if remote_auth_plugin_func is not None:
            return cls._use_auth_plugin(
                global_options,
                full_options=full_options,
                env=env,
                prior_result=prior_result,
                remote_auth_plugin_func=remote_auth_plugin_func,
            )
        return cls._use_no_auth(global_options), None

    @classmethod
    def _use_no_auth(cls, bootstrap_options: OptionValueContainer) -> DynamicRemoteOptions:
        provider = cast(RemoteProvider, bootstrap_options.remote_provider)
        execution = cast(bool, bootstrap_options.remote_execution)
        cache_read = cast(bool, bootstrap_options.remote_cache_read)
        cache_write = cast(bool, bootstrap_options.remote_cache_write)
        store_address = cast("str | None", bootstrap_options.remote_store_address)
        execution_address = cast("str | None", bootstrap_options.remote_execution_address)
        instance_name = cast("str | None", bootstrap_options.remote_instance_name)
        execution_headers = cast("dict[str, str]", bootstrap_options.remote_execution_headers)
        store_headers = cast("dict[str, str]", bootstrap_options.remote_store_headers)
        parallelism = cast(int, bootstrap_options.process_execution_remote_parallelism)
        store_rpc_concurrency = cast(int, bootstrap_options.remote_store_rpc_concurrency)
        store_batch_load_enabled = cast(bool, bootstrap_options.remote_store_batch_load_enabled)
        cache_rpc_concurrency = cast(int, bootstrap_options.remote_cache_rpc_concurrency)
        execution_rpc_concurrency = cast(int, bootstrap_options.remote_execution_rpc_concurrency)
        return cls(
            provider=provider,
            execution=execution,
            cache_read=cache_read,
            cache_write=cache_write,
            instance_name=instance_name,
            store_address=cls._normalize_address(store_address),
            execution_address=cls._normalize_address(execution_address),
            store_headers=store_headers,
            execution_headers=execution_headers,
            parallelism=parallelism,
            store_rpc_concurrency=store_rpc_concurrency,
            store_batch_load_enabled=store_batch_load_enabled,
            cache_rpc_concurrency=cache_rpc_concurrency,
            execution_rpc_concurrency=execution_rpc_concurrency,
        )

    @classmethod
    def _use_auth_plugin(
        cls,
        bootstrap_options: OptionValueContainer,
        full_options: Options,
        env: CompleteEnvironmentVars,
        prior_result: AuthPluginResult | None,
        remote_auth_plugin_func: Callable,
    ) -> tuple[DynamicRemoteOptions, AuthPluginResult | None]:
        provider = cast(RemoteProvider, bootstrap_options.remote_provider)
        execution = cast(bool, bootstrap_options.remote_execution)
        cache_read = cast(bool, bootstrap_options.remote_cache_read)
        cache_write = cast(bool, bootstrap_options.remote_cache_write)
        store_address = cast("str | None", bootstrap_options.remote_store_address)
        execution_address = cast("str | None", bootstrap_options.remote_execution_address)
        instance_name = cast("str | None", bootstrap_options.remote_instance_name)
        execution_headers = cast("dict[str, str]", bootstrap_options.remote_execution_headers)
        store_headers = cast("dict[str, str]", bootstrap_options.remote_store_headers)
        parallelism = cast(int, bootstrap_options.process_execution_remote_parallelism)
        store_rpc_concurrency = cast(int, bootstrap_options.remote_store_rpc_concurrency)
        store_batch_load_enabled = cast(bool, bootstrap_options.remote_store_batch_load_enabled)
        cache_rpc_concurrency = cast(int, bootstrap_options.remote_cache_rpc_concurrency)
        execution_rpc_concurrency = cast(int, bootstrap_options.remote_execution_rpc_concurrency)
        auth_plugin_result = cast(
            AuthPluginResult,
            remote_auth_plugin_func(
                initial_execution_headers=execution_headers,
                initial_store_headers=store_headers,
                options=full_options,
                env=dict(env),
                prior_result=prior_result,
            ),
        )
        plugin_name = (
            auth_plugin_result.plugin_name
            or f"{remote_auth_plugin_func.__module__}.{remote_auth_plugin_func.__name__}"
        )
        if not auth_plugin_result.is_available:
            # NB: This is debug because we expect plugins to log more informative messages.
            logger.debug(
                f"Disabling remote caching and remote execution because authentication was not available via the plugin {plugin_name} (from `[GLOBAL].remote_auth_plugin`)."
            )
            return cls.disabled(), None

        logger.debug(
            f"Remote auth plugin `{plugin_name}` succeeded. Remote caching/execution will be attempted."
        )
        provider = auth_plugin_result.provider
        execution_headers = auth_plugin_result.execution_headers
        store_headers = auth_plugin_result.store_headers
        plugin_provided_opt_log = "Setting `[GLOBAL].remote_{opt}` is not needed and will be ignored since it is provided by the auth plugin: {plugin_name}."
        if auth_plugin_result.instance_name is not None:
            if instance_name is not None:
                logger.warning(
                    plugin_provided_opt_log.format(opt="instance_name", plugin_name=plugin_name)
                )
            instance_name = auth_plugin_result.instance_name
        if auth_plugin_result.store_address is not None:
            if store_address is not None:
                logger.warning(
                    plugin_provided_opt_log.format(opt="store_address", plugin_name=plugin_name)
                )
            store_address = auth_plugin_result.store_address
        if auth_plugin_result.execution_address is not None:
            if execution_address is not None:
                logger.warning(
                    plugin_provided_opt_log.format(opt="execution_address", plugin_name=plugin_name)
                )
            execution_address = auth_plugin_result.execution_address

        opts = cls(
            provider=provider,
            execution=execution,
            cache_read=cache_read,
            cache_write=cache_write,
            instance_name=instance_name,
            store_address=cls._normalize_address(store_address),
            execution_address=cls._normalize_address(execution_address),
            store_headers=store_headers,
            execution_headers=execution_headers,
            parallelism=parallelism,
            store_rpc_concurrency=store_rpc_concurrency,
            store_batch_load_enabled=store_batch_load_enabled,
            cache_rpc_concurrency=cache_rpc_concurrency,
            execution_rpc_concurrency=execution_rpc_concurrency,
        )
        return opts, auth_plugin_result

    @classmethod
    def _normalize_address(cls, address: str | None) -> str | None:
        # NB: Tonic expects the schemes `http` and `https`, even though they are gRPC requests.
        # We validate that users set `grpc` and `grpcs` in the options system / plugin for clarity,
        # but then normalize to `http`/`https`.
        # TODO: move this logic into the actual remote providers
        return re.sub(r"^grpc", "http", address) if address else None


@dataclass(frozen=True)
class ExecutionOptions:
    """A collection of all options related to (remote) execution of processes.

    TODO: These options should move to a Subsystem once we add support for "bootstrap" Subsystems (ie,
    allowing Subsystems to be consumed before the Scheduler has been created).
    """

    remote_provider: RemoteProvider

    remote_execution: bool
    remote_cache_read: bool
    remote_cache_write: bool

    remote_instance_name: str | None
    remote_ca_certs_path: str | None
    remote_client_certs_path: str | None
    remote_client_key_path: str | None

    use_sandboxer: bool
    local_cache: bool
    process_execution_local_parallelism: int
    process_execution_local_enable_nailgun: bool
    process_execution_remote_parallelism: int
    process_execution_cache_namespace: str | None
    process_execution_graceful_shutdown_timeout: int
    cache_content_behavior: CacheContentBehavior

    process_total_child_memory_usage: int | None
    process_per_child_memory_usage: int

    remote_store_address: str | None
    remote_store_headers: dict[str, str]
    remote_store_chunk_bytes: Any
    remote_store_rpc_retries: int
    remote_store_rpc_concurrency: int
    remote_store_batch_api_size_limit: int
    remote_store_batch_load_enabled: bool
    remote_store_rpc_timeout_millis: int

    remote_cache_warnings: RemoteCacheWarningsBehavior
    remote_cache_rpc_concurrency: int
    remote_cache_rpc_timeout_millis: int

    remote_execution_address: str | None
    remote_execution_headers: dict[str, str]
    remote_execution_overall_deadline_secs: int
    remote_execution_rpc_concurrency: int

    remote_execution_append_only_caches_base_path: str | None

    @classmethod
    def from_options(
        cls,
        bootstrap_options: OptionValueContainer,
        dynamic_remote_options: DynamicRemoteOptions,
    ) -> ExecutionOptions:
        return cls(
            remote_provider=dynamic_remote_options.provider,
            # Remote execution strategy.
            remote_execution=dynamic_remote_options.execution,
            remote_cache_read=dynamic_remote_options.cache_read,
            remote_cache_write=dynamic_remote_options.cache_write,
            # General remote setup.
            remote_instance_name=dynamic_remote_options.instance_name,
            remote_ca_certs_path=bootstrap_options.remote_ca_certs_path,
            remote_client_certs_path=bootstrap_options.remote_client_certs_path,
            remote_client_key_path=bootstrap_options.remote_client_key_path,
            # Process execution setup.
            use_sandboxer=bootstrap_options.sandboxer,
            local_cache=bootstrap_options.local_cache,
            process_execution_local_parallelism=bootstrap_options.process_execution_local_parallelism,
            process_execution_remote_parallelism=dynamic_remote_options.parallelism,
            process_execution_cache_namespace=bootstrap_options.process_execution_cache_namespace,
            process_execution_graceful_shutdown_timeout=bootstrap_options.process_execution_graceful_shutdown_timeout,
            process_execution_local_enable_nailgun=bootstrap_options.process_execution_local_enable_nailgun,
            cache_content_behavior=bootstrap_options.cache_content_behavior,
            process_total_child_memory_usage=bootstrap_options.process_total_child_memory_usage,
            process_per_child_memory_usage=bootstrap_options.process_per_child_memory_usage,
            # Remote store setup.
            remote_store_address=dynamic_remote_options.store_address,
            remote_store_headers=cls.with_user_agent(dynamic_remote_options.store_headers),
            remote_store_chunk_bytes=bootstrap_options.remote_store_chunk_bytes,
            remote_store_rpc_retries=bootstrap_options.remote_store_rpc_retries,
            remote_store_rpc_concurrency=dynamic_remote_options.store_rpc_concurrency,
            remote_store_batch_api_size_limit=bootstrap_options.remote_store_batch_api_size_limit,
            remote_store_batch_load_enabled=bootstrap_options.remote_store_batch_load_enabled,
            remote_store_rpc_timeout_millis=bootstrap_options.remote_store_rpc_timeout_millis,
            # Remote cache setup.
            remote_cache_warnings=bootstrap_options.remote_cache_warnings,
            remote_cache_rpc_concurrency=dynamic_remote_options.cache_rpc_concurrency,
            remote_cache_rpc_timeout_millis=bootstrap_options.remote_cache_rpc_timeout_millis,
            # Remote execution setup.
            remote_execution_address=dynamic_remote_options.execution_address,
            remote_execution_headers=cls.with_user_agent(dynamic_remote_options.execution_headers),
            remote_execution_overall_deadline_secs=bootstrap_options.remote_execution_overall_deadline_secs,
            remote_execution_rpc_concurrency=dynamic_remote_options.execution_rpc_concurrency,
            remote_execution_append_only_caches_base_path=bootstrap_options.remote_execution_append_only_caches_base_path,
        )

    @classmethod
    def with_user_agent(cls, headers: dict[str, str]) -> dict[str, str]:
        has_user_agent = any(k.lower() == "user-agent" for k in headers.keys())
        if has_user_agent:
            return headers
        return {"user-agent": f"pants/{VERSION}"} | headers


@dataclass(frozen=True)
class LocalStoreOptions:
    """A collection of all options related to the local store.

    TODO: These options should move to a Subsystem once we add support for "bootstrap" Subsystems (ie,
    allowing Subsystems to be consumed before the Scheduler has been created).
    """

    store_dir: str = os.path.join(get_pants_cachedir(), "lmdb_store")
    processes_max_size_bytes: int = 16 * GIGABYTES
    files_max_size_bytes: int = 256 * GIGABYTES
    directories_max_size_bytes: int = 16 * GIGABYTES
    shard_count: int = 16

    def target_total_size_bytes(self) -> int:
        """Returns the target total size of all of the stores.

        The `max_size` values are caps on the total size of each store: the "target" size
        is the size that garbage collection will attempt to shrink the stores to each time
        it runs.

        NB: This value is not currently configurable, but that could be desirable in the future.
        """
        max_total_size_bytes = (
            self.processes_max_size_bytes
            + self.files_max_size_bytes
            + self.directories_max_size_bytes
        )
        return max_total_size_bytes // 10

    @classmethod
    def from_options(cls, options: OptionValueContainer) -> LocalStoreOptions:
        return cls(
            store_dir=str(Path(options.local_store_dir).resolve()),
            processes_max_size_bytes=options.local_store_processes_max_size_bytes,
            files_max_size_bytes=options.local_store_files_max_size_bytes,
            directories_max_size_bytes=options.local_store_directories_max_size_bytes,
            shard_count=options.local_store_shard_count,
        )


_PER_CHILD_MEMORY_USAGE = "512MiB"


DEFAULT_EXECUTION_OPTIONS = ExecutionOptions(
    remote_provider=RemoteProvider.reapi,
    # Remote execution strategy.
    remote_execution=False,
    remote_cache_read=False,
    remote_cache_write=False,
    # General remote setup.
    remote_instance_name=None,
    remote_ca_certs_path=None,
    remote_client_certs_path=None,
    remote_client_key_path=None,
    # Process execution setup.
    process_total_child_memory_usage=None,
    process_per_child_memory_usage=memory_size(_PER_CHILD_MEMORY_USAGE),
    process_execution_local_parallelism=CPU_COUNT,
    process_execution_remote_parallelism=128,
    process_execution_cache_namespace=None,
    use_sandboxer=False,
    local_cache=True,
    cache_content_behavior=CacheContentBehavior.fetch,
    process_execution_local_enable_nailgun=True,
    process_execution_graceful_shutdown_timeout=3,
    # Remote store setup.
    remote_store_address=None,
    remote_store_headers={},
    remote_store_chunk_bytes=1024 * 1024,
    remote_store_rpc_retries=2,
    remote_store_rpc_concurrency=128,
    remote_store_batch_api_size_limit=4194304,
    remote_store_rpc_timeout_millis=30000,
    remote_store_batch_load_enabled=False,
    # Remote cache setup.
    remote_cache_warnings=RemoteCacheWarningsBehavior.backoff,
    remote_cache_rpc_concurrency=128,
    remote_cache_rpc_timeout_millis=1500,
    # Remote execution setup.
    remote_execution_address=None,
    remote_execution_headers={},
    remote_execution_overall_deadline_secs=60 * 60,  # one hour
    remote_execution_rpc_concurrency=128,
