build/bazel/scripts/bazel_action_impl.py

#!/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 dataclasses
import json
import os
import shlex
import sys
from functools import partial
from pathlib import Path

import thread_pool_helpers

_SCRIPT_DIR = os.path.dirname(__file__)
sys.path.insert(0, _SCRIPT_DIR)
# LINT.IfChange(imports)
import bazel_action_utils
import build_utils
import stdio_redirection
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,
)

# 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


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)

# 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."""


@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]


@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

        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
    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.

        debug_symbol_manifest_paths:  A list of paths to debug symbol manifests

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

    configured_args: list[str]
    debug_symbol_manifest_paths: list[str]
    output_files: list[Path]


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,
        )

        # Create the base command-line from the command to run (build, etc.), the platform
        # configuration, and the set of targets to run the command against.
        cmd_args: list[str] = [
            command,
            *configured_args,
            *targets,
        ]

        # 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",
        ]

        # 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"]

        # 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_args
                )
            )
        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_args
                )
                + "\n",
            )

        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 debug symbol manifests. However, they should
        # be only printed when quiet is False.
        debug_symbol_manifest_paths: list[str] = []

        debug_symbol_manifest_filtering_sink = (
            bazel_action_utils.BazelStderrDebugLineFilter(
                stderr_sink,
                partial(bazel_debug_line_filter, debug_symbol_manifest_paths),
            )
        )

        with stdio_redirection.PipeOutputSink(
            debug_symbol_manifest_filtering_sink, 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()

        # If we're building, and 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.
        if command == "build" and outputs.packages:
            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)
        all_output_files = output_copier.copy(outputs, time_profile)

        return BazelActionResult(
            configured_args=configured_args,
            debug_symbol_manifest_paths=debug_symbol_manifest_paths,
            output_files=all_output_files,
        )

    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 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 bazel_debug_line_filter(manifest_paths: list[str], line: bytes) -> bool:
    """Filters a DEBUG line from Bazel stderr and extracts debug manifest paths.

    This function is used as a filter for `BazelStderrDebugLineFilter`.  See
    BazelStderrDebugLineFilter documentation for details.

    Args:
        manifest_paths: A list of strings. If a debug symbol manifest path is found
            in the `line`, it will be decoded and appended to this list. This list
            is mutated by the function.
        line: The bytes line read from Bazel's stderr.

    Returns:
        True if the line contains a debug symbol manifest path and should be
        filtered out (i.e., not printed to the actual stderr). False otherwise.

    """
    path_prefix = _BAZEL_DEBUG_SYMBOLS_MANIFEST_PATH_PREFIX
    pos = line.find(path_prefix)
    if pos < 0:
        return False  # Keep this line
    manifest_paths.append(
        line[pos + len(path_prefix) :].decode("utf-8").strip()
    )
    return True  # Skip this line


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 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 BazelActionError(
                "\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 BazelActionError(
                "\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 BazelActionError(
                "\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 BazelActionError(
                "\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:
        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:
            time_profile.start("symlink_outputs", "Symlink output files.")
            # 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)