#!/usr/bin/env python3
# Copyright 2026 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"Implement the running of a Bazel command from Ninja and copying the outputs back into the ninja outdir."

import contextlib
import dataclasses
import json
import os
import shlex
import shutil
import sys
import typing as T
from pathlib import Path

_SCRIPT_DIR = os.path.dirname(__file__)
sys.path.insert(0, _SCRIPT_DIR)
# LINT.IfChange(imports)
import bazel_action_utils
import bazel_label_mapper
import bazel_rust_analyzer_utils
import bazel_source_path_mapper
import build_utils
import stdio_redirection
import thread_pool_helpers
from bazel_action_file_copy_utils import (
    FilePath,
    check_if_need_to_copy_file,
    copy_directory_if_changed,
    hardlink_or_copy_writable,
    write_file_if_changed,
)

_BUILD_API_DIR = os.path.join(_SCRIPT_DIR, "../../api")
sys.path.insert(0, _BUILD_API_DIR)
from debug_symbols import (
    DebugSymbolEntryType,
    DebugSymbolExporter,
    DebugSymbolsManifestParser,
)

# LINT.ThenChange(//build/bazel/bazel_action.gni:bazel_action_impl_imports)


# Set this to True to debug operations locally in this script.
# IMPORTANT: Setting this to True will result in Ninja timeouts in CQ
# due to the stdout/stderr logs being too large.
_DEBUG = False

# Set this to True to debug the bazel query cache.
_DEBUG_BAZEL_QUERIES = _DEBUG

# Set this to True to debug the export of debug symbols.
_DEBUG_SYMBOL_EXPORT = _DEBUG


def debug(msg: str) -> None:
    # Print debug message to stderr if _DEBUG is True.
    if _DEBUG:
        print("BAZEL_ACTION_DEBUG: " + msg, file=sys.stderr)


# The prefix that appears in Bazel stderr DEBUG lines, generated by
# the debug symbol aspect.
# LINT.IfChange(debug_symbols_manifest_prefix)
_BAZEL_DEBUG_SYMBOLS_MANIFEST_PATH_PREFIX = b"DEBUG_SYMBOLS_MANIFEST_PATH="
# LINT.ThenChange(//build/bazel/debug_symbols/aspects.bzl:debug_symbols_manifest_prefix)

# The prefix that appears in Bazel stderr DEBUG lines, generated by
# the source files list aspect.
# LINT.IfChange(source_files_list_path_prefix)
_BAZEL_SOURCE_FILES_MANIFEST_PATH_PREFIX = b"FUCHSIA_SOURCES_MANIFEST_PATH="
# LINT.ThenChange(//build/bazel/aspects/source_files.bzl:source_files_list_path_prefix)

# Directory where to find Starlark input files.
_STARLARK_DIR = os.path.join(os.path.dirname(_SCRIPT_DIR), "starlark")


class BazelActionError(Exception):
    """A Bazel action error was encountered."""


class BazelActionScriptError(BazelActionError):
    """Errors detected by the Bazel action script, not Bazel itself."""


class SourceFilesNotFoundError(BazelActionScriptError):
    """No source files found for a target."""


class FileOutputConfigError(BazelActionScriptError):
    """Invalid configuration for file outputs."""


class DirectoryOutputConfigError(BazelActionScriptError):
    """Invalid configuration for directory outputs."""


class MissingOutputDirectoryError(BazelActionScriptError):
    """Missing output directory after build."""


class MissingTrackedFileError(BazelActionScriptError):
    """Missing tracked file inside output directory."""


@dataclasses.dataclass
class _InputFileGenQueryInfo(object):
    """Class for holding info about the genqueries generated for each target."""

    genquery_target_label: str
    genquery_output_path: Path


@dataclasses.dataclass
class BazelActionOutputs(object):
    """The outputs to copy from Bazel to GN for a given Bazel action."""

    files: list[bazel_action_utils.FileOutput]
    directories: list[bazel_action_utils.DirectoryOutput]
    packages: list[bazel_action_utils.PackageOutput]
    final_symlinks: list[bazel_action_utils.FinalSymlinkOutput]

    def __len__(self) -> int:
        return (
            len(self.files)
            + len(self.directories)
            + len(self.packages)
            + len(self.final_symlinks)
        )


@dataclasses.dataclass
class BazelExtraOutputs(object):
    """A collection of optional outputs to write as debugging aids, and the paths to write them at.

    Fields:

        build_event_json_file: A build event json file from Bazel

        command_file: The Bazel command itself

        command_profile: A command profile tgz file

        debug_symbols_manifest: A manifest of debug symbols

        explain_file: A file which explains why Bazel re-ran each action
    """

    build_event_json_file: Path | None = None
    command_file: Path | None = None
    command_profile: Path | None = None
    debug_symbols_manifest: Path | None = None
    explain_file: Path | None = None


@dataclasses.dataclass
class BazelActionResult(object):
    """Wrapper for results of executing a bazel action.

    Fields:

        configured_args:  A list of args usable for bazel queries in the same configuration.

        output_files:  A list of the output paths that were actually copied to (updated)
          by this action.

        source_files:  A dict of lists of the Bazel build files and input source files
            that were used by each of the built targets, as standard filesystem paths,
            keyed by the bazel target.

        rust_project_crates: A list of CrateSpec dictionaries describing the crates of the
            the build targets and their dependencies.
    """

    configured_args: list[str]
    output_files: list[Path]
    source_files: dict[str, list[str]]
    rust_crates: list[bazel_rust_analyzer_utils.Crate]


class BazelActionRunner(object):
    """Used to run Bazel actions and copy all the outputs back to the Ninja outdir.

    This class is configured based on the global context of the build (both static
    args set via GN metadata and build-specific environment variables).  It can
    then be used to run multiple Bazel actions in sequence, with that global
    configuration.

    See the `run()` method for per-action calling.

    """

    def __init__(
        self,
        bazel_paths: build_utils.BazelPaths,
        global_args: bazel_action_utils.BazelGlobalArguments,
        query_cache: build_utils.BazelQueryCache,
    ) -> None:
        self.paths = bazel_paths
        self.global_args = global_args
        self.launcher = build_utils.BazelLauncher(
            bazel_paths.launcher,
            log_err=lambda msg: (
                print(f"BAZEL_ACTION_ERROR: {msg}", file=sys.stderr)
                if _DEBUG
                else None
            ),
        )
        self.rbe_settings = (
            bazel_action_utils.BazelRbeSettings.create_from_build_dir(
                bazel_paths.ninja_build_dir
            )
        )
        self.query_cache = query_cache

    def run(
        self,
        command: str,
        platform_config: str,
        platform_label: str,
        targets: list[str],
        outputs: BazelActionOutputs,
        extra_outputs: BazelExtraOutputs,
        time_profile: build_utils.TimeProfile,
    ) -> BazelActionResult:
        """Run a bazel command.

        Args:

            targets: The list of bazel target labels to build/test/etc.

            extra_args: (temporary) list of bazel cli args to append to the command

        Returns:
            (temporary) A list of debug symbol manifest paths

        Raises:
            BazelActionError on any failure.
        """
        time_profile.start(
            "construct_cmd_args",
            "Constructing the command-line arguments for Bazel",
        )

        # Calculate the platform configuration needed for the build and subsequent queries.
        configured_args = calculate_platform_config_args(
            platform_config,
            platform_label,
            self.rbe_settings,
        )

        # These are the args that are listed after the command, targets, and configured
        # args.
        cmd_args = []

        # If a build event json file is requested, tell Bazel to create one.
        if extra_outputs.build_event_json_file:
            # Create parent directory to avoid Bazel complaining it cannot
            # write the events log file.
            extra_outputs.build_event_json_file.parent.mkdir(
                parents=True, exist_ok=True
            )
            cmd_args += [
                "--build_event_json_file",
                str(extra_outputs.build_event_json_file.resolve()),
            ]

        # If an explain file is requested, tell Bazel to create one.
        if extra_outputs.explain_file:
            cmd_args += [
                "--explain",
                self.paths.ninja_build_dir / extra_outputs.explain_file,
            ]

        # If a command profile is requested, tell Bazel to create one.
        if extra_outputs.command_profile:
            cmd_args += [
                "--profile",
                self.paths.ninja_build_dir / extra_outputs.command_profile,
            ]

        # if build-event uploading is enabled in global args, then append the config for that
        if self.global_args.upload_build_events:
            cmd_args += [f"--config={self.global_args.upload_build_events}"]

        jobs = calculate_jobs_param(self.rbe_settings)
        if jobs:
            cmd_args += [jobs]

        if self.global_args.quiet:
            cmd_args += ["--config=quiet"]
        else:
            # Normal builds always need to print INFO level messages so that
            # warnings can be emitted from bazel ctx.run_shell() commands,
            # otherwise any stderr/stdout warnings from tools are eaten by
            # the build.
            #
            #  Known uses:
            #     - size checker outputs
            #     - developer overrides warnings for assembly
            #
            cmd_args += ["--config=verbose"]

        cmd_args += [
            # Ensure that all debug symbols are properly generated during the build
            # The aspect will also generate extra manifests whose paths will be
            # printed to Bazel's stderr DEBUG lines, which will be filtered
            # below to retrieve them (and hide the output from the user).
            #
            # Doing so avoids doing an extra `bazel build` command to get the
            # same data. See https://fxbug.dev/452591388
            "--output_groups=+debug_symbol_files",
            "--aspects=//build/bazel/debug_symbols:aspects.bzl%generate_manifest",
        ]

        cmd_args += [
            # Ensure rust-analyzer manifests are generated during the build.
            "--output_groups={}".format(
                ",".join(
                    f"+{group}"
                    for group in bazel_rust_analyzer_utils.RUST_ANALYZER_OUTPUT_GROUPS
                )
            ),
            "--aspects=//build/bazel/aspects:rust_analyzer.bzl%generate_rust_analyzer_manifest",
        ]

        cmd_args += [
            # Ensure source file lists are generated during the build to avoid doing one slow
            # query for each top-level Bazel target.
            "--output_groups=+fuchsia_sources_manifest",
            "--aspects=//build/bazel/aspects:source_files.bzl%generate_source_files_manifest",
        ]

        # Add --sandbox_debug if enabled in the build environment.
        if self.global_args.sandbox_debug:
            cmd_args += ["--sandbox_debug"]

        # Always enable verbose failures
        cmd_args += ["--verbose_failures"]

        # To capture the set of dependencies for targets, a genquery that can find
        # all the BUILD.bazel and .bzl files needed for each target.
        #
        # The build file must be deleted after each build command.
        time_profile.start(
            "input_files_genquery", "Generating buildfiles_genquery/BUILD.bazel"
        )
        (
            input_files_genqueries,
            genquery_build_file,
        ) = self._create_buildfiles_genqueries(targets)

        # Add the genquery target to the list of targets and add those to the
        # command line for bazel. IMPORTANT: The aspects listed above will also apply to
        # these targets and generate corresponding files in the buildfiles_genquery/ directory.
        cmd_args += targets
        cmd_args += [
            genquery.genquery_target_label
            for genquery in input_files_genqueries.values()
        ]

        # Construct the entire command (this is done here, as a single block, for
        # clarity on how the various lists above are brought together to make the
        # command line that we'll use with Bazel).
        cmd: list[str] = [command]
        cmd += configured_args
        cmd += cmd_args

        # Now that there's a complete command string calculated, print it to debug or the command
        # file output if we have one.
        if _DEBUG:
            # This is all arguments on one line, so that they can be run via cut/paste.
            debug(
                "BUILD_CMD: "
                + build_utils.cmd_args_to_string([self.paths.launcher] + cmd)
            )
        if extra_outputs.command_file:
            # This file is one argument per line.
            write_file_if_changed(
                extra_outputs.command_file,
                " \\\n  ".join(
                    shlex.quote(str(c)) for c in [self.paths.launcher] + cmd
                )
                + "\n",
            )

        with contextlib.ExitStack() as on_exit:
            # Remove all files in buildfiles_genquery/ after the action completes, to make sure
            # that ninja doesn't see them as files that can cause consistency or non-convergence
            # issues.
            on_exit.callback(
                shutil.rmtree,
                self.paths.workspace / "bazel-bin/buildfiles_genquery",
                ignore_errors=True,
            )

            aspect_prefix_map = {
                "debug_symbol_manifest_paths": _BAZEL_DEBUG_SYMBOLS_MANIFEST_PATH_PREFIX,
                "rust_analyzer_manifest_paths": bazel_rust_analyzer_utils.FUCHSIA_RUST_ANALYZER_MANIFEST_PATH_PREFIX,
                "source_files_manifest_paths": _BAZEL_SOURCE_FILES_MANIFEST_PATH_PREFIX,
            }

            aspect_recorded_map = self._invoke_bazel_and_return_aspect_outputs(
                targets,
                cmd,
                aspect_prefix_map,
                time_profile,
            )

            def get_recorded_aspect_paths(key_name: str) -> list[str]:
                """Return the recorded manifest paths recorded by aspects, filtering the genquery ones.

                Args:
                    key_name: One of the aspect_prefix_map keys.
                Returns:
                    A list of manifest paths, relative to the Bazel execroot.
                """
                return [
                    path
                    for path in aspect_recorded_map[key_name]
                    if not "buildfiles_genquery" in path
                ]

            debug_symbol_manifest_paths = get_recorded_aspect_paths(
                "debug_symbol_manifest_paths"
            )
            rust_manifest_paths = get_recorded_aspect_paths(
                "rust_analyzer_manifest_paths"
            )

            source_files_manifest_paths = aspect_recorded_map[
                "source_files_manifest_paths"
            ]

            input_files = {}
            if command == "build":
                time_profile.start(
                    "read_genquery_outputs",
                    "Read the results from the genqueries",
                )

                if not self.global_args.quiet:
                    print(
                        f"Gathering input paths from {len(targets)} targets for ninja depfile generation..."
                    )

                input_files = self._parse_buildfiles_genquery_results_and_source_files_manifests(
                    input_files_genqueries,
                    source_files_manifest_paths,
                    time_profile,
                )

        if command == "build" and outputs.packages:
            # If we have packages, query to get the paths to the package archives,
            # as they need to be copied to the Ninja outdir along with any other files.
            time_profile.start(
                "package_info",
                "Run cquery to extract Fuchsia package archiveinformation",
            )
            package_archive_files = self.query_for_package_archives(
                configured_args, outputs.packages
            )
            outputs.files.extend(package_archive_files)

        # Now copy all the outputs.  This is a separate function to help break up the run()
        # functionality for better clarity.
        output_copier = _BazelOutputCopier(self.paths)
        if not self.global_args.quiet:
            print(f"Copying {len(outputs)} outputs from Bazel...")
        all_output_files = output_copier.copy(outputs, time_profile)

        # Perform the merging of debug symbols data, and optionally copy
        # and write the output to a manifest.
        need_to_copy_debug_symbols = any(
            entry.copy_debug_symbols
            for entry in outputs.packages + outputs.directories
        )
        self._handle_debug_symbols(
            debug_symbol_manifest_paths,
            need_to_copy_debug_symbols,
            extra_outputs.debug_symbols_manifest,
            time_profile,
        )

        # Handle the rust analyzer manifests.
        crate_spec_paths: set[Path] = set()
        for manifest_path in rust_manifest_paths:
            manifest_path = self.paths.execroot / manifest_path
            with manifest_path.open() as f:
                manifest_json = json.load(f)
                crate_spec_paths.update(
                    self.paths.execroot / spec
                    for spec in manifest_json["crate_specs"]
                )

        rust_crate_specs = (
            bazel_rust_analyzer_utils.load_crate_specs_from_json_files(
                list(crate_spec_paths),
                self.paths,
            )
        )

        rust_crates = bazel_rust_analyzer_utils.convert_crate_specs_to_rust_project_crates(
            rust_crate_specs
        )

        return BazelActionResult(
            configured_args=configured_args,
            output_files=all_output_files,
            source_files=input_files,
            rust_crates=rust_crates,
        )

    def query_for_package_archives(
        self,
        configured_args: list[str],
        packages: list[bazel_action_utils.PackageOutput],
    ) -> list[bazel_action_utils.FileOutput]:
        """Given a list of PackageOutputs, query to find all the archive files to copy.

        This basically is converting PackageOutput entries into FileOutput entries.
        """

        # Query against all package output targets at once.  The query results
        # embed the target labels so that they can be matched with the ninja
        # output paths.
        query_result = run_starlark_cquery(
            self.query_cache,
            self.launcher,
            configured_args,
            [entry.package_label for entry in packages],
            "FuchsiaPackageInfo_archive.cquery",
        )

        # The results are lines of json, so split them and parse each.  They're
        # each a dict of:
        #   target:  "@@<the bazel target>""
        #   archive_path: "<path to the archive file>""
        results: dict[str, str] = {
            p["target"].removeprefix("@@"): p["archive_path"]
            for p in [json.loads(line) for line in query_result]
        }

        # Now match each package output label to its results and add to the file
        # copies to perform.
        package_archive_paths: list[bazel_action_utils.FileOutput] = []
        for entry in packages:
            bazel_archive_path = results.get(entry.package_label)

            # This is just in case we don't get a result for a label, or they
            # change format on us.
            assert bazel_archive_path

            package_archive_paths.append(
                bazel_action_utils.FileOutput(
                    bazel_path=str(self.paths.execroot / bazel_archive_path),
                    ninja_path=entry.archive_path,
                )
            )
        return package_archive_paths

    def _create_buildfiles_genqueries(
        self, targets: list[str]
    ) -> tuple[dict[str, _InputFileGenQueryInfo], Path]:
        """Create a BUILD.bazel file that defines a genquery that returns all build files used.

        Generate a BUILD.bazel file that defines a genquery() target for listing
        all build files that we need to include in the depfile for this target,
        i.e. any changes in these files should trigger a rebuild of this target.

        Args:
            targets: The list of Bazel targets to create the query for

        Returns:
            Tuple of:
                - Dict of _InputFileGenQueryInfo objects, keyed by the Bazel target the query is for
                - Path to the generated source file

            IMPORTANT: These files should be removed after the Bazel build command below
            to ensure our consistency builders do not flake on it. Otherwise the content
            of $BAZEL_EXECROOT/bazel-out/<config_dir>/bin/buildfiles_genquery/genquery
            after a full build will reflect the last bazel_action() command that was
            invoked by Ninja, whose scheduling is not deterministic. In certain cases
            Ninja may decide between two builds to schedule bazel_action() commands
            in a different order, leaving two files with different content in each
            build's relative file path, creating a puzzling consistency error.
        """

        query_buildfile_lines = [
            "# AUTO-GENERATED - DO NOT EDIT!",
            "",
            "",
        ]

        genquery_dir = self.paths.workspace / "bazel-bin/buildfiles_genquery"

        generated_targets = {}
        for target in targets:
            input_files_target_name = filename_from_target_label(
                target, "buildfiles.txt"
            )
            query_buildfile_lines += [
                f"# Generated queries for the target: {target}",
                f"genquery(",
                f'    name = "{input_files_target_name}",',
                f'    expression = "buildfiles(deps({target}))",',
                f'    scope = ["{target}"],',
                f'    opts = ["--output=label"],',
                f"    strict = False,",  # See https://fxbug.dev/500609897
                f")",
                "",
                "",
            ]

            label = f"//buildfiles_genquery:{input_files_target_name}"
            output_path = genquery_dir / input_files_target_name

            generated_targets[target] = _InputFileGenQueryInfo(
                label,
                output_path,
            )

        genquery_build_content = "\n".join(query_buildfile_lines)

        if False:
            print(
                f"GENQUERY BUILD CONTENT:\n{genquery_build_content}\n",
                file=sys.stderr,
            )

        genquery_build_file = (
            self.paths.workspace / "buildfiles_genquery/BUILD.bazel"
        )
        write_file_if_changed(genquery_build_file, genquery_build_content)

        return generated_targets, genquery_build_file

    def _parse_buildfiles_genquery_results_and_source_files_manifests(
        self,
        genqueries: dict[str, _InputFileGenQueryInfo],
        source_files_manifest_paths: list[str],
        time_profile: build_utils.TimeProfile,
    ) -> dict[str, list[str]]:
        """Parse the output files of the genqueries, then query for each targets source files.

        Args:
            genqueries: A dictionary mapping target labels to their genquery info.
            source_files_manifest_paths: A list of paths to the source files manifests.
            time_profile: Time profile to use for timing.

        Returns:
            A dictionary mapping target labels to their source file paths, relative to
            the Ninja build directory.
        """

        time_profile.start("parse_genquery_results_threadpooled")

        def _parse_labels(
            target: str,
            info: _InputFileGenQueryInfo,
        ) -> tuple[str, list[str]]:
            """A worker function for the threadpool to read the genquery output

            This reads the output of one query, and maps the labels into file paths.
            """
            return (
                target,
                info.genquery_output_path.read_text().splitlines(),
            )

        # Use a thread pool for this to to read and parse all the files in parallel
        # For smaller sets of targets, this is slower, but as the number of targets
        # scales up it makes the parsing faster.
        input_file_labels = dict(
            thread_pool_helpers.starmap_threaded(
                _parse_labels, genqueries.items()
            )
        )

        time_profile.start("map_labels_to_paths")
        # now map all the labels into source files.  This is rather slow, so
        # reuse the same label mapper so that it can cache results that are
        # in common between the different targets.
        label_mapper = bazel_label_mapper.BazelLabelMapper(
            str(self.paths.workspace), str(self.paths.ninja_build_dir)
        )
        input_files = dict(
            [
                (target, list(label_mapper.get_sources_for_labels(labels)))
                for target, labels in input_file_labels.items()
            ]
        )

        time_profile.start("map_source_paths")
        source_mapper = bazel_source_path_mapper.BazelSourcePathMapper(
            self.paths
        )
        # Read the source files manifests and build a { label -> sources } map.
        input_file_paths: dict[str, set[str]] = {}
        for manifest_path in source_files_manifest_paths:
            with (self.paths.execroot / manifest_path).open() as f:
                manifest_json = json.load(f)
                # LINT.IfChange(source_files_list_schema)
                label = manifest_json["label"]
                assert isinstance(label, str)
                sources = manifest_json["sources"]
                assert isinstance(sources, list)
                # LINT.ThenChange(//build/bazel/aspects/source_files.bzl:source_files_list_schema)

                assert label.startswith(
                    "@@//"
                ), f"Invalid target label in source file manifest (should start with @@//): {label}"
                label = label.removeprefix("@@")
                # The aspect also walked about out buildfiles_genquery files, ignore them.
                if label.startswith("//buildfiles_genquery"):
                    continue
                input_file_paths[label] = set(sources)

        # The prefix to prepend to all source paths to make them relative to the ninja outdir
        # as expected for the Ninja depfile.
        source_prefix = (
            os.path.relpath(self.paths.fuchsia_dir, self.paths.ninja_build_dir)
            + "/"
        )

        for target, paths in input_file_paths.items():
            sources = []
            for path in paths:
                source_path = source_mapper.resolve_path(path)
                if source_path:
                    sources.append(source_prefix + source_path)
            input_files[target].extend(sources)

        time_profile.stop()

        return input_files

    def _invoke_bazel_and_return_aspect_outputs(
        self,
        targets: list[str],
        cmd_args: list[str],
        aspect_prefix_map: dict[str, bytes],
        time_profile: build_utils.TimeProfile,
    ) -> dict[str, list[str]]:
        """A helper function to handle the invocation of Bazel

        For performance reasons, several aspects will be run during the build, and
        will print the paths of various manifest files of interests to stderr as DEBUG
        outputs.

        This function uses a BazelStderrDebugLineRecorder to filter and record the
        content of such lines, which is controlled through the aspect_prefix_map
        argument. The function's result provides the recorded values to be
        extracted by the caller.

        Args:
            targets: List of Bazel target labels to build.
            cmd_args: Additional arguments to pass to Bazel.
            aspect_prefix_map: A dictionary mapping aspect filter names to their DEBUG line
              prefixes. Used to initialize a BazelStderrDebugLineRecorder instance and
              construct the result value.
            time_profile: Time profile to use for timing.

        Returns:
            A BazelStderrDebugLineRecorder.get_all_recorded_values() dictionary, mapping
            aspect filter names to recorded lists of output strings (one per line).
        """

        if self.global_args.quiet:
            # Still capture stdout to print its content if there is an error
            # running the build command below.
            stdout_sink = stdio_redirection.BytesOutputSink()
            stderr_sink = stdio_redirection.BytesOutputSink()
            is_stdout_pty = False
            is_stderr_pty = False
        else:
            # Send output to regular stdout / stderr instead.
            stdout_sink = stdio_redirection.StdoutOutputSink()
            stderr_sink = stdio_redirection.StderrOutputSink()
            is_stdout_pty = os.isatty(sys.stdout.fileno())
            is_stderr_pty = os.isatty(sys.stderr.fileno())

        # The Bazel stderr must always be captured to extract DEBUG statements
        # that include the paths to manifest files generated by aspects. However,
        # they should be only printed when quiet is False.
        bazel_debug_line_recorder = (
            bazel_action_utils.BazelStderrDebugLineRecorder(
                stderr_sink, aspect_prefix_map
            )
        )

        with stdio_redirection.PipeOutputSink(
            bazel_debug_line_recorder, use_pty=is_stderr_pty
        ) as pty_stderr:
            # This makes mypy happy, as it can't detect the type correctly in the 'with' statement
            assert isinstance(pty_stderr, stdio_redirection.PipeOutputSink)

            with stdio_redirection.PipeOutputSink(
                stdout_sink, use_pty=is_stdout_pty
            ) as pty_stdout:
                # This makes mypy happy, as it can't detect the type correctly in the 'with' statement
                assert isinstance(pty_stdout, stdio_redirection.PipeOutputSink)

                time_profile.start(
                    "run_bazel_command", "Run the Bazel command."
                )
                ret = self.launcher.run_bazel_command(
                    cmd_args,
                    stdout=pty_stdout.get_write_fd(),
                    stderr=pty_stderr.get_write_fd(),
                )

        if ret.returncode != 0:
            if self.global_args.quiet:
                # Assert that the output sinks are the expected types for quiet mode.
                assert isinstance(
                    stdout_sink, stdio_redirection.BytesOutputSink
                )
                assert isinstance(
                    stderr_sink, stdio_redirection.BytesOutputSink
                )

                # Print the captured outputs in quiet mode to help debugging build errors.
                if stdout_sink.data:
                    sys.stdout.buffer.write(stdout_sink.data)
                    sys.stdout.flush()
                if stderr_sink.data:
                    sys.stderr.buffer.write(stderr_sink.data)
                    sys.stderr.flush()

            # Detect the error message corresponding to a Bazel target
            # referencing a @gn_targets//<dir>:<name> label
            # that does not exist. This happens when the GN bazel_action()
            # fails to depend on the proper bazel_input_file() or
            # bazel_input_directory() dependency.
            #
            # NOTE: Path to command.log should be stable, because we explicitly set
            # output_base. See https://bazel.build/run/scripts#command-log.
            if verify_unknown_gn_targets(
                (self.paths.output_base / "command.log")
                .read_text()
                .splitlines(),
                targets,
            ):
                raise BazelActionError()

            # This is a different error, just print it as is.
            #
            # Note most build users are not interested in executing bazel directly, so hiding this
            # message bechind a flag.
            print(
                "\nERROR when calling Bazel. To reproduce, run this in the Ninja output directory:\n\n  %s\n"
                % " ".join(shlex.quote(c) for c in ret.args),
                file=sys.stderr,
            )
            raise BazelActionError()

        return bazel_debug_line_recorder.get_all_recorded_values()

    def _handle_debug_symbols(
        self,
        bazel_manifest_paths: list[str],
        perform_copy: bool,
        manifest_path: Path | None,
        time_profile: build_utils.TimeProfile,
    ) -> None:
        """Perform any post-build operations that need to happen with the debug symbols.

        If they need to be copied, then do so.

        If they need to be written to an output file, then do so.
        """

        if perform_copy or manifest_path:
            debug_symbols_manifest = merge_debug_symbol_manifests(
                bazel_manifest_paths,
                bazel_execroot=self.paths.execroot,
                build_dir=self.paths.ninja_build_dir,
                manifest_output_path=manifest_path,
                time_profile=time_profile,
            )

            if perform_copy:
                time_profile.start(
                    "copy_debug_symbols",
                    "Copy debug symbols to Ninja build directory.",
                )
                copy_debug_symbols_to_build_dir(
                    self.paths.ninja_build_dir, debug_symbols_manifest
                )

            if manifest_path:
                # Write the debug symbols manifest. This is referenced by {BUILD_DIR}/debug_symbols.json
                # which will be used by artifactory to upload the symbols to cloud storage on infra
                # builds.
                with open(manifest_path, "wt") as f:
                    json.dump(debug_symbols_manifest, f, indent=2)


def calculate_platform_config_args(
    platform_config: str,
    platform_label: str,
    rbe_settings: bazel_action_utils.BazelRbeSettings,
) -> list[str]:
    """Constructs the CLI args that define the 'common platform configuration' as needed by the command and any subsequent queries.

    This is an internal helper that creates the cli args, including the RBE
    settings, so that any post-build queries can use the same platform config
    as the action itself did.

    Args:
        platform_config: The name of the bazel config to enable for the platform
        platform_label: The bazel label of the platform
        rbe_settings: The global RBE settings for the build

    Returns:
        List of strings that are the CLI args that together configure Bazel
        correctly for the build and for any subsequent queries.
    """
    platform_config_args = [
        f"--config={platform_config}",
        f"--platforms={platform_label}",
    ]

    # When remote builds are enabled, append the right build arguments.
    # These must appear on the Bazel command-line otherwise remote builds
    # will fail on infra (the reason being that the Bazel wrapper script
    # detects these options to add infra-specific proxy configuration
    # the the final command-line).
    #
    # RBE settings are included in the platform config so that the remote cache can be
    # utilized for queries.
    if rbe_settings.enabled:
        assert rbe_settings.exec_strategy != None
        # If RBE is enabled, append the chosen RBE config to
        # the command line.
        platform_config_args += [f"--config={rbe_settings.exec_strategy}"]
    return platform_config_args


def calculate_jobs_param(
    rbe_settings: bazel_action_utils.BazelRbeSettings,
) -> str | None:
    """Given the RBE settings and the environment vars, determine what --jobs param to use, if any."""
    jobs = None
    # When running jobs remotely, increase the number of allowed jobs to 10x
    # when running jobs locally.  This is different from the reclient config
    # because this controls the _running_ of jobs, not the checking of the
    # cache for jobs.
    if rbe_settings.enabled and rbe_settings.exec_strategy == "remote":
        cpus = os.cpu_count()
        if cpus:
            jobs = 10 * cpus

    if jobs is None:
        # If an explicit job count was passed to `fx build`, tell Bazel to respect it.
        # See https://fxbug.dev/351623259
        job_count = os.environ.get("FUCHSIA_BAZEL_JOB_COUNT")
        if job_count:
            jobs = int(job_count)

    return f"--jobs={jobs}" if jobs else None


def verify_unknown_gn_targets(
    build_files_error: list[str],
    bazel_targets: list[str],
) -> int:
    """Check for unknown @gn_targets// dependencies.

    Args:
        build_files_error: list of error lines from bazel build or query.
        bazel_targets: list of Bazel targets invoked by the GN bazel_action() target.

    Returns:
        On success, simply return 0. On failure, print a human friendly
        error message explaining the situation to stderr, then return 1.
    """
    missing_ninja_outputs = set()
    missing_ninja_packages = set()
    for error_line in build_files_error:
        if not ("ERROR: " in error_line and "@gn_targets" in error_line):
            continue

        pos = error_line.find("@@gn_targets")
        if pos < 0:
            # Should not happen, do not assert and let the caller print the full error
            # after this.
            print(f"UNSUPPORTED ERROR LINE: {error_line}", file=sys.stderr)
            return 0

        ending_pos = error_line.find("'", pos)
        if ending_pos < 0:
            print(f"UNSUPPORTED ERROR LINE: {error_line}", file=sys.stderr)
            return 0

        label = error_line[pos + 1 : ending_pos]  # skip first @.
        if error_line[:pos].endswith(": no such package '"):
            # The line looks like the following when the BUILD.bazel references a label
            # that does not belong to a @gn_targets package.
            #
            # ERROR: <abspath>/BUILD.bazel:<line>:<column>: no such package '@@gn_targets//<dir>': ...
            #
            # This happens when the GN bazel_action() targets fails to dependon the corresponding
            # bazel_input_file() or bazel_input_directory() target, and that none of its other
            # dependencies expose other targets from the same package / directory. The error message
            # does not give any information about the target name being evaluated by the query though.
            missing_ninja_packages.add(label)

        elif error_line[:pos].endswith(": no such target '"):
            # The line looks like this when the BUILD.bazel files references the wrong
            # label from a package exposed in @gn_targets//:
            # ERROR: <abspath>/BUILD.bazel:<line>:<column>: no such target '@@gn_targets//<dir>:<name>' ...
            missing_ninja_outputs.add(label)

    if not missing_ninja_outputs and not missing_ninja_packages:
        return 0

    missing_outputs = sorted(missing_ninja_outputs)
    missing_packages = sorted(missing_ninja_packages)

    _ERROR = """
BAZEL_ACTION_ERROR: Unknown @gn_targets targets.

The following Bazel target(s):  {bazel_targets}
""".format(
        bazel_targets=" ".join(bazel_targets),
    )

    if not missing_packages:
        _ERROR += """Which reference the following unknown @gn_targets labels:

  {missing_bazel_labels}

To fix this, ensure that bazel_input_file() or bazel_input_directory()
targets are defined in the GN graph for:

  {missing_gn_labels}

Then ensure that the GN target depends on them transitively.
""".format(
            missing_bazel_labels="\n  ".join(missing_outputs),
            missing_gn_labels="\n  ".join(
                f"//{o.removeprefix('@gn_targets//')}" for o in missing_outputs
            ),
        )

    else:
        missing_labels = missing_outputs + missing_packages
        missing_build_files = set()
        for label in missing_labels:
            label = label.removeprefix("@gn_targets//")
            gn_dir, sep, gn_name = label.partition(":")
            if sep != ":":
                gn_dir = label
            missing_build_files.add(f"//{gn_dir}/BUILD.gn")

        _ERROR += """Which reference the following unknown @gn_targets labels or packages:

  {missing_bazel_labels}

To fix this, ensure that bazel_input_file() or bazel_input_directory()
targets are defined in the following build files:

  {missing_build_files}

Then ensure that the GN target depends on them transitively.
""".format(
            missing_bazel_labels="\n  ".join(
                missing_outputs + missing_packages
            ),
            missing_build_files="\n  ".join(sorted(missing_build_files)),
        )

    print(_ERROR, file=sys.stderr)
    return 1


def merge_target_info_outputs(
    target_infos: T.Sequence[bazel_action_utils.BazelTargetInfo],
) -> tuple[BazelActionOutputs, list[Path]]:
    """Merge all the outputs and needed input gn targets for the given list of BazelTargetInfos.

    Args:
      Sequence of BazelTargetInfo objects

    Returns:
      BazelActionOutputs that merges all target_info's outputs together
      List of manifests to GN targets needed by the BazelTargetInfos
    """
    # Merge all the expected outputs together
    file_outputs: list[bazel_action_utils.FileOutput] = []
    directory_outputs: list[bazel_action_utils.DirectoryOutput] = []
    package_outputs: list[bazel_action_utils.PackageOutput] = []
    final_symlink_outputs: list[bazel_action_utils.FinalSymlinkOutput] = []

    # This is the set of gn input targets manifests
    gn_target_manifests: set[Path] = set()

    for target_info in target_infos:
        file_outputs.extend(target_info.copy_outputs)
        directory_outputs.extend(target_info.directory_outputs)
        package_outputs.extend(target_info.package_outputs)
        final_symlink_outputs.extend(target_info.final_symlink_outputs)

        gn_target_manifests.add(Path(target_info.gn_targets_manifest))

    return (
        BazelActionOutputs(
            file_outputs,
            directory_outputs,
            package_outputs,
            final_symlink_outputs,
        ),
        list(gn_target_manifests),
    )


def run_starlark_cquery(
    query_cache: build_utils.BazelQueryCache,
    bazel_launcher: build_utils.BazelLauncher,
    configured_args: list[str],
    query_targets: list[str],
    starlark_filename: str,
) -> list[str]:
    """Run a Bazel cquery and process its output with a starlark file.

    Args:
        query_targets: A list of Bazel targets to run the query over.
        starlark_filename: Name of starlark file from //build/bazel/starlark.
    Returns:
        A list of output lines.
    Raises:
        AssertionError in case of failure.
    """
    result = run_bazel_query(
        query_cache,
        bazel_launcher,
        "cquery",
        [
            "--config=quiet",
            "--output=starlark",
            "--starlark:file",
            get_input_starlark_file_path(starlark_filename),
        ]
        + configured_args
        + ["set(%s)" % " ".join(query_targets)],
    )
    assert result is not None
    return result


def get_input_starlark_file_path(filename: FilePath) -> str:
    """Return the path of a input starlark file for Bazel queries.

    Args:
       filename: File name, searched in //build/bazel/starlark/
    Returns:
       file path to the corresponding file.
    """
    result = os.path.join(_STARLARK_DIR, filename)
    assert os.path.isfile(result), f"Missing starlark input file: {result}"
    return result


def run_bazel_query(
    query_cache: build_utils.BazelQueryCache,
    bazel_launcher: build_utils.BazelLauncher,
    query_cmd: str,
    query_args: list[str],
) -> list[str] | None:
    """Run a Bazel query, return output as list of lines.

    Args:
        query_cmd: Query command ("query", "cquery" or "aquery").
        query_args: Query arguments.
    Returns:
        On success, a list of output lines. On failure return None.
    """
    return query_cache.get_query_output(
        query_cmd,
        query_args,
        bazel_launcher,
        log=lambda m: (
            print(f"DEBUG: {m}", file=sys.stderr)
            if _DEBUG_BAZEL_QUERIES
            else None
        ),
    )


class _BazelOutputCopier(object):
    """This exists to consolidate together all of the Bazel output-copying logic."""

    def __init__(self, paths: build_utils.BazelPaths):
        self.paths = paths

    def copy(
        self,
        outputs: BazelActionOutputs,
        time_profile: build_utils.TimeProfile,
    ) -> list[Path]:
        """Copy all the given outputs.

        Returns a list of all output destination files that were copied to, including tracked
        files in DirectoryOutputs.
        """

        output_files = self._copy_files(outputs.files, time_profile)
        tracked_files = self._copy_directories(
            outputs.directories, time_profile
        )
        self._make_final_symlinks(outputs.final_symlinks, time_profile)

        return output_files + tracked_files

    def _copy_files(
        self,
        file_outputs: list[bazel_action_utils.FileOutput],
        time_profile: build_utils.TimeProfile,
    ) -> list[Path]:
        """Perform all of the file-copying logic.

        Returns a list of the destination file paths that were copied to.
        """
        time_profile.start(
            "check_outputs_for_copying",
            "Validate output files to copy are actually files.",
        )

        file_copies: list[tuple[Path, Path]] = []
        unwanted_dirs = []

        for file_output in file_outputs:
            src_path = self.paths.workspace / file_output.bazel_path
            if src_path.is_dir():
                unwanted_dirs.append(src_path)
                continue
            dst_path = Path(file_output.ninja_path)
            file_copies.append((src_path, dst_path))

        if unwanted_dirs:
            raise FileOutputConfigError(
                "\nDirectories are not allowed in --file-outputs Bazel paths, got directories:\n\n%s\n"
                % "\n".join(str(d) for d in unwanted_dirs)
            )

        if file_copies:
            time_profile.start(
                "check_copy_files",
                "Check to see if files need to be copied or not.",
            )

            files_to_copy = [
                file_copy
                for file_copy in thread_pool_helpers.filter_threaded(
                    check_if_need_to_copy_file, file_copies
                )
                if file_copy
            ]

            if files_to_copy:
                time_profile.start(
                    "copy_files",
                    "Copy Bazel output files to Ninja build directory",
                )

                thread_pool_helpers.starmap_threaded(
                    hardlink_or_copy_writable,
                    [
                        (src, dst, self.paths.output_base)
                        for src, dst in files_to_copy
                    ],
                )
        return [dst for _, dst in file_copies]

    def _copy_directories(
        self,
        directory_outputs: list[bazel_action_utils.DirectoryOutput],
        time_profile: build_utils.TimeProfile,
    ) -> list[Path]:
        time_profile.start(
            "check_output_directories",
            "Validate that output directories are ready to be copied.",
        )

        dir_copies: list[tuple[Path, Path, list[FilePath]]] = []
        missing_directories: list[Path] = []
        unwanted_files: list[Path] = []
        invalid_tracked_files: list[Path] = []

        for dir_output in directory_outputs:
            src_path = self.paths.workspace / dir_output.bazel_path
            if not src_path.exists():
                missing_directories.append(src_path)
                continue
            if not src_path.is_dir():
                unwanted_files.append(src_path)
                continue
            for tracked_file in dir_output.tracked_files:
                tracked_file = src_path / tracked_file
                if not tracked_file.is_file():
                    invalid_tracked_files.append(tracked_file)
            dst_path = Path(dir_output.ninja_path)
            dir_copies.append(
                (
                    src_path,
                    dst_path,
                    [Path(f) for f in dir_output.tracked_files],
                )
            )

        if missing_directories:
            raise MissingOutputDirectoryError(
                "\nError: Directory provided to --directory-outputs is missing, got:\n\n%s\n"
                % "\n".join(str(d) for d in missing_directories)
            )

        if unwanted_files:
            raise DirectoryOutputConfigError(
                "\nError: Non-directories are not allowed in --directory-outputs Bazel path, got:\n\n%s\n"
                % "\n".join(str(f) for f in unwanted_files)
            )

        if invalid_tracked_files:
            raise MissingTrackedFileError(
                "\nError: Missing or non-directory tracked files from --directory-outputs Bazel path:\n\n%s\n"
                % "\n".join(str(f) for f in invalid_tracked_files)
            )

        if dir_copies:
            time_profile.start(
                "copy_directories",
                "Copy Bazel output directories to Ninja build directory",
            )
            for src_path, dst_path, tracked_files in dir_copies:
                copy_directory_if_changed(src_path, dst_path, tracked_files)

        # Return all of the full paths to the tracked files in the directories, as these are
        # the "destination output files" of these directories.
        all_tracked_files: list[Path] = []
        for dst, _, tracked_files in dir_copies:
            all_tracked_files.extend([dst / file for file in tracked_files])
        return all_tracked_files

    def _make_final_symlinks(
        self,
        final_symlink_outputs: list[bazel_action_utils.FinalSymlinkOutput],
        time_profile: build_utils.TimeProfile,
    ) -> None:
        time_profile.start("symlink_outputs", "Symlink output files.")
        final_symlinks: list[tuple[Path, Path]] = []
        for final_symlink_output in final_symlink_outputs:
            src_path = self.paths.workspace / final_symlink_output.bazel_path
            target_path = src_path.resolve()
            link_path = Path(final_symlink_output.ninja_path)
            final_symlinks.append((target_path, link_path))

        if final_symlinks:
            # This doesn't need to use a thread pool because as of today there's only ever
            # one file, in one action, that uses this codepath.
            for target_path, link_path in final_symlinks:
                build_utils.force_symlink(link_path, target_path)


def filename_from_target_label(
    target: str, extension: str | None = None
) -> str:
    """Convert a target label into string that can be a filename.

    This isn't a reversible function, as the resultant filename can be
    converted to multiple different targets.

    TODO: consider using a hash instead to ensure that there are not possible
          name collisions

    Examples:
        > filename_from_target_label("//foo/bar:baz")
        "foo_bar_baz"

        > filename_from_target_label("//foo/bar:baz", "txt)
        "foo_bar_baz.txt"

    Args:

        target - The bazel target to flatten into a filename.
        extension - An optional extension to add to the filename.
    """
    temp = target.removeprefix("//").replace("/", "_").replace(":", "_")

    if extension:
        return temp + "." + extension
    else:
        return temp


def merge_debug_symbol_manifests(
    debug_symbol_manifest_paths: list[str],
    bazel_execroot: Path,
    build_dir: Path,
    manifest_output_path: Path | None,
    time_profile: build_utils.TimeProfile,
) -> list[DebugSymbolEntryType]:
    """Generate final debug symbol manifest.

    Args:
        debug_symbol_manifest_paths: The list of debug symbol manifest paths
            generated by the debug symbol aspect, which were extracted from
            the Bazel stderr's DEBUG lines.
        bazel_execroot: Path to Bazel execroot.
        build_dir: Path to Ninja build direvtory.
        manifest_output_path: Path where the manifest will be written.
            only used in error messages, this function does not write
            the file itself.
        time_profile: A TimeProfile instance.
    Returns:
        a list of dictionaries describing debug symbols according to the
        GN //:debug_symbols schema.
    """
    time_profile.start(
        "merge_debug_symbol_manifests",
        "Merge target-specific manifests into final version.",
    )

    output_manifest = []
    recorded_entries: set[str] = set()

    for manifest_path in debug_symbol_manifest_paths:
        input_manifest_path = os.path.join(bazel_execroot, manifest_path)
        with open(input_manifest_path, "rt") as f:
            input_manifest = json.load(f)

        for input_entry in input_manifest:
            src_debug = input_entry["debug"]
            if src_debug in recorded_entries:
                continue  # Ignore duplicates.

            recorded_entries.add(src_debug)

            # Adjust paths
            entry = input_entry.copy()
            for key in ("debug", "stripped", "breakpad", "elf_build_id"):
                src_path = entry.get(key, "")
                if src_path:
                    # Convert execroot-relative paths to a build-dir relative ones, using
                    # os.path.realpath() to resolve symlinks properly.
                    entry[key] = os.path.relpath(
                        os.path.realpath(
                            os.path.join(bazel_execroot, src_path)
                        ),
                        build_dir,
                    )
            output_manifest.append(entry)

    time_profile.start(
        "compute_elf_build_ids",
        "Extra GNU build-id values from ELF binaries.",
    )
    parser = DebugSymbolsManifestParser()
    parser.enable_build_id_resolution()
    parser.parse_manifest_json(output_manifest, manifest_output_path)

    if _DEBUG_SYMBOL_EXPORT:
        print("DEBUG SYMBOLS:\n%s" % output_manifest)

    time_profile.stop()
    return parser.entries


def copy_debug_symbols_to_build_dir(
    build_dir: Path, debug_symbols_manifest: list[DebugSymbolEntryType]
) -> None:
    """Copy debug symbols from the Bazel execroot to {NINJA_BUILD_DIR}/.build-id

    Useful when performing local debugging and symbolization. This does not affect
    infra builds which use artifactory instead to upload the symbols to cloud
    storage.

    Args:
        build_dir: Path to Ninja build directory.
        debug_symbols_manifest: A list of DebugSymbolEntryType values, similar
            to what is returned by merge_debug_symbol_manifests().
    """
    exporter = DebugSymbolExporter(
        build_dir,
        log=lambda m: (
            print(f"DEBUG: {m}", file=sys.stderr)
            if _DEBUG_SYMBOL_EXPORT
            else None
        ),
    )
    exporter.parse_debug_symbols(debug_symbols_manifest)
    debug_copies = exporter.get_debug_symbols_to_build_id_copies(
        os.path.join(build_dir, ".build-id")
    )

    def copy_build_id_file(src_path: str, dst_path: str) -> None:
        os.makedirs(os.path.dirname(dst_path), exist_ok=True)
        shutil.copyfile(src_path, dst_path)

    thread_pool_helpers.starmap_threaded(
        copy_build_id_file,
        debug_copies,
    )