diff --git a/Cargo.lock b/Cargo.lock
index ef632433429..dc2fc553563 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2475,7 +2475,9 @@ dependencies = [
  "sha2 0.10.9",
  "tempfile",
  "thiserror 2.0.17",
+ "tokio",
  "tracing",
+ "walkdir",
  "warp_core",
  "warp_errors",
 ]
diff --git a/crates/build_cache/Cargo.toml b/crates/build_cache/Cargo.toml
index 7d2a2b610d3..7be7177b5f8 100644
--- a/crates/build_cache/Cargo.toml
+++ b/crates/build_cache/Cargo.toml
@@ -9,6 +9,7 @@ license.workspace = true
 [dependencies]
 async-io.workspace = true
 command.workspace = true
+futures.workspace = true
 futures-lite.workspace = true
 hex.workspace = true
 instant.workspace = true
@@ -21,9 +22,11 @@ serde_json.workspace = true
 sha2.workspace = true
 tempfile.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["rt", "sync"] }
 tracing.workspace = true
 warp_core.workspace = true
 warp_errors.workspace = true
+walkdir.workspace = true
 
 [dev-dependencies]
-futures.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread", "time"] }
diff --git a/crates/build_cache/examples/validate_spacectl.rs b/crates/build_cache/examples/validate_spacectl.rs
index 385b0a6c6be..e4dd001cc81 100644
--- a/crates/build_cache/examples/validate_spacectl.rs
+++ b/crates/build_cache/examples/validate_spacectl.rs
@@ -1,9 +1,8 @@
-use std::cell::RefCell;
 use std::collections::BTreeMap;
 use std::ffi::OsStr;
 use std::path::{Path, PathBuf};
 use std::process::ExitCode;
-use std::rc::Rc;
+use std::sync::{Arc, Mutex};
 use std::{env, fs};
 
 use build_cache::{
@@ -11,7 +10,6 @@ use build_cache::{
     setup_cache,
 };
 use command::r#async::Command;
-use futures_lite::future;
 use serde_json::Value;
 
 struct Fixture {
@@ -22,6 +20,10 @@ struct Fixture {
 const CARGO_TOML: &str =
     "[package]\nname = \"cache-fixture\"\nversion = \"0.1.0\"\nedition = \"2021\"\n";
 const GO_MOD: &str = "module example.com/cache-fixture\n\ngo 1.22\n";
+const PACKAGE_JSON: &str =
+    "{\"name\":\"cache-fixture\",\"version\":\"1.0.0\",\"lockfileVersion\":3}\n";
+const PACKAGE_LOCK: &str =
+    "{\"name\":\"cache-fixture\",\"version\":\"1.0.0\",\"lockfileVersion\":3,\"packages\":{}}\n";
 
 const FIXTURES: &[Fixture] = &[
     Fixture {
@@ -49,6 +51,16 @@ const FIXTURES: &[Fixture] = &[
         files: &[("Cargo.toml", CARGO_TOML), ("go.mod", GO_MOD)],
         expected_modes: &["go", "rust"],
     },
+    Fixture {
+        name: "nested",
+        files: &[
+            ("Cargo.toml", CARGO_TOML),
+            ("frontend/package.json", PACKAGE_JSON),
+            ("frontend/package-lock.json", PACKAGE_LOCK),
+            ("backend/go.mod", GO_MOD),
+        ],
+        expected_modes: &[],
+    },
     Fixture {
         name: "node",
         files: &[(
@@ -129,13 +141,16 @@ fn run() -> Result<bool, String> {
         println!("  {}: {}", repository.name, repository.cwd.display());
     }
 
-    let responses = Rc::new(RefCell::new(Vec::new()));
-    let report = future::block_on(setup_cache(
+    let responses = Arc::new(Mutex::new(Vec::new()));
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .map_err(|error| error.to_string())?;
+    let report = runtime.block_on(setup_cache(
         cache_root,
         repositories,
         additional_global_modes,
         {
-            let responses = Rc::clone(&responses);
+            let responses = Arc::clone(&responses);
             move |mut command| {
                 let cwd = command
                     .get_current_dir()
@@ -145,11 +160,11 @@ fn run() -> Result<bool, String> {
                     .get_args()
                     .any(|argument| argument == OsStr::new("--dry_run=true"));
                 configure_isolated_environment(&mut command, &isolated_home, &command_path);
-                let responses = Rc::clone(&responses);
+                let responses = Arc::clone(&responses);
                 async move {
                     let bytes = default_run_command(command).await?;
                     let value = serde_json::from_slice(&bytes).unwrap_or(Value::Null);
-                    responses.borrow_mut().push(CapturedResponse {
+                    responses.lock().unwrap().push(CapturedResponse {
                         cwd,
                         dry_run,
                         value,
@@ -165,11 +180,13 @@ fn run() -> Result<bool, String> {
     print_environment(&report);
 
     let expected_mode_failures = validate_fixture_modes(&report, &fixtures);
+    let nested_fixture_failures = validate_nested_fixture(&report, &fixtures);
     let repository_cache_root_failures = validate_repository_cache_roots(&report);
-    let mount_failures = validate_mounts(&responses.borrow());
+    let mount_failures = validate_mounts(&responses.lock().unwrap());
     let degradation_count = report.degradations().count();
     let missing_plan = usize::from(report.plan.is_none());
     let failure_count = expected_mode_failures
+        + nested_fixture_failures
         + repository_cache_root_failures
         + mount_failures
         + degradation_count
@@ -182,6 +199,7 @@ fn run() -> Result<bool, String> {
         println!(
             "validation failed: {degradation_count} degraded invocation(s), \
              {expected_mode_failures} mode mismatch(es), \
+             {nested_fixture_failures} nested fixture mismatch(es), \
              {repository_cache_root_failures} duplicate repository cache root(s), \
              {mount_failures} mount mismatch(es), \
              {missing_plan} missing plan(s)"
@@ -390,6 +408,9 @@ fn validate_fixture_modes(report: &build_cache::CacheSetupReport, fixtures: &[&F
 
     let mut failures = 0;
     for fixture in fixtures {
+        if fixture.name == "nested" {
+            continue;
+        }
         let actual_modes = actual.get(fixture.name).copied().unwrap_or_default();
         let expected_modes = fixture.expected_modes;
         if actual_modes == expected_modes {
@@ -406,6 +427,59 @@ fn validate_fixture_modes(report: &build_cache::CacheSetupReport, fixtures: &[&F
     }
     failures
 }
+
+fn validate_nested_fixture(report: &build_cache::CacheSetupReport, fixtures: &[&Fixture]) -> usize {
+    if !fixtures.iter().any(|fixture| fixture.name == "nested") {
+        return 0;
+    }
+
+    println!();
+    println!("nested fixture checks:");
+    let Some(plan) = &report.plan else {
+        println!("  mismatch: no cache plan");
+        return 1;
+    };
+    let configurations = plan
+        .configurations
+        .iter()
+        .filter(|configuration| {
+            matches!(
+                &configuration.scope,
+                CacheScope::Repository { name, .. } if name == "nested"
+            )
+        })
+        .collect::<Vec<_>>();
+    let expected = [
+        (Path::new("nested"), "rust"),
+        (Path::new("nested/backend"), "go"),
+        (Path::new("nested/frontend"), "npm"),
+    ];
+    let mut failures = 0;
+    for (suffix, mode) in expected {
+        let matched = configurations.iter().any(|configuration| {
+            configuration.cwd.ends_with(suffix)
+                && configuration.modes.iter().any(|actual| actual == mode)
+        });
+        if matched {
+            println!("  ok {}: {mode}", suffix.display());
+        } else {
+            println!("  mismatch {}: expected {mode}", suffix.display());
+            failures += 1;
+        }
+    }
+    let distinct_cache_roots = configurations
+        .iter()
+        .map(|configuration| &configuration.relative_cache_dir)
+        .collect::<std::collections::BTreeSet<_>>()
+        .len();
+    if distinct_cache_roots == configurations.len() {
+        println!("  ok: distinct nested cache roots");
+    } else {
+        println!("  mismatch: nested cache roots are not distinct");
+        failures += 1;
+    }
+    failures
+}
 fn validate_repository_cache_roots(report: &build_cache::CacheSetupReport) -> usize {
     println!();
     println!("repository cache root checks:");
diff --git a/crates/build_cache/src/discovery.rs b/crates/build_cache/src/discovery.rs
new file mode 100644
index 00000000000..4ef1c2ad215
--- /dev/null
+++ b/crates/build_cache/src/discovery.rs
@@ -0,0 +1,346 @@
+//! Deterministic discovery of repository roots that may need independent build caches.
+//!
+//! Repositories are scanned in cache-key order. Each repository root is emitted before marked
+//! descendants selected by a sorted depth-first walk, so scan limits always retain the same roots.
+use std::collections::BTreeSet;
+use std::path::{Component, Path, PathBuf};
+
+use sha2::{Digest, Sha256};
+use tokio::sync::mpsc;
+use walkdir::{DirEntry, WalkDir};
+
+use crate::{RepoCacheKey, RepositoryCacheSource};
+
+pub(super) const DETECTION_CONCURRENCY: usize = 8;
+
+const MAX_WALK_DEPTH: usize = 7;
+const MAX_VISITED_DIRECTORIES: usize = 10_000;
+const MAX_CHILD_CANDIDATES: usize = 32;
+
+const IGNORED_DIRECTORIES: &[&str] = &[
+    ".git",
+    "node_modules",
+    "target",
+    "Pods",
+    "vendor",
+    "dist",
+    "build",
+    ".venv",
+    ".tox",
+    "DerivedData",
+];
+
+const CODEBASE_MARKER_FILENAMES: &[&str] = &[
+    "Brewfile",
+    "bun.lock",
+    "Podfile",
+    "composer.json",
+    "deno.lock",
+    "go.mod",
+    "go.work",
+    ".golangci.yml",
+    ".golangci.yaml",
+    "gradlew",
+    "build.gradle",
+    "pom.xml",
+    "mise.toml",
+    ".mise.toml",
+    ".tool-versions",
+    "flake.nix",
+    "shell.nix",
+    "default.nix",
+    "package-lock.json",
+    "pnpm-lock.yaml",
+    "poetry.lock",
+    "requirements.txt",
+    "Gemfile",
+    "Cargo.toml",
+    "Package.swift",
+    "Tuist.swift",
+    "tuist.toml",
+    "uv.lock",
+    "yarn.lock",
+];
+
+const CODEBASE_MARKER_PATHS: &[&[&str]] = &[
+    &["mise", "config.toml"],
+    &[".mise", "config.toml"],
+    &[".config", "mise.toml"],
+    &[".config", "mise", "config.toml"],
+];
+
+/// Canonical detection order: repository key, then root before normalized descendant paths.
+#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
+pub(super) struct CandidateKey {
+    pub repo_key: RepoCacheKey,
+    pub normalized_relative_path: Option<PathBuf>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct CacheCandidate {
+    pub key: CandidateKey,
+    pub source: RepositoryCacheSource,
+    pub relative_cache_dir: PathBuf,
+    pub stable_child_id: Option<String>,
+}
+
+/// Emits one repository root followed by its distinct marked descendants.
+///
+/// Returning `false` means the receiver was dropped and all discovery must stop.
+fn produce_repository_candidates(
+    key: RepoCacheKey,
+    source: RepositoryCacheSource,
+    sender: &mpsc::Sender<CacheCandidate>,
+) -> bool {
+    let span = tracing::info_span!(
+        target: "build_cache",
+        "discover_repository_cache_roots",
+        tags.cloud_agent = true,
+        repo_key = %key,
+        visited_directory_count = tracing::field::Empty,
+        selected_child_count = tracing::field::Empty,
+        truncation_reason = tracing::field::Empty,
+    );
+    let _guard = span.enter();
+    let mut selected_paths = BTreeSet::new();
+    let mut visited_directories = 1;
+    let mut truncation = None;
+    let mut receiver_open = sender
+        .blocking_send(root_candidate(key.clone(), source.clone()))
+        .is_ok();
+
+    if receiver_open {
+        let walker = WalkDir::new(&source.cwd)
+            .min_depth(1)
+            .max_depth(MAX_WALK_DEPTH)
+            .follow_links(false)
+            .follow_root_links(false)
+            .sort_by_file_name()
+            .into_iter()
+            .filter_entry(|entry| {
+                !(entry.file_type().is_symlink()
+                    || entry.file_type().is_dir()
+                        && IGNORED_DIRECTORIES
+                            .iter()
+                            .any(|ignored| entry.file_name() == *ignored))
+            });
+
+        'walk: for entry in walker {
+            if sender.is_closed() {
+                receiver_open = false;
+                break;
+            }
+            let entry = match entry {
+                Ok(entry) => entry,
+                Err(error) => {
+                    tracing::warn!(
+                        target: "build_cache",
+                        ?error,
+                        "build cache root discovery skipped unreadable entry"
+                    );
+                    continue;
+                }
+            };
+            if entry.file_type().is_dir() {
+                if visited_directories == MAX_VISITED_DIRECTORIES {
+                    truncation = Some(TruncationReason::DirectoryLimit);
+                    break;
+                }
+                visited_directories += 1;
+            }
+            let Some(path) = find_candidate_for_entry(&entry, &source.cwd) else {
+                continue;
+            };
+            let Some(normalized_relative_path) = normalize_relative_path(&source.cwd, &path) else {
+                continue;
+            };
+            if selected_paths.contains(&normalized_relative_path) {
+                continue;
+            }
+            if selected_paths.len() == MAX_CHILD_CANDIDATES {
+                truncation = Some(TruncationReason::CandidateLimit);
+                break 'walk;
+            }
+
+            selected_paths.insert(normalized_relative_path.clone());
+            let candidate =
+                child_candidate(key.clone(), source.clone(), path, normalized_relative_path);
+            if sender.blocking_send(candidate).is_err() {
+                receiver_open = false;
+                break 'walk;
+            }
+        }
+    }
+
+    span.record("visited_directory_count", visited_directories as u64);
+    span.record("selected_child_count", selected_paths.len() as u64);
+    if let Some(reason) = truncation {
+        span.record("truncation_reason", reason.as_str());
+        tracing::warn!(
+            target: "build_cache",
+            truncation_reason = reason.as_str(),
+            "build cache root discovery was truncated"
+        );
+    }
+    receiver_open
+}
+
+#[derive(Clone, Copy, Debug)]
+enum TruncationReason {
+    DirectoryLimit,
+    CandidateLimit,
+}
+
+impl TruncationReason {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::DirectoryLimit => "directory_limit",
+            Self::CandidateLimit => "candidate_limit",
+        }
+    }
+}
+
+/// A child of the current cache-setup span is entered on the blocking thread so per-repository
+/// diagnostics retain the correct trace parent. Dropping the receiver unblocks a pending send and
+/// cancels further scans.
+pub(super) fn candidate_receiver(
+    repositories: Vec<RepositoryCacheSource>,
+) -> mpsc::Receiver<CacheCandidate> {
+    let (sender, receiver) = mpsc::channel(DETECTION_CONCURRENCY);
+    let discovery_span = tracing::info_span!(
+        target: "build_cache",
+        "discover_cache_roots",
+        tags.cloud_agent = true,
+    );
+    tokio::task::spawn_blocking(move || {
+        let _guard = discovery_span.enter();
+        produce_candidates(repositories, sender);
+    });
+    receiver
+}
+
+pub(super) fn produce_candidates(
+    repositories: Vec<RepositoryCacheSource>,
+    sender: mpsc::Sender<CacheCandidate>,
+) {
+    let mut repositories = repositories
+        .into_iter()
+        .map(|source| (RepoCacheKey::derive(&source.identity), source))
+        .collect::<Vec<_>>();
+    repositories.sort();
+    for (key, source) in repositories {
+        if !produce_repository_candidates(key, source, &sender) {
+            return;
+        }
+    }
+}
+
+fn root_candidate(key: RepoCacheKey, source: RepositoryCacheSource) -> CacheCandidate {
+    CacheCandidate {
+        relative_cache_dir: PathBuf::from("repos").join(key.as_str()),
+        key: CandidateKey {
+            repo_key: key,
+            normalized_relative_path: None,
+        },
+        source,
+        stable_child_id: None,
+    }
+}
+
+fn child_candidate(
+    key: RepoCacheKey,
+    mut source: RepositoryCacheSource,
+    cwd: PathBuf,
+    normalized_relative_path: PathBuf,
+) -> CacheCandidate {
+    let stable_child_id = stable_child_id(&normalized_relative_path);
+    source.cwd = cwd;
+    CacheCandidate {
+        relative_cache_dir: PathBuf::from("repos")
+            .join(key.as_str())
+            .join("nested")
+            .join(&stable_child_id),
+        key: CandidateKey {
+            repo_key: key,
+            normalized_relative_path: Some(normalized_relative_path),
+        },
+        source,
+        stable_child_id: Some(stable_child_id),
+    }
+}
+
+/// Hashes normalized path bytes so child cache identities are stable across Linux and macOS.
+fn stable_child_id(normalized_relative_path: &Path) -> String {
+    hex::encode(Sha256::digest(
+        normalized_relative_path.as_os_str().as_encoded_bytes(),
+    ))
+}
+
+/// Returns a non-empty, relative UTF-8 path containing only normal components.
+fn normalize_relative_path(root: &Path, path: &Path) -> Option<PathBuf> {
+    let relative = path.strip_prefix(root).ok()?;
+    let mut normalized = PathBuf::new();
+    for component in relative.components() {
+        let Component::Normal(component) = component else {
+            return None;
+        };
+        component.to_str()?;
+        normalized.push(component);
+    }
+    if normalized.as_os_str().is_empty() {
+        return None;
+    }
+    Some(normalized)
+}
+
+/// Returns the root for the most-specific codebase marker matched by `entry`.
+fn find_candidate_for_entry(entry: &DirEntry, root: &Path) -> Option<PathBuf> {
+    if entry.file_type().is_dir()
+        && (entry.file_name() == "Tuist"
+            || entry
+                .file_name()
+                .to_str()
+                .is_some_and(|name| name.ends_with(".xcodeproj") || name.ends_with(".xcworkspace")))
+    {
+        return entry.path().parent().map(Path::to_path_buf);
+    }
+    if !entry.file_type().is_file() {
+        return None;
+    }
+
+    let relative = entry.path().strip_prefix(root).ok()?;
+    let components = relative
+        .components()
+        .map(|component| match component {
+            Component::Normal(component) => Some(component),
+            Component::Prefix(_)
+            | Component::RootDir
+            | Component::CurDir
+            | Component::ParentDir => None,
+        })
+        .collect::<Option<Vec<_>>>()?;
+    let relative_marker_length = CODEBASE_MARKER_PATHS
+        .iter()
+        .filter(|marker| {
+            components.len() >= marker.len()
+                && components[components.len() - marker.len()..]
+                    .iter()
+                    .zip(**marker)
+                    .all(|(component, marker)| component == marker)
+        })
+        .map(|marker| marker.len());
+    let direct_marker_length = CODEBASE_MARKER_FILENAMES
+        .iter()
+        .any(|marker| entry.file_name() == *marker)
+        .then_some(1);
+    let marker_length = relative_marker_length.chain(direct_marker_length).max()?;
+    let mut candidate = root.to_path_buf();
+    for component in &components[..components.len() - marker_length] {
+        candidate.push(component);
+    }
+    Some(candidate)
+}
+
+#[cfg(test)]
+#[path = "discovery_tests.rs"]
+mod tests;
diff --git a/crates/build_cache/src/discovery_tests.rs b/crates/build_cache/src/discovery_tests.rs
new file mode 100644
index 00000000000..c99670f3c81
--- /dev/null
+++ b/crates/build_cache/src/discovery_tests.rs
@@ -0,0 +1,278 @@
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use super::{MAX_CHILD_CANDIDATES, MAX_VISITED_DIRECTORIES, produce_candidates, stable_child_id};
+use crate::{RepoIdentity, RepositoryCacheSource};
+
+fn source(root: &Path) -> RepositoryCacheSource {
+    RepositoryCacheSource {
+        name: "warp/example".to_owned(),
+        identity: RepoIdentity::new("github.com", "warp", "example"),
+        cwd: root.to_path_buf(),
+    }
+}
+
+fn candidates(sources: Vec<RepositoryCacheSource>) -> Vec<super::CacheCandidate> {
+    let capacity = sources.len().max(1) * (MAX_CHILD_CANDIDATES + 1);
+    let (sender, mut receiver) = tokio::sync::mpsc::channel(capacity);
+    produce_candidates(sources, sender);
+    std::iter::from_fn(|| receiver.blocking_recv()).collect()
+}
+
+fn child_paths(root: &Path) -> Vec<PathBuf> {
+    let mut candidates = candidates(vec![source(root)]).into_iter();
+    let root = candidates.next().unwrap();
+    assert_eq!(root.key.normalized_relative_path, None);
+    candidates
+        .map(|candidate| candidate.key.normalized_relative_path.unwrap())
+        .collect()
+}
+
+fn touch(root: &Path, path: &str) {
+    let path = root.join(path);
+    fs::create_dir_all(path.parent().unwrap()).unwrap();
+    fs::write(path, "").unwrap();
+}
+
+#[test]
+fn direct_markers_select_their_containing_directories() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "rust/Cargo.toml");
+    touch(temp.path(), "javascript/package-lock.json");
+    touch(temp.path(), "gradle/build.gradle");
+
+    let paths = child_paths(temp.path());
+    assert_eq!(
+        paths,
+        [
+            PathBuf::from("gradle"),
+            PathBuf::from("javascript"),
+            PathBuf::from("rust"),
+        ]
+    );
+}
+
+#[test]
+fn relative_and_directory_markers_select_the_expected_ancestors() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "a/mise/config.toml");
+    touch(temp.path(), "b/.mise/config.toml");
+    touch(temp.path(), "c/.config/mise.toml");
+    touch(temp.path(), "d/.config/mise/config.toml");
+    fs::create_dir_all(temp.path().join("e/Tuist")).unwrap();
+    fs::create_dir_all(temp.path().join("f/App.xcodeproj")).unwrap();
+    fs::create_dir_all(temp.path().join("g/App.xcworkspace")).unwrap();
+
+    let paths = child_paths(temp.path());
+    assert_eq!(
+        paths,
+        [
+            PathBuf::from("a"),
+            PathBuf::from("b"),
+            PathBuf::from("c"),
+            PathBuf::from("d"),
+            PathBuf::from("e"),
+            PathBuf::from("f"),
+            PathBuf::from("g"),
+        ]
+    );
+}
+
+#[test]
+fn non_markers_and_ignored_subtrees_are_skipped() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "frontend/package.json");
+    touch(temp.path(), "backend/pyproject.toml");
+    touch(temp.path(), "gradle/settings.gradle");
+    touch(temp.path(), "kotlin/build.gradle.kts");
+    touch(temp.path(), "node_modules/nested/Cargo.toml");
+    touch(temp.path(), "target/nested/go.mod");
+    touch(temp.path(), "valid/Cargo.toml");
+
+    assert_eq!(child_paths(temp.path()), [PathBuf::from("valid")]);
+}
+
+#[test]
+fn multiple_markers_deduplicate_exact_roots_but_keep_nested_roots() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "project/Cargo.toml");
+    touch(temp.path(), "project/package-lock.json");
+    touch(temp.path(), "project/nested/go.mod");
+
+    assert_eq!(
+        child_paths(temp.path()),
+        [PathBuf::from("project"), PathBuf::from("project/nested")]
+    );
+}
+
+#[test]
+fn traversal_is_sorted_depth_first() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "z/Cargo.toml");
+    touch(temp.path(), "a/nested/Cargo.toml");
+    touch(temp.path(), "a/Cargo.toml");
+
+    assert_eq!(
+        child_paths(temp.path()),
+        [
+            PathBuf::from("a"),
+            PathBuf::from("a/nested"),
+            PathBuf::from("z"),
+        ]
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn symlinked_roots_and_entries_are_not_followed() {
+    use std::os::unix::fs::symlink;
+
+    let temp = tempfile::tempdir().unwrap();
+    let external = tempfile::tempdir().unwrap();
+    touch(external.path(), "project/Cargo.toml");
+    symlink(external.path().join("project"), temp.path().join("linked")).unwrap();
+
+    assert!(child_paths(temp.path()).is_empty());
+
+    let linked_root = temp.path().join("root-link");
+    symlink(external.path(), &linked_root).unwrap();
+    assert!(child_paths(&linked_root).is_empty());
+}
+
+#[test]
+fn walk_bound_includes_every_reached_candidate() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "one/two/three/four/.config/mise/config.toml");
+    touch(temp.path(), "one/two/three/four/five/Cargo.toml");
+
+    assert_eq!(
+        child_paths(temp.path()),
+        [
+            PathBuf::from("one/two/three/four"),
+            PathBuf::from("one/two/three/four/five"),
+        ]
+    );
+}
+
+#[test]
+fn child_limit_retains_root_and_earliest_children() {
+    let temp = tempfile::tempdir().unwrap();
+    for index in 0..MAX_CHILD_CANDIDATES + 1 {
+        touch(temp.path(), &format!("{index:02}/Cargo.toml"));
+    }
+    let candidates = candidates(vec![source(temp.path())]);
+
+    assert_eq!(candidates.len(), MAX_CHILD_CANDIDATES + 1);
+    assert_eq!(candidates[0].key.normalized_relative_path, None);
+    assert_eq!(
+        candidates
+            .last()
+            .unwrap()
+            .key
+            .normalized_relative_path
+            .as_deref(),
+        Some(Path::new("31"))
+    );
+}
+
+#[test]
+fn directory_limit_stops_before_later_marker() {
+    let temp = tempfile::tempdir().unwrap();
+    for index in 0..MAX_VISITED_DIRECTORIES {
+        fs::create_dir(temp.path().join(format!("{index:05}"))).unwrap();
+    }
+    touch(temp.path(), "zzzzz/Cargo.toml");
+
+    assert!(child_paths(temp.path()).is_empty());
+}
+
+#[test]
+fn stable_child_ids_hash_normalized_relative_paths() {
+    assert_eq!(
+        stable_child_id(Path::new("frontend/web")),
+        "4984839f9fe7d9730ec3fd45d7ededa43b0b5bfaf82f21666f7d9c25d1cf234c"
+    );
+    assert_eq!(stable_child_id(Path::new("frontend/web")).len(), 64);
+}
+
+#[test]
+fn repositories_and_roots_are_produced_in_canonical_order() {
+    let temp = tempfile::tempdir().unwrap();
+    let first = temp.path().join("first");
+    let second = temp.path().join("second");
+    fs::create_dir_all(&first).unwrap();
+    fs::create_dir_all(&second).unwrap();
+    touch(&first, "nested/Cargo.toml");
+    touch(&second, "nested/Cargo.toml");
+    let sources = vec![
+        RepositoryCacheSource {
+            name: "z/repo".to_owned(),
+            identity: RepoIdentity::new("github.com", "z", "repo"),
+            cwd: first,
+        },
+        RepositoryCacheSource {
+            name: "a/repo".to_owned(),
+            identity: RepoIdentity::new("github.com", "a", "repo"),
+            cwd: second,
+        },
+    ];
+    let candidates = candidates(sources);
+    let keys = candidates
+        .iter()
+        .map(|candidate| {
+            (
+                candidate.key.repo_key.clone(),
+                candidate.key.normalized_relative_path.clone(),
+            )
+        })
+        .collect::<Vec<_>>();
+
+    assert!(keys.windows(2).all(|pair| pair[0] <= pair[1]));
+}
+
+#[test]
+fn nested_cache_path_uses_stable_id_and_root_path_is_unchanged() {
+    let temp = tempfile::tempdir().unwrap();
+    touch(temp.path(), "frontend/Cargo.toml");
+    let mut candidates = candidates(vec![source(temp.path())]).into_iter();
+    let root = candidates.next().unwrap();
+    let child = candidates.next().unwrap();
+
+    assert_eq!(
+        root.relative_cache_dir,
+        PathBuf::from("repos").join(root.key.repo_key.as_str())
+    );
+    assert_eq!(
+        child.relative_cache_dir,
+        PathBuf::from("repos")
+            .join(child.key.repo_key.as_str())
+            .join("nested")
+            .join(child.stable_child_id.unwrap())
+    );
+}
+
+#[test]
+fn dropping_bounded_receiver_stops_blocking_producer() {
+    let temp = tempfile::tempdir().unwrap();
+    for index in 0..MAX_CHILD_CANDIDATES {
+        touch(temp.path(), &format!("{index:02}/Cargo.toml"));
+    }
+    let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
+    let sources = vec![source(temp.path())];
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_time()
+        .build()
+        .unwrap();
+
+    runtime.block_on(async move {
+        let producer = tokio::task::spawn_blocking(move || produce_candidates(sources, sender));
+        assert!(receiver.recv().await.is_some());
+        drop(receiver);
+
+        tokio::time::timeout(Duration::from_secs(1), producer)
+            .await
+            .unwrap()
+            .unwrap();
+    });
+}
diff --git a/crates/build_cache/src/lib.rs b/crates/build_cache/src/lib.rs
index 4e95e7e0421..ed6d162b0eb 100644
--- a/crates/build_cache/src/lib.rs
+++ b/crates/build_cache/src/lib.rs
@@ -23,11 +23,13 @@ use std::fmt;
 use std::future::Future;
 use std::io::ErrorKind;
 use std::path::{Component, Path, PathBuf};
+use std::sync::Arc;
 use std::time::Duration;
 
 use async_io::Timer;
 use command::Stdio;
 use command::r#async::Command;
+use futures::stream::{self, StreamExt as _};
 use futures_lite::future;
 use is_executable::IsExecutable as _;
 use itertools::Itertools;
@@ -35,9 +37,13 @@ use sha2::{Digest, Sha256};
 use warp_core::safe_info;
 use warp_errors::{ErrorExt, register_error};
 
+mod discovery;
 pub mod spacectl;
 
-use spacectl::{MountResponse, run_spacectl_mount};
+#[cfg(test)]
+use discovery::produce_candidates;
+use discovery::{CacheCandidate, CandidateKey, DETECTION_CONCURRENCY, candidate_receiver};
+use spacectl::{MountContext, MountResponse, run_spacectl_mount};
 
 const SPACECTL_TIMEOUT: Duration = Duration::from_secs(60);
 const MAX_CAPTURED_STDERR_BYTES: usize = 4 * 1024;
@@ -180,7 +186,8 @@ impl CacheSetupPlan {
     /// Validate that a cache plan is valid. A valid plan:
     /// * Contains one or more cache configurations
     /// * Ends in a globally-scoped cache configuration
-    /// * Lists each repository exactly once, in order by cache key
+    /// * Lists repository configurations in order by cache key
+    /// * Uses unique repository working directories and cache locations
     /// * Only uses safe cache locations within the cache volume (no absolute paths, `..`, or `.` components)
     pub fn validate(&self) -> Result<(), PlanInvariantError> {
         let Some((global, repositories)) = self.configurations.split_last() else {
@@ -191,6 +198,8 @@ impl CacheSetupPlan {
         }
 
         let mut previous_key: Option<&RepoCacheKey> = None;
+        let mut repository_cwds = BTreeSet::new();
+        let mut repository_cache_dirs = BTreeSet::new();
         for configuration in repositories {
             let CacheScope::Repository { key, .. } = &configuration.scope else {
                 return Err(PlanInvariantError);
@@ -199,6 +208,11 @@ impl CacheSetupPlan {
                 return Err(PlanInvariantError);
             }
             previous_key = Some(key);
+            if !repository_cwds.insert(&configuration.cwd)
+                || !repository_cache_dirs.insert(&configuration.relative_cache_dir)
+            {
+                return Err(PlanInvariantError);
+            }
         }
 
         for configuration in &self.configurations {
@@ -248,6 +262,8 @@ pub enum CacheSetupError {
     Timeout,
     #[error("failed to export build cache environment variables")]
     EnvExportFailed,
+    #[error("cache setup plan invariant violated")]
+    PlanInvariantFailed,
 }
 
 impl CacheSetupError {
@@ -259,6 +275,7 @@ impl CacheSetupError {
             Self::JsonParseFailed => "json_parse_failed",
             Self::Timeout => "timeout",
             Self::EnvExportFailed => "env_export_failed",
+            Self::PlanInvariantFailed => "plan_invariant_failed",
         }
     }
 
@@ -269,7 +286,8 @@ impl CacheSetupError {
             | Self::SpawnFailed
             | Self::JsonParseFailed
             | Self::Timeout
-            | Self::EnvExportFailed => None,
+            | Self::EnvExportFailed
+            | Self::PlanInvariantFailed => None,
         }
     }
 }
@@ -277,7 +295,7 @@ impl CacheSetupError {
 impl ErrorExt for CacheSetupError {
     fn is_actionable(&self) -> bool {
         match self {
-            Self::JsonParseFailed | Self::EnvExportFailed => true,
+            Self::JsonParseFailed | Self::EnvExportFailed | Self::PlanInvariantFailed => true,
             Self::RootCreationFailed
             | Self::SpawnFailed
             | Self::NonzeroExit { .. }
@@ -332,11 +350,19 @@ impl CacheSetupReport {
 
 #[derive(Clone)]
 struct DetectedCacheModes {
+    order: CandidateKey,
     source: RepositoryCacheSource,
-    key: RepoCacheKey,
+    relative_cache_dir: PathBuf,
     modes: Vec<String>,
 }
 
+struct CandidateDetection {
+    order: CandidateKey,
+    invocation: CachePreparationReport,
+    detected: Option<DetectedCacheModes>,
+    scheduled: bool,
+}
+
 /// Calculates cache modes corresponding to global tools like package managers, which are not
 /// detected from an individual repo.
 pub fn global_cache_modes() -> Vec<String> {
@@ -363,12 +389,9 @@ fn has_command(command: &str) -> bool {
 /// target and then transfer ownership of that target directory to the current effective user.
 /// Intermediate directories are not chowned, and any unavailable or unsuccessful fallback
 /// operation degrades to [`CacheSetupError::RootCreationFailed`].
-async fn create_cache_dir_all<F, Fut>(
-    path: &Path,
-    run_command: &mut F,
-) -> Result<(), CacheSetupError>
+async fn create_cache_dir_all<F, Fut>(path: &Path, run_command: &F) -> Result<(), CacheSetupError>
 where
-    F: FnMut(Command) -> Fut,
+    F: Fn(Command) -> Fut,
     Fut: Future<Output = Result<Vec<u8>, CacheSetupError>>,
 {
     if path.is_dir() {
@@ -496,76 +519,80 @@ fn bounded_stderr(stderr: &[u8]) -> String {
 ///
 /// This should only be called once per sandbox, as it modifies shared filesystem locations.
 /// The calling process need not run with superuser privileges, but the implementation may
-/// escalate privileges with `sudo` or similar.
-#[tracing::instrument(name = "setup_caches", skip_all, fields(tags.cloud_agent = true))]
+/// escalate privileges with `sudo` or similar. It must be called from a Tokio runtime because
+/// repository discovery uses Tokio's blocking pool.
+#[tracing::instrument(
+    name = "setup_caches",
+    skip_all,
+    fields(
+        tags.cloud_agent = true,
+        detection_limit = DETECTION_CONCURRENCY,
+        total_scheduled_detects = tracing::field::Empty,
+    )
+)]
 pub async fn setup_cache<F, Fut>(
     cache_root: PathBuf,
     repositories: Vec<RepositoryCacheSource>,
     additional_global_modes: Vec<String>,
-    mut run_command: F,
+    run_command: F,
 ) -> CacheSetupReport
 where
-    F: FnMut(Command) -> Fut,
+    F: Fn(Command) -> Fut,
     Fut: Future<Output = Result<Vec<u8>, CacheSetupError>>,
 {
     let mut report = CacheSetupReport::default();
-    let mut keyed_repositories: Vec<_> = repositories
-        .into_iter()
-        .map(|source| {
-            let key = RepoCacheKey::derive(&source.identity);
-            (key, source)
-        })
-        .collect();
-    keyed_repositories.sort();
-
-    // Step 1: Detect the cache modes that apply to each repository. A mode corresponds to a tool
-    // or language runtime, such as `apt-get` or Swift.
-    let mut detected_modes = Vec::new();
-    for (key, source) in keyed_repositories {
-        let relative_cache_dir = PathBuf::from("repos").join(key.as_str());
-        let configuration_root = cache_root.join(&relative_cache_dir);
-        let scope = CacheScope::Repository {
-            name: source.name.clone(),
-            key: key.clone(),
-        };
-
-        // We create the scoped cache directory here, as `spacectl` fails if it doesn't exist.
-        if create_cache_dir_all(&configuration_root, &mut run_command)
-            .await
-            .is_err()
-        {
-            report.invocations.push(failed_invocation(
-                scope,
-                Vec::new(),
-                relative_cache_dir,
-                CacheSetupError::RootCreationFailed,
-                Duration::ZERO,
-            ));
-            continue;
+    let run_command = Arc::new(run_command);
+
+    let prepare_run_command = Arc::clone(&run_command);
+    let prepare_cache_root = cache_root.clone();
+    // Preparing inside the source stream prevents permission fallbacks from racing while still
+    // allowing each preparation to overlap dry-run detections already in flight.
+    let candidates = stream::unfold(candidate_receiver(repositories), move |mut receiver| {
+        let run_command = Arc::clone(&prepare_run_command);
+        let cache_root = prepare_cache_root.clone();
+        async move {
+            let candidate = receiver.recv().await?;
+            let configuration_root = cache_root.join(&candidate.relative_cache_dir);
+            let preparation_error = create_cache_dir_all(&configuration_root, run_command.as_ref())
+                .await
+                .err();
+            Some(((candidate, configuration_root, preparation_error), receiver))
         }
-
-        // Run `spacectl` in dry-run mode, so that it detects all relevant cache modes.
-        let invocation = run_spacectl_mount(
-            scope,
-            Vec::new(),
-            true,
-            relative_cache_dir,
-            &configuration_root,
-            &source.cwd,
-            &mut run_command,
-        )
-        .await;
-        if let Some(response) = &invocation.response {
-            let modes = canonical_modes(response.input.modes.clone());
-            if !modes.is_empty() {
-                detected_modes.push(DetectedCacheModes { source, key, modes });
+    });
+    let detect_run_command = Arc::clone(&run_command);
+    let mut detection_results = candidates
+        .map(move |(candidate, configuration_root, preparation_error)| {
+            let run_command = Arc::clone(&detect_run_command);
+            async move {
+                detect_candidate(
+                    candidate,
+                    configuration_root,
+                    preparation_error,
+                    run_command.as_ref(),
+                )
+                .await
             }
+        })
+        .buffer_unordered(DETECTION_CONCURRENCY)
+        .collect::<Vec<_>>()
+        .await;
+    tracing::Span::current().record(
+        "total_scheduled_detects",
+        detection_results
+            .iter()
+            .filter(|result| result.scheduled)
+            .count() as u64,
+    );
+    detection_results.sort_by(|left, right| left.order.cmp(&right.order));
+    let mut detected_modes = Vec::new();
+    for result in detection_results {
+        if let Some(detected) = result.detected {
+            detected_modes.push(detected);
         }
-        report.invocations.push(invocation);
+        report.invocations.push(result.invocation);
     }
 
-    // Step 2: Given the per-repository results, construct the cache plan. This tells us which
-    // caches to set up, and in what order.
+    // Canonical ordering keeps detection timing from changing the resulting mount plan.
     let plan = match construct_plan(cache_root, detected_modes, additional_global_modes) {
         Ok(Some(plan)) => plan,
         Ok(None) => return report,
@@ -582,13 +609,12 @@ where
         }
     };
 
-    // Step 3: Run `spacectl cache mount` for real, setting up all the cache mounts.
+    // Real mounts remain serial because cache destinations may overlap across scopes.
     let mut repository_env = BTreeMap::new();
     let mut global_env = None;
     for configuration in &plan.configurations {
         let configuration_root = plan.cache_root.join(&configuration.relative_cache_dir);
-        // All repo-scoped cache roots should already exist. However, we still need to create the global root.
-        let invocation = if create_cache_dir_all(&configuration_root, &mut run_command)
+        let invocation = if create_cache_dir_all(&configuration_root, run_command.as_ref())
             .await
             .is_err()
         {
@@ -608,10 +634,13 @@ where
                 configuration.scope.clone(),
                 configuration.modes.clone(),
                 false,
-                configuration.relative_cache_dir.clone(),
-                &configuration_root,
-                &configuration.cwd,
-                &mut run_command,
+                MountContext {
+                    relative_cache_dir: configuration.relative_cache_dir.clone(),
+                    cache_root: configuration_root,
+                    cwd: configuration.cwd.clone(),
+                    stable_child_id: String::new(),
+                },
+                run_command.as_ref(),
             )
             .await
         };
@@ -637,12 +666,7 @@ where
         report.invocations.push(invocation);
     }
 
-    // Step 4: Construct the merged environment variable map. If multiple repo-scoped cache
-    // configurations set the same environment variable, we'll already have deduplicated them
-    // (with last-repo-wins semantics) above. Here, we prefer using the globally-scoped set of
-    // environment variables, but fall back to the combined set of repository environment variables.
-    // We don't need to merge the two - the global cache configuration includes all modes set
-    // by per-repo configurations, so it should have all the same variables.
+    // The global response covers all detected modes; repository values are only a fallback.
     report.add_envs = global_env.unwrap_or(repository_env);
     report.add_envs.retain(|name, _| {
         if is_valid_env_name(name) {
@@ -659,6 +683,69 @@ where
     report
 }
 
+async fn detect_candidate<F, Fut>(
+    candidate: CacheCandidate,
+    configuration_root: PathBuf,
+    preparation_error: Option<CacheSetupError>,
+    run_command: &F,
+) -> CandidateDetection
+where
+    F: Fn(Command) -> Fut,
+    Fut: Future<Output = Result<Vec<u8>, CacheSetupError>>,
+{
+    let scope = CacheScope::Repository {
+        name: candidate.source.name.clone(),
+        key: candidate.key.repo_key.clone(),
+    };
+    if let Some(error) = preparation_error {
+        return CandidateDetection {
+            order: candidate.key,
+            invocation: failed_invocation(
+                scope,
+                Vec::new(),
+                candidate.relative_cache_dir,
+                error,
+                Duration::ZERO,
+            ),
+            detected: None,
+            scheduled: false,
+        };
+    }
+
+    let invocation = run_spacectl_mount(
+        scope,
+        Vec::new(),
+        true,
+        MountContext {
+            relative_cache_dir: candidate.relative_cache_dir.clone(),
+            cache_root: configuration_root,
+            cwd: candidate.source.cwd.clone(),
+            stable_child_id: candidate.stable_child_id.clone().unwrap_or_default(),
+        },
+        run_command,
+    )
+    .await;
+    let detected = invocation.response.as_ref().and_then(|response| {
+        let modes = canonical_modes(response.input.modes.clone());
+        if modes.is_empty() {
+            None
+        } else {
+            Some(DetectedCacheModes {
+                order: candidate.key.clone(),
+                source: candidate.source,
+                relative_cache_dir: candidate.relative_cache_dir,
+                modes,
+            })
+        }
+    });
+    CandidateDetection {
+        order: candidate.key,
+        invocation,
+        detected,
+        scheduled: true,
+    }
+}
+
 /// Construct a plan for setting up build caches on the current system. This requires:
 /// - Analysis of the toolchains used in each repository (`detections`)
 /// - System-level toolchains such as package managers
@@ -694,9 +781,10 @@ fn construct_plan(
         "additional_global_modes",
         additional_global_modes.iter().join(", "),
     );
+    detections.sort_by(|left, right| left.order.cmp(&right.order));
     for detection in &mut detections {
         detection.modes = canonical_modes(std::mem::take(&mut detection.modes));
-        tracing::info!(modes = ?detection.modes, repo_key = %detection.key, "Adding detected cache modes");
+        tracing::info!(modes = ?detection.modes, repo_key = %detection.order.repo_key, "Adding detected cache modes");
     }
     let mut global_modes = BTreeSet::new();
     for detection in &detections {
@@ -722,19 +810,13 @@ fn construct_plan(
         .map(|detection| CacheConfiguration {
             scope: CacheScope::Repository {
                 name: detection.source.name,
-                key: detection.key.clone(),
+                key: detection.order.repo_key,
             },
             cwd: detection.source.cwd,
-            relative_cache_dir: PathBuf::from("repos").join(detection.key.as_str()),
+            relative_cache_dir: detection.relative_cache_dir,
             modes: detection.modes,
         })
         .collect::<Vec<_>>();
-    configurations.sort_by(|left, right| {
-        left.scope
-            .repo_key()
-            .expect("repository configuration")
-            .cmp(right.scope.repo_key().expect("repository configuration"))
-    });
     tracing::Span::current().record("resolved_modes", global_modes.iter().join(", "));
     configurations.push(CacheConfiguration {
         scope: CacheScope::Global,
@@ -744,7 +826,7 @@ fn construct_plan(
     });
     CacheSetupPlan::try_new(cache_root, configurations)
         .map(Some)
-        .map_err(|_| CacheSetupError::RootCreationFailed)
+        .map_err(|_| CacheSetupError::PlanInvariantFailed)
 }
 
 /// Create a temporary scratch directory for setting up the global cache scope.
diff --git a/crates/build_cache/src/lib_tests.rs b/crates/build_cache/src/lib_tests.rs
index a68c650fb88..98bbeea2f26 100644
--- a/crates/build_cache/src/lib_tests.rs
+++ b/crates/build_cache/src/lib_tests.rs
@@ -1,21 +1,23 @@
-use std::cell::RefCell;
 use std::collections::{BTreeMap, VecDeque};
 use std::ffi::OsString;
 use std::fs;
-use std::path::Path;
-use std::rc::Rc;
+use std::future::Future;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
+use async_io::Timer;
 use command::r#async::Command;
-use futures::executor::block_on;
 #[cfg(unix)]
 use instant::Instant;
 use warp_errors::ErrorExt as _;
 
 use super::{
-    CacheScope, CacheSetupError, DetectedCacheModes, RepoCacheKey, RepoIdentity,
-    RepositoryCacheSource, aggregate_mode_stats, construct_plan, create_retained_scratch_directory,
-    is_valid_env_name, run_command_with_timeout, setup_cache,
+    CacheConfiguration, CacheScope, CacheSetupError, CacheSetupPlan, CandidateKey,
+    DetectedCacheModes, RepoCacheKey, RepoIdentity, RepositoryCacheSource, aggregate_mode_stats,
+    construct_plan, create_retained_scratch_directory, is_valid_env_name, produce_candidates,
+    run_command_with_timeout, setup_cache,
 };
 #[cfg(unix)]
 use super::{create_cache_dir_all, current_owner};
@@ -36,8 +38,13 @@ fn source(root: &Path, host: &str, owner: &str, repo: &str) -> RepositoryCacheSo
 }
 
 fn detection(source: RepositoryCacheSource, modes: &[&str]) -> DetectedCacheModes {
+    let key = RepoCacheKey::derive(&source.identity);
     DetectedCacheModes {
-        key: RepoCacheKey::derive(&source.identity),
+        order: CandidateKey {
+            repo_key: key.clone(),
+            normalized_relative_path: None,
+        },
+        relative_cache_dir: Path::new("repos").join(key.as_str()),
         source,
         modes: modes.iter().map(ToString::to_string).collect(),
     }
@@ -68,6 +75,14 @@ fn response(modes: &[&str], envs: &[(&str, &str)], mounts: &[(&str, bool)]) -> V
 fn command_args(command: &Command) -> Vec<OsString> {
     command.get_args().map(ToOwned::to_owned).collect()
 }
+
+fn block_on<F: Future>(future: F) -> F::Output {
+    tokio::runtime::Builder::new_multi_thread()
+        .enable_time()
+        .build()
+        .unwrap()
+        .block_on(future)
+}
 #[cfg(unix)]
 #[test]
 fn permission_denied_cache_directory_uses_noninteractive_sudo_mkdir_and_chown() {
@@ -82,18 +97,18 @@ fn permission_denied_cache_directory_uses_noninteractive_sudo_mkdir_and_chown()
     fs::create_dir(&locked).unwrap();
     fs::set_permissions(&locked, fs::Permissions::from_mode(0o500)).unwrap();
     let target = locked.join("child").join("grandchild");
-    let commands = Rc::new(RefCell::new(Vec::new()));
-    let result = block_on(create_cache_dir_all(&target, &mut {
-        let commands = Rc::clone(&commands);
+    let commands = Arc::new(Mutex::new(Vec::new()));
+    let result = block_on(create_cache_dir_all(&target, &{
+        let commands = Arc::clone(&commands);
         move |command| {
-            commands.borrow_mut().push(command_args(&command));
+            commands.lock().unwrap().push(command_args(&command));
             futures::future::ready(Ok(Vec::new()))
         }
     }));
     fs::set_permissions(&locked, fs::Permissions::from_mode(0o700)).unwrap();
 
     assert_eq!(result, Ok(()));
-    let commands = commands.borrow();
+    let commands = commands.lock().unwrap();
     assert_eq!(commands.len(), 2);
     assert_eq!(
         commands[0],
@@ -298,16 +313,16 @@ fn json_parse_failure_is_classified_and_does_not_abort_later_repos() {
         source(temp.path(), "github.com", "warp", "one"),
         source(temp.path(), "github.com", "warp", "two"),
     ];
-    let calls = Rc::new(RefCell::new(0usize));
+    let calls = Arc::new(Mutex::new(0usize));
     let report = block_on(setup_cache(
         temp.path().join("cache"),
         repositories,
         Vec::new(),
         {
-            let calls = Rc::clone(&calls);
+            let calls = Arc::clone(&calls);
             move |_| {
                 let call = {
-                    let mut count = calls.borrow_mut();
+                    let mut count = calls.lock().unwrap();
                     *count += 1;
                     *count
                 };
@@ -319,7 +334,7 @@ fn json_parse_failure_is_classified_and_does_not_abort_later_repos() {
             }
         },
     ));
-    assert_eq!(*calls.borrow(), 4);
+    assert_eq!(*calls.lock().unwrap(), 4);
     assert_eq!(
         report.invocations[0].error,
         Some(CacheSetupError::JsonParseFailed)
@@ -330,16 +345,16 @@ fn json_parse_failure_is_classified_and_does_not_abort_later_repos() {
 #[test]
 fn destructive_execution_uses_resolved_modes_without_redetection() {
     let temp = tempfile::tempdir().unwrap();
-    let commands = Rc::new(RefCell::new(Vec::new()));
+    let commands = Arc::new(Mutex::new(Vec::new()));
     let report = block_on(setup_cache(
         temp.path().join("cache"),
         vec![source(temp.path(), "github.com", "warp", "client")],
         Vec::new(),
         {
-            let commands = Rc::clone(&commands);
+            let commands = Arc::clone(&commands);
             move |command| {
                 let detect = is_detect(&command);
-                commands.borrow_mut().push(command_args(&command));
+                commands.lock().unwrap().push(command_args(&command));
                 futures::future::ready(Ok(if detect {
                     response(&["go", "cargo", "go"], &[], &[])
                 } else {
@@ -349,7 +364,7 @@ fn destructive_execution_uses_resolved_modes_without_redetection() {
         },
     ));
     assert!(report.plan.is_some());
-    let commands = commands.borrow();
+    let commands = commands.lock().unwrap();
     assert_eq!(commands.len(), 3);
     for args in &commands[1..] {
         assert!(args.iter().any(|arg| arg == "--mode=cargo,go"));
@@ -361,7 +376,7 @@ fn destructive_execution_uses_resolved_modes_without_redetection() {
 #[test]
 fn repo_failure_continues_and_global_still_executes() {
     let temp = tempfile::tempdir().unwrap();
-    let destructive_calls = Rc::new(RefCell::new(0));
+    let destructive_calls = Arc::new(Mutex::new(0));
     let report = block_on(setup_cache(
         temp.path().join("cache"),
         vec![
@@ -370,12 +385,12 @@ fn repo_failure_continues_and_global_still_executes() {
         ],
         Vec::new(),
         {
-            let destructive_calls = Rc::clone(&destructive_calls);
+            let destructive_calls = Arc::clone(&destructive_calls);
             move |command| {
                 if is_detect(&command) {
                     return futures::future::ready(Ok(response(&["cargo"], &[], &[])));
                 }
-                let mut calls = destructive_calls.borrow_mut();
+                let mut calls = destructive_calls.lock().unwrap();
                 *calls += 1;
                 if *calls == 1 {
                     futures::future::ready(Err(CacheSetupError::NonzeroExit {
@@ -388,7 +403,7 @@ fn repo_failure_continues_and_global_still_executes() {
             }
         },
     ));
-    assert_eq!(*destructive_calls.borrow(), 3);
+    assert_eq!(*destructive_calls.lock().unwrap(), 3);
     assert!(
         report
             .invocations
@@ -400,7 +415,7 @@ fn repo_failure_continues_and_global_still_executes() {
 #[test]
 fn spacectl_calls_are_bounded_by_two_repos_plus_one_global() {
     let temp = tempfile::tempdir().unwrap();
-    let calls = Rc::new(RefCell::new(0));
+    let calls = Arc::new(Mutex::new(0));
     let report = block_on(setup_cache(
         temp.path().join("cache"),
         vec![
@@ -409,16 +424,197 @@ fn spacectl_calls_are_bounded_by_two_repos_plus_one_global() {
         ],
         Vec::new(),
         {
-            let calls = Rc::clone(&calls);
+            let calls = Arc::clone(&calls);
             move |_| {
-                *calls.borrow_mut() += 1;
+                *calls.lock().unwrap() += 1;
                 futures::future::ready(Ok(response(&["cargo"], &[], &[])))
             }
         },
     ));
-    assert_eq!(*calls.borrow(), 5);
+    assert_eq!(*calls.lock().unwrap(), 5);
     assert_eq!(report.invocations.len(), 5);
 }
+#[test]
+fn nested_roots_share_one_bounded_detection_pool_and_mount_serially() {
+    let temp = tempfile::tempdir().unwrap();
+    let repositories = vec![
+        source(temp.path(), "github.com", "warp", "client-a"),
+        source(temp.path(), "github.com", "warp", "client-b"),
+    ];
+    for repository in &repositories {
+        for index in 0..6 {
+            let child = repository.cwd.join(format!("child-{index:02}"));
+            fs::create_dir_all(&child).unwrap();
+            fs::write(child.join("Cargo.toml"), "").unwrap();
+        }
+    }
+    let (sender, mut receiver) = tokio::sync::mpsc::channel(64);
+    produce_candidates(repositories.clone(), sender);
+    let mut expected_candidates =
+        std::iter::from_fn(|| receiver.blocking_recv()).collect::<Vec<_>>();
+    expected_candidates.sort_by(|left, right| left.key.cmp(&right.key));
+    let expected_detection_order = expected_candidates
+        .iter()
+        .map(|candidate| candidate.relative_cache_dir.clone())
+        .collect::<Vec<_>>();
+    let detect_delays = expected_candidates
+        .iter()
+        .rev()
+        .enumerate()
+        .map(|(index, candidate)| (candidate.source.cwd.clone(), index as u64 + 1))
+        .collect::<BTreeMap<_, _>>();
+    let active_detects = Arc::new(AtomicUsize::new(0));
+    let max_active_detects = Arc::new(AtomicUsize::new(0));
+    let detect_count = Arc::new(AtomicUsize::new(0));
+    let active_mounts = Arc::new(AtomicUsize::new(0));
+    let max_active_mounts = Arc::new(AtomicUsize::new(0));
+    let mount_order = Arc::new(Mutex::new(Vec::new()));
+    let report = block_on(setup_cache(
+        temp.path().join("cache"),
+        repositories,
+        Vec::new(),
+        {
+            let active_detects = Arc::clone(&active_detects);
+            let max_active_detects = Arc::clone(&max_active_detects);
+            let detect_count = Arc::clone(&detect_count);
+            let active_mounts = Arc::clone(&active_mounts);
+            let max_active_mounts = Arc::clone(&max_active_mounts);
+            let mount_order = Arc::clone(&mount_order);
+            move |command| {
+                let active_detects = Arc::clone(&active_detects);
+                let max_active_detects = Arc::clone(&max_active_detects);
+                let detect_count = Arc::clone(&detect_count);
+                let active_mounts = Arc::clone(&active_mounts);
+                let max_active_mounts = Arc::clone(&max_active_mounts);
+                let mount_order = Arc::clone(&mount_order);
+                let delay = command
+                    .get_current_dir()
+                    .and_then(|cwd| detect_delays.get(cwd))
+                    .copied()
+                    .unwrap_or_default();
+                async move {
+                    if is_detect(&command) {
+                        detect_count.fetch_add(1, Ordering::SeqCst);
+                        let active = active_detects.fetch_add(1, Ordering::SeqCst) + 1;
+                        max_active_detects.fetch_max(active, Ordering::SeqCst);
+                        Timer::after(Duration::from_millis(delay)).await;
+                        active_detects.fetch_sub(1, Ordering::SeqCst);
+                    } else {
+                        let active = active_mounts.fetch_add(1, Ordering::SeqCst) + 1;
+                        max_active_mounts.fetch_max(active, Ordering::SeqCst);
+                        Timer::after(Duration::from_millis(1)).await;
+                        mount_order
+                            .lock()
+                            .unwrap()
+                            .push(command.get_current_dir().unwrap().to_path_buf());
+                        active_mounts.fetch_sub(1, Ordering::SeqCst);
+                    }
+                    Ok(response(&["cargo"], &[], &[]))
+                }
+            }
+        },
+    ));
+
+    assert_eq!(detect_count.load(Ordering::SeqCst), 14);
+    assert!(max_active_detects.load(Ordering::SeqCst) > 1);
+    assert!(max_active_detects.load(Ordering::SeqCst) <= 8);
+    assert_eq!(max_active_mounts.load(Ordering::SeqCst), 1);
+    assert_eq!(
+        report.invocations[..expected_detection_order.len()]
+            .iter()
+            .map(|invocation| invocation.relative_cache_dir.clone())
+            .collect::<Vec<_>>(),
+        expected_detection_order
+    );
+    let plan = report.plan.as_ref().unwrap();
+    assert_eq!(plan.configurations.len(), 15);
+    assert_eq!(
+        plan.configurations
+            .iter()
+            .filter(|configuration| matches!(configuration.scope, CacheScope::Repository { .. }))
+            .count(),
+        14
+    );
+    let mount_order = mount_order.lock().unwrap();
+    assert_eq!(
+        mount_order.as_slice(),
+        plan.configurations
+            .iter()
+            .map(|configuration| configuration.cwd.clone())
+            .collect::<Vec<_>>()
+    );
+}
+
+#[test]
+fn plan_accepts_repeated_repo_keys_with_distinct_roots_and_cache_paths() {
+    let temp = tempfile::tempdir().unwrap();
+    let root = source(temp.path(), "github.com", "warp", "client");
+    let mut child = root.clone();
+    child.cwd = root.cwd.join("nested");
+    fs::create_dir_all(&child.cwd).unwrap();
+    let key = RepoCacheKey::derive(&root.identity);
+    let plan = CacheSetupPlan::try_new(
+        temp.path().join("cache"),
+        vec![
+            CacheConfiguration {
+                scope: CacheScope::Repository {
+                    name: root.name,
+                    key: key.clone(),
+                },
+                cwd: root.cwd,
+                relative_cache_dir: Path::new("repos").join(key.as_str()),
+                modes: vec!["cargo".to_owned()],
+            },
+            CacheConfiguration {
+                scope: CacheScope::Repository {
+                    name: child.name,
+                    key,
+                },
+                cwd: child.cwd,
+                relative_cache_dir: PathBuf::from("repos/key/nested/id"),
+                modes: vec!["cargo".to_owned()],
+            },
+            CacheConfiguration {
+                scope: CacheScope::Global,
+                cwd: temp.path().join("scratch"),
+                relative_cache_dir: PathBuf::from("shared"),
+                modes: vec!["cargo".to_owned()],
+            },
+        ],
+    );
+
+    assert!(plan.is_ok());
+}
+
+#[test]
+fn plan_rejects_duplicate_repository_working_or_cache_directories() {
+    let temp = tempfile::tempdir().unwrap();
+    let repo = source(temp.path(), "github.com", "warp", "client");
+    let key = RepoCacheKey::derive(&repo.identity);
+    let configuration = CacheConfiguration {
+        scope: CacheScope::Repository {
+            name: repo.name,
+            key,
+        },
+        cwd: repo.cwd,
+        relative_cache_dir: PathBuf::from("repos/key"),
+        modes: vec!["cargo".to_owned()],
+    };
+    let global = CacheConfiguration {
+        scope: CacheScope::Global,
+        cwd: temp.path().join("scratch"),
+        relative_cache_dir: PathBuf::from("shared"),
+        modes: vec!["cargo".to_owned()],
+    };
+
+    assert!(
+        CacheSetupPlan::try_new(
+            temp.path().join("cache"),
+            vec![configuration.clone(), configuration, global],
+        )
+        .is_err()
+    );
+}
 
 #[test]
 fn shared_success_replaces_complete_repo_env_overlay() {
@@ -664,6 +860,7 @@ fn cache_setup_error_variants_have_expected_is_actionable_classification() {
     assert!(!CacheSetupError::Timeout.is_actionable());
     assert!(CacheSetupError::JsonParseFailed.is_actionable());
     assert!(CacheSetupError::EnvExportFailed.is_actionable());
+    assert!(CacheSetupError::PlanInvariantFailed.is_actionable());
 }
 
 #[test]
@@ -711,7 +908,7 @@ fn failure_categories_are_preserved() {
 #[test]
 fn queued_executor_can_return_each_failure_category() {
     let temp = tempfile::tempdir().unwrap();
-    let queue = Rc::new(RefCell::new(VecDeque::from([
+    let queue = Arc::new(Mutex::new(VecDeque::from([
         Err(CacheSetupError::JsonParseFailed),
         Err(CacheSetupError::Timeout),
     ])));
@@ -723,8 +920,8 @@ fn queued_executor_can_return_each_failure_category() {
         ],
         Vec::new(),
         {
-            let queue = Rc::clone(&queue);
-            move |_| futures::future::ready(queue.borrow_mut().pop_front().unwrap())
+            let queue = Arc::clone(&queue);
+            move |_| futures::future::ready(queue.lock().unwrap().pop_front().unwrap())
         },
     ));
     assert_eq!(report.invocations.len(), 2);
diff --git a/crates/build_cache/src/spacectl.rs b/crates/build_cache/src/spacectl.rs
index fb76b4710e4..5aad23fcbe0 100644
--- a/crates/build_cache/src/spacectl.rs
+++ b/crates/build_cache/src/spacectl.rs
@@ -59,6 +59,12 @@ pub struct DiskUsage {
     pub total: String,
     pub used: String,
 }
+pub(super) struct MountContext {
+    pub relative_cache_dir: PathBuf,
+    pub cache_root: PathBuf,
+    pub cwd: PathBuf,
+    pub stable_child_id: String,
+}
 
 /// Construct a `spacectl` command for detecting all cache modes that apply to
 /// `cwd`. Currently, this uses `spacectl cache mount`, though we could use
@@ -102,7 +108,8 @@ fn mount_command(cache_root: &Path, cwd: &Path, modes: &[String]) -> Command {
         repo_key = scope.repo_key().map(RepoCacheKey::as_str).unwrap_or(""),
         modes = tracing::field::Empty,
         dry_run,
-        relative_cache_dir = %relative_cache_dir.display(),
+        relative_cache_dir = %context.relative_cache_dir.display(),
+        stable_child_id = context.stable_child_id.as_str(),
         duration_ms = tracing::field::Empty,
         disk_usage_total = tracing::field::Empty,
         disk_usage_used = tracing::field::Empty,
@@ -117,19 +124,17 @@ pub(super) async fn run_spacectl_mount<F, Fut>(
     scope: CacheScope,
     modes: Vec<String>,
     dry_run: bool,
-    relative_cache_dir: PathBuf,
-    cache_root: &Path,
-    cwd: &Path,
-    run_command: &mut F,
+    context: MountContext,
+    run_command: &F,
 ) -> CachePreparationReport
 where
-    F: FnMut(Command) -> Fut,
+    F: Fn(Command) -> Fut,
     Fut: Future<Output = Result<Vec<u8>, CacheSetupError>>,
 {
     let command = if dry_run {
-        detect_command(cache_root, cwd)
+        detect_command(&context.cache_root, &context.cwd)
     } else {
-        mount_command(cache_root, cwd, &modes)
+        mount_command(&context.cache_root, &context.cwd, &modes)
     };
     tracing::info!(?command, "Executing spacectl");
     let started = Instant::now();
@@ -167,7 +172,7 @@ where
             CachePreparationReport {
                 scope,
                 modes: selected_modes,
-                relative_cache_dir,
+                relative_cache_dir: context.relative_cache_dir,
                 response: Some(response),
                 error: None,
                 duration,
@@ -181,7 +186,7 @@ where
             span.record("otel.status_code", "ERROR");
             span.record("otel.status_description", err.to_string());
             tracing::error!(error = ?err, "spacectl cache mount failed");
-            failed_invocation(scope, modes, relative_cache_dir, err, duration)
+            failed_invocation(scope, modes, context.relative_cache_dir, err, duration)
         }
     }
 }
@@ -198,7 +203,8 @@ fn mount_error_diagnostic(error: &CacheSetupError) -> Cow<'_, str> {
         | CacheSetupError::SpawnFailed
         | CacheSetupError::JsonParseFailed
         | CacheSetupError::Timeout
-        | CacheSetupError::EnvExportFailed => Cow::Owned(error.to_string()),
+        | CacheSetupError::EnvExportFailed
+        | CacheSetupError::PlanInvariantFailed => Cow::Owned(error.to_string()),
     }
 }
 
diff --git a/specs/REMOTE-3146/TECH.md b/specs/REMOTE-3146/TECH.md
new file mode 100644
index 00000000000..435f911fa64
--- /dev/null
+++ b/specs/REMOTE-3146/TECH.md
@@ -0,0 +1,337 @@
+# Nested spacectl discovery and bounded concurrent detection
+
+Linear: [REMOTE-3146](https://linear.app/warpdotdev/issue/REMOTE-3146/discover-nested-build-tools-with-concurrent-spacectl-cache-setup)
+
+Originating Slack thread:
+[C0BDQDW8V5E / 1788767403.717799](https://warpdev.slack.com/archives/C0BDQDW8V5E/p1788767403717799)
+
+Code references use warp commit
+[`51242b5f0af80fff81613ff6561eed29ba8922fa`](https://github.com/warpdotdev/warp/tree/51242b5f0af80fff81613ff6561eed29ba8922fa)
+on `master`.
+
+## Summary
+
+Build-cache setup detects tools only at each repository root. Nested projects are missed.
+Implement one ordered blocking discovery producer that scans repositories for detector-aligned
+markers and sends the bounded candidate set through a bounded channel into one shared concurrent
+detector. Keep cache-directory creation single-file and keep all real mounts serial. Keep the
+synthetic global mount last.
+
+## Context
+
+- [`prepare_environment_impl`](https://github.com/warpdotdev/warp/blob/51242b5f0af80fff81613ff6561eed29ba8922fa/app/src/ai/agent_sdk/driver/environment.rs#L373-L452)
+  runs cache setup after cloning and before setup commands. Cache failures do not abort environment
+  preparation.
+- [`setup_caches`](https://github.com/warpdotdev/warp/blob/51242b5f0af80fff81613ff6561eed29ba8922fa/app/src/ai/agent_sdk/driver/cache_setup.rs#L44-L112)
+  creates one `RepositoryCacheSource` per checkout and reports invocation failures.
+- [`setup_cache`](https://github.com/warpdotdev/warp/blob/51242b5f0af80fff81613ff6561eed29ba8922fa/crates/build_cache/src/lib.rs#L450-L646)
+  detects repositories serially, constructs a plan, and applies every mount serially.
+- [`construct_plan`](https://github.com/warpdotdev/warp/blob/51242b5f0af80fff81613ff6561eed29ba8922fa/crates/build_cache/src/lib.rs#L688-L773)
+  appends a global union of all detected modes. The global configuration must remain last because a
+  mode can mix cwd-relative paths with shared paths.
+- [`run_spacectl_mount`](https://github.com/warpdotdev/warp/blob/51242b5f0af80fff81613ff6561eed29ba8922fa/crates/build_cache/src/spacectl.rs#L116-L193)
+  uses the command cwd for both detection and mounting. The default runner applies a 60-second
+  timeout and `kill_on_drop(true)`.
+- `spacectl` 0.12.2 detection is cwd-only. A local triage fixture measured four detects at about
+  196 ms serially and 127 ms concurrently. These measurements show benefit, not a performance
+  guarantee.
+
+## Technical design
+
+### 1. Produce candidate roots with `walkdir`
+
+Add `walkdir.workspace = true` to `crates/build_cache/Cargo.toml`. In one
+`tokio::task::spawn_blocking` task, sort `RepositoryCacheSource` values by `RepoCacheKey`, send each
+repository root first, then drive one `walkdir::WalkDir` iterator for that repository. All traversal
+state stays local to the blocking task.
+
+Configure each iterator with:
+
+- `min_depth(1)`, because the producer handles the always-included root separately;
+- `max_depth(7)`, which bounds traversal while still reaching `.config/mise/config.toml` for a
+  candidate root at depth 4;
+- `follow_links(false)` and `follow_root_links(false)`;
+- `sort_by_file_name()`; and
+- `into_iter().filter_entry(...)` to reject ignored directory entries and symlink entries before
+  descent.
+
+`WalkDir` yields a directory before its contents and uses depth-first traversal. Sorted sibling
+names therefore define deterministic depth-first selection. Do not reconstruct breadth-first
+traversal around `WalkDir`; that would restore a custom directory queue and defeat the reuse.
+Changing from breadth-first to depth-first can change which 32 roots a truncated repository retains.
+This is intentional and is covered by fixtures.
+
+Apply these limits and error rules:
+
+- Always include the repository root. It has depth 0 and does not count against the child limit.
+- Accept every candidate whose marker is reached by the bounded walk. Do not apply another
+  candidate-depth restriction after traversal.
+- Visit at most 10,000 non-ignored, non-symlink directories per repository, including the root.
+  Files do not count against this limit.
+- Retain at most 32 child candidates per repository. When a 33rd distinct child candidate is found,
+  mark the scan truncated and stop that repository's iterator.
+- When the directory limit is reached with iterator work remaining, mark the scan truncated and stop
+  that repository's iterator.
+- On truncation, retain the root and the deterministic candidates already yielded. Continue cache
+  setup.
+- Reject a directory entry in `filter_entry` when its name is `.git`, `node_modules`, `target`,
+  `Pods`, `vendor`, `dist`, `build`, `.venv`, `.tox`, or `DerivedData`. The rejected directory and
+  its subtree do not count as visited. Do not reject a file with one of these names.
+- Reject symlink entries in `filter_entry`. `follow_links(false)` prevents descent through nested
+  links. `follow_root_links(false)` prevents the special default behavior that otherwise follows a
+  symlink passed as the traversal root. A symlink is not a marker.
+- Handle every `walkdir::Error` in place, emit a warning with `Error::depth()` and the underlying
+  `io::ErrorKind` when present, then continue iteration. `WalkDir` does not descend when it cannot
+  open a directory. Do not log raw error paths. A missing or unreadable repository root still
+  proceeds to root detection, which preserves the existing per-invocation error path.
+- Do not set `max_open`; use the crate's bounded default. This setting changes the file-descriptor
+  versus memory trade-off, not yielded results.
+
+Normalize a child root into a `PathBuf` by stripping the repository root and accepting only
+non-empty normal UTF-8 components. Preserve case and Unicode bytes. Skip a child path that is
+non-UTF-8 or contains a root, prefix, `.` or `..` component. Do not canonicalize child paths or
+resolve symlinks. Hash the normalized path's `OsStr` encoded byte slice directly. Accepted UTF-8
+paths have the same encoding on Namespace Linux and macOS.
+
+Deduplicate exact normalized roots. A directory with multiple markers is one candidate. Retain both
+a parent project root and a nested project root when each has a marker.
+
+### 2. Align marker rules with spacectl
+
+The marker table must mirror the detector inputs in the spacectl version shipped on Namespace
+workers. For spacectl 0.12.2, use these rules:
+
+- Exact entries: `Brewfile`, `bun.lock`, `Podfile`, `composer.json`, `deno.lock`, `go.mod`,
+  `go.work`, `.golangci.yml`, `.golangci.yaml`, `gradlew`, `build.gradle`, `pom.xml`, `mise.toml`,
+  `.mise.toml`, `.tool-versions`, `flake.nix`, `shell.nix`, `default.nix`, `package-lock.json`,
+  `pnpm-lock.yaml`, `poetry.lock`, `requirements.txt`, `Gemfile`, `Cargo.toml`, `Package.swift`,
+  `Tuist.swift`, `tuist.toml`, `uv.lock`, and `yarn.lock`.
+- Exact relative entries: `mise/config.toml`, `.mise/config.toml`, `.config/mise.toml`, and
+  `.config/mise/config.toml`. The candidate is the ancestor from which spacectl checks that relative
+  path, not the marker's immediate parent.
+- Directory entry: `Tuist`.
+- Suffix entries: directories ending in `.xcodeproj` or `.xcworkspace`.
+
+For exact, directory, and suffix entries, the candidate is the directory that contains the matched
+entry. A marker entry is never itself the candidate. When one file matches multiple marker rules,
+select only the longest relative marker so `.config/mise.toml` and
+`.config/mise/config.toml` identify the directory containing `.config`.
+
+Do not add looser markers that 0.12.2 does not use, including bare `package.json`,
+`pyproject.toml`, `settings.gradle`, or `build.gradle.kts`. Tool-binary checks remain spacectl's
+responsibility. Binary-only modes such as `apt`, Kotlin Native, and Playwright are discovered at the
+always-included repository root; they do not cause child candidates.
+
+Before implementation, verify the worker's shipped spacectl version and compare its provider source
+with this table. If detector semantics differ, update this spec and the table in the same PR.
+
+### 3. Prepare stable isolated cache roots
+
+After receiving a candidate and before yielding its detection future, create that candidate's
+configuration root and await any permission fallback. The receiving stream prepares only one
+directory at a time. A preparation may overlap already-running dry-run detections, but it must not
+overlap another preparation or any real mount. This overlap is safe because each candidate has a
+distinct cache root, and dry-run detection does not apply mounts. A creation failure yields a keyed
+non-fatal degradation result for that candidate and does not schedule spacectl.
+
+- Preserve the current root cache path: `repos/<repo-key>`.
+- Use `repos/<repo-key>/nested/<stable-id>` for a child root.
+- Compute `<stable-id>` as lowercase hexadecimal SHA-256 of the normalized relative path's encoded
+  bytes. Do not hash an absolute checkout path.
+- Validate that all configuration cache paths are safe relative paths and unique.
+- If two distinct roots produce the same configuration path, reject the plan before real mounts,
+  record one non-fatal plan-invariant degradation, and continue environment preparation. Never share
+  the path.
+
+This scheme preserves existing root cache hits and isolates equal relative mount names such as
+`frontend/target` and `backend/target`.
+
+### 4. Pipeline candidates through one shared detector limit
+
+Add `futures.workspace = true` and Tokio with its `rt` and `sync` features to the normal dependencies
+in `crates/build_cache/Cargo.toml`. Use `tokio::task::spawn_blocking` for the synchronous filesystem
+walk and a `tokio::sync::mpsc` channel with capacity 8 between discovery and the async receiving
+stream. Use `futures::stream::StreamExt::buffer_unordered(8)` as the detection-concurrency primitive.
+Do not add a custom semaphore.
+
+The blocking producer owns the sorted repositories and keeps the current `WalkDir`, counters,
+deduplication state, and scan diagnostics as local variables. It uses `blocking_send`, so a full
+channel blocks traversal instead of accumulating an unbounded candidate queue. The async receiver
+prepares each candidate's cache directory serially and yields its detection future. Apply
+`buffer_unordered(8)` once to this stream and collect the results.
+
+- The buffer's limit of 8 is the only detection limit and is shared across all repositories.
+- At most eight yielded detection futures are in flight. The receiver pulls and prepares another
+  candidate only when the detector buffer has capacity. At most eight additional unprepared
+  candidates wait in the bounded channel; no unbounded candidate queue or channel is permitted.
+- Selection remains deterministic even though production is demand-driven. The producer alone
+  advances each sorted `WalkDir` iterator and applies that repository's 10,000-directory and
+  32-child limits. Detection completion order can change when production resumes, but it cannot
+  change the next candidate selected.
+- Change the command hook from exclusive `FnMut` use to a concurrency-safe `Fn` shape. Wrap it in
+  `Arc` inside `setup_cache`; each yielded future owns an `Arc` clone. Pass shared references to
+  both directory preparation and spacectl invocation. Tests must put mutable fake-runner state
+  behind shared synchronization such as `Arc<Mutex<...>>`; do not serialize the production
+  scheduler behind the fake-runner API.
+- Run `spacectl cache mount --detect='*' --dry_run=true` with each candidate as cwd and its isolated
+  cache root.
+- Preserve the 60-second timeout and `kill_on_drop(true)` for every invocation.
+- Dropping cache setup drops the channel receiver. A producer blocked in `blocking_send` wakes with
+  an error and exits; between sends it checks `Sender::is_closed()` on each `WalkDir` entry and
+  exits. Tokio cannot forcibly abort a running `spawn_blocking` closure, so an in-progress
+  filesystem operation must return before the closure observes receiver closure. No producer task
+  or queued candidate keeps cache setup resources alive after that point.
+- An invocation failure, timeout, malformed response, or empty mode set affects only that root.
+- Do not cancel siblings after a failure.
+- Attach the canonical key `(RepoCacheKey, root-first flag, normalized child path)` to each
+  preparation failure and detection result. Sort all keyed results before constructing the plan or
+  returning the report. Completion order must not affect the plan, mount order, environment
+  overlay, telemetry report order, or truncation result.
+
+### 5. Plan and apply mounts serially
+
+Create one repository-scoped `CacheConfiguration` for every successful non-empty detection. Multiple
+configurations may share a `RepoCacheKey`, but every configuration must have a unique cwd and cache
+directory.
+
+- Update `CacheSetupPlan::validate` and its documentation to permit repeated ordered repository keys
+  and require unique repository configuration paths.
+- Sort repository configurations by repo key, then root before child, then normalized child path.
+- Union all successful detected modes with `additional_global_modes` for one global configuration.
+- Run every real repository mount serially in canonical plan order.
+- Run the global mount serially after all repository mounts.
+- Create the global cache directory serially.
+- Preserve current last-successful-repository environment overlay behavior and global-environment
+  precedence. Resolve any duplicate repository environment keys by canonical plan order.
+- Preserve `prepare_environment_impl` behavior: any cache degradation is reported, but environment
+  preparation continues.
+
+Do not attempt concurrent real mounts in v1. Rust, for example, can combine `./target` with shared
+Cargo paths. Concurrent mounts can race even when cache-root leaves differ.
+
+Nested discovery applies wherever the existing build-cache gate enables setup. V1 must work on
+Namespace Linux and macOS without enabling caching on any new platform. Keep filesystem helpers and
+unit tests platform-neutral so the crate continues to compile on other supported targets.
+
+### 6. Logging and telemetry
+
+Create one child span for the whole discovery process and one child span per repository. Record
+visited directory count, selected child count, and truncation reason (`directory_limit` or
+`candidate_limit`) on each repository span. Record total scheduled detects and the configured
+detection limit on the cache-setup span.
+
+Add the stable child ID to detection spans. Do not put raw absolute checkout paths in safe logs or
+Sentry extras. Emit one warning per truncated repository and one privacy-safe warning with error
+depth and `io::ErrorKind` per unreadable entry. Expected limit truncation is non-fatal and must not
+cancel detection or mounting. Create the whole-discovery span under the active cache-setup span and
+enter it in the blocking closure so every repository discovery span and warning remains in the
+setup trace.
+
+## Decisions
+
+- **Marker scan instead of spacectl in every directory.** A bounded marker scan avoids process spam
+  and matches cwd-based detector semantics. Calling spacectl for every directory was rejected
+  because repository breadth and 60-second per-process timeouts make latency unbounded.
+- **Pipeline discovery into detection.** Completing every scan before detection is simpler, but it
+  adds scan latency to the critical path and retains the full candidate set. One blocking producer
+  and a bounded channel keep traversal state local while providing backpressure. Selection limits
+  belong only to producer state, each cache root is prepared before its future is yielded, and
+  keyed results are sorted after completion.
+- **Use `buffer_unordered` instead of a custom limiter.** The workspace already depends on
+  `futures`. `StreamExt::buffer_unordered(8)` directly bounds a stream of detection futures and
+  provides backpressure. A custom futures semaphore would duplicate this behavior.
+- **Use the app's Tokio runtime for blocking discovery.** `build_cache` is a native-only dependency
+  of the app, whose native runtime is Tokio. `spawn_blocking` removes the boxed iterator and
+  resumable discovery structs without adding a runtime to wasm builds. Standalone native callers,
+  including `validate_spacectl`, must enter a Tokio runtime before calling `setup_cache`.
+- **Use sorted depth-first `WalkDir` traversal.** `WalkDir` supplies bounded descriptors, depth
+  limits, symlink controls, subtree filtering, and recoverable errors. Retaining breadth-first
+  selection would require a custom queue. Sorted depth-first selection is deterministic and makes
+  the 32-root truncation policy explicit.
+- **Eight shared detection slots.** This captures the measured concurrency benefit while bounding
+  process and detector fan-out. A per-repository limit was rejected because multiple repositories
+  could exceed the intended host-wide limit.
+- **Serial real mounts.** Concurrent mounts were rejected for v1 because isolated cache leaves do
+  not isolate shared destination paths. The global mount remains last.
+- **Preserve the root cache path.** Moving all roots under a new namespace was rejected because it
+  would discard existing root cache hits.
+- **Hash normalized child paths.** Raw relative paths are easier to inspect but can be long and
+  platform-sensitive. SHA-256 of the path's encoded bytes produces a stable safe component across
+  Namespace Linux and macOS. Telemetry retains the stable ID for correlation.
+
+## Assumptions
+
+- The Namespace worker still ships spacectl detector semantics equivalent to 0.12.2. Implementation
+  must verify this before coding.
+- Repository-relative project paths are UTF-8. A non-UTF-8 child path is skipped rather than given a
+  platform-specific cache identity.
+- The current cache setup remains before user setup commands. Tools installed only by setup commands
+  remain unavailable to detection.
+- Overlapping real spacectl mounts are not proven safe on Linux or macOS. V1 does not rely on that
+  behavior.
+- `WalkDir::sort_by_file_name()` is deterministic for a fixed filesystem and platform. Cross-platform
+  traversal order for non-UTF-8 entry names is not part of the cache identity contract; such paths
+  cannot become candidates.
+
+## Out of scope
+
+- Recursive detection changes in spacectl or Namespace.
+- Calling spacectl in directories without detector-aligned markers.
+- Moving cache setup after user setup commands.
+- Concurrent cache-directory creation or real mount invocations. Cache-directory preparation may
+  overlap dry-run detection for a different candidate.
+- New detectors or support for looser manifests that the shipped spacectl does not recognize.
+- UI changes or computer-use verification.
+
+## Validation criteria
+
+1. `cargo nextest run -p build_cache` passes and includes unit coverage for:
+   - representative direct markers and every relative, directory, and suffix marker rule;
+   - non-markers such as bare `package.json`, ignored trees, and symlinks;
+   - exact deduplication while retaining marked parent and child roots;
+   - sorted depth-first `WalkDir` selection, the 10,000-directory limit, and 32 children plus root;
+   - deterministic truncation and unreadable-entry isolation;
+   - `max_depth(7)` finding `.config/mise/config.toml` for a depth-4 candidate and accepting deeper
+     candidates whose markers the walk reaches;
+   - ignored directory subtrees, symlinked nested directories, and a symlink traversal root are not
+     followed;
+   - stable Linux/macOS child IDs, preserved root cache paths, and unique safe cache paths;
+   - a fake runner that observes more than one and no more than eight simultaneous detects across
+     multiple repositories;
+   - detection starts before the final scan completes, cache-directory preparations never overlap,
+     and a preparation can overlap an active dry-run detection;
+   - producer backpressure keeps at most eight queued candidates and at most eight detection
+     futures in flight, receiver drop stops the blocking producer, and selection is identical
+     across deliberately permuted completion orders;
+   - per-root failure and timeout isolation, `kill_on_drop`, deterministic keyed report ordering,
+     serial mount execution, and the global mount last;
+   - repeated ordered repository keys and unique cache-directory plan invariants.
+2. `cargo nextest run -p warp cache_setup` passes to confirm Namespace gating, source mapping,
+   degradation reporting, and environment export behavior remain compatible.
+3. Extend `crates/build_cache/examples/validate_spacectl.rs` with one repository containing root,
+   `frontend`, and `backend` fixtures. With the worker's spacectl version available,
+   `cargo run -p build_cache --example validate_spacectl -- --reset` must run within the validator's
+   Tokio runtime without a missing-reactor panic and show:
+   - one detect per selected root;
+   - the expected nested modes;
+   - distinct nested cache roots;
+   - serial real mounts in canonical order;
+   - one final global mount.
+4. Record five-run medians for a 32-child fixture with serial detection and the concurrency-8
+   implementation on a Namespace Linux worker. Concurrent median wall time must not exceed the
+   serial median. Record scan time and process counts; do not add a hardware-dependent unit-test
+   latency threshold.
+5. Verify the marker table against the exact spacectl provider source deployed on the validation
+   worker. Link the source tag or commit in the implementation PR.
+6. Before any follow-up enables concurrent real mounts, run controlled overlapping-mount tests for
+   mixed relative/global modes on Namespace Linux and macOS. V1 passes without this experiment
+   because all real mounts remain serial.
+7. Run `./script/format`, the clippy command selected by `./script/presubmit`, and `git diff --check`
+   before implementation review. No computer-use artifact is required.
+
+## Parallelization
+
+Use one implementer for discovery, plan changes, runner refactoring, and unit tests because these
+changes share the `setup_cache` contract and fake-runner seam. After unit tests pass, Linux timing
+validation and the optional macOS mount-safety investigation can run independently. Land all spec,
+implementation, and validation updates in this PR.