Download cached source directories in the runfiles of top-level targets (#31473)

### Description

The files below a source directory of a repo that has been retrieved from the remote repo contents cache are only downloaded when the directory is requested from the prefetcher. For the runfiles of a top-level target, only artifacts with remote metadata were requested, which a source directory doesn't have.

Source directories in external repos are now requested as well if they are selected for download.

### Motivation

With a source directory of a cached repo in the runfiles of a binary, `bazel run` ran the binary without the directory in its runfiles tree.

Work towards #31456

### Build API Changes

No

### Release Notes

RELNOTES: None

Closes #31473

COPYBARA_INTEGRATE_REVIEW=https://github.com/bazelbuild/bazel/pull/31473 from fmeum:rrcc-run-source-directories 99e1973eee3e244b8e44e7191c63822c55f78f52
PiperOrigin-RevId: 999899308
Change-Id: I72c3806cc741f20d19b3aaeeb0202ef8c49de322
diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteImportantOutputHandler.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteImportantOutputHandler.java
index 0cab05e..6f7e82e 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/RemoteImportantOutputHandler.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteImportantOutputHandler.java
@@ -27,6 +27,7 @@
 import com.google.devtools.build.lib.actions.Artifact.DerivedArtifact;
 import com.google.devtools.build.lib.actions.Artifact.TreeFileArtifact;
 import com.google.devtools.build.lib.actions.FileArtifactValue;
+import com.google.devtools.build.lib.actions.FileStateType;
 import com.google.devtools.build.lib.actions.ImportantOutputHandler;
 import com.google.devtools.build.lib.actions.InputMetadataProvider;
 import com.google.devtools.build.lib.profiler.SilentCloseable;
@@ -221,8 +222,18 @@
         return;
       }
 
-      if (remoteOutputChecker.shouldDownloadOutput(artifact, metadata)) {
-        ensuredOutputMetadata.add(metadata);
+      // The metadata of a source directory in an external repo doesn't tell whether the files
+      // below it are only available remotely, which is the case as long as the repo is served from
+      // the remote repo contents cache.
+      boolean isSourceDirectory =
+          artifact.isSourceArtifact() && metadata.getType() == FileStateType.DIRECTORY;
+      if (isSourceDirectory
+          ? remoteOutputChecker.shouldDownloadOutput(
+              artifact.getExecPath(), /* treeRootExecPath= */ null)
+          : remoteOutputChecker.shouldDownloadOutput(artifact, metadata)) {
+        if (!isSourceDirectory) {
+          ensuredOutputMetadata.add(metadata);
+        }
         futures.add(
             actionInputPrefetcher.prefetchFiles(
                 artifact instanceof DerivedArtifact derivedArtifact
diff --git a/src/test/py/bazel/bzlmod/remote_repo_contents_cache_test.py b/src/test/py/bazel/bzlmod/remote_repo_contents_cache_test.py
index 3ce03d1..f6a44e8 100644
--- a/src/test/py/bazel/bzlmod/remote_repo_contents_cache_test.py
+++ b/src/test/py/bazel/bzlmod/remote_repo_contents_cache_test.py
@@ -2475,6 +2475,82 @@
     self.assertIn('hello from my_bin', '\n'.join(stdout))
     self.assertFalse(os.path.exists(os.path.join(repo_dir, 'BUILD')))
 
+  def testRunWithSourceDirectoryInRunfiles(self):
+    if self.IsWindows():
+      self.skipTest('requires a shell script')
+    self.ScratchFile(
+        'MODULE.bazel',
+        [
+            'repo = use_repo_rule("//:repo.bzl", "repo")',
+            'repo(name = "my_repo")',
+        ],
+    )
+    self.ScratchFile('BUILD.bazel')
+    self.ScratchFile(
+        'repo.bzl',
+        [
+            'def _repo_impl(rctx):',
+            '  rctx.file("BUILD", "exports_files([\'dir\'])")',
+            '  rctx.file("dir/data.txt", "hello from a source directory")',
+            '  print("JUST FETCHED")',
+            '  return rctx.repo_metadata(reproducible=True)',
+            'repo = repository_rule(_repo_impl)',
+        ],
+    )
+    self.ScratchFile(
+        'main/launcher.bzl',
+        [
+            'def _launcher_impl(ctx):',
+            '  exe = ctx.actions.declare_file(ctx.label.name + ".sh")',
+            '  ctx.actions.write(',
+            '    exe,',
+            '    "#!/bin/sh\\ncat \\"$0.runfiles/data/data.txt\\"\\n",',
+            '    is_executable = True,',
+            '  )',
+            '  return [DefaultInfo(',
+            '    executable = exe,',
+            (
+                '    runfiles = ctx.runfiles(root_symlinks = {"data":'
+                ' ctx.file.src}),'
+            ),
+            '  )]',
+            'launcher = rule(',
+            '  implementation = _launcher_impl,',
+            '  attrs = {"src": attr.label(allow_single_file = True)},',
+            '  executable = True,',
+            ')',
+        ],
+    )
+    self.ScratchFile(
+        'main/BUILD.bazel',
+        [
+            'load("//main:launcher.bzl", "launcher")',
+            'launcher(name = "launcher", src = "@my_repo//:dir")',
+        ],
+    )
+
+    # First fetch: not cached
+    _, stdout, stderr = self.RunBazel(['run', '//main:launcher'])
+    self.assertIn('JUST FETCHED', '\n'.join(stderr))
+    self.assertIn('hello from a source directory', '\n'.join(stdout))
+
+    # After expunging: cached. Building the binary without downloading its
+    # outputs doesn't download the contents of the source directory.
+    repo_dir = self.RepoDir('my_repo')
+    self.RunBazel(['clean', '--expunge'])
+    _, _, stderr = self.RunBazel(
+        ['build', '--remote_download_minimal', '//main:launcher']
+    )
+    self.assertNotIn('JUST FETCHED', '\n'.join(stderr))
+    self.assertFalse(os.path.exists(os.path.join(repo_dir, 'dir/data.txt')))
+
+    # Running the binary downloads them since it has the source directory in
+    # its runfiles.
+    _, stdout, stderr = self.RunBazel(['run', '//main:launcher'])
+    self.assertNotIn('JUST FETCHED', '\n'.join(stderr))
+    self.assertIn('hello from a source directory', '\n'.join(stdout))
+    self.assertTrue(os.path.exists(os.path.join(repo_dir, 'dir/data.txt')))
+
   def testReverseDependencyDirection(self):
     # Set up two repos that retain their predeclared input hashes across two
     # builds but still reverse their dependency direction. Depending on how repo