diff --git a/crates/sqlbuild-cache/src/artifacts/_helpers/files.rs b/crates/sqlbuild-cache/src/artifacts/_helpers/files.rs
index 7365790654..5f81f2986a 100644
--- a/crates/sqlbuild-cache/src/artifacts/_helpers/files.rs
+++ b/crates/sqlbuild-cache/src/artifacts/_helpers/files.rs
@@ -141,23 +141,50 @@ pub(crate) fn remove_stale(
     compiled_dir: &Path,
     managed: &HashSet<PathBuf>,
 ) -> Result<usize, ArtifactError> {
+    remove_stale_listing(compiled_dir, managed).map(|(removed, _)| removed)
+}
+
+/// `remove_stale`, with every directory left after the walk, or `None` when one was not read.
+pub(crate) fn remove_stale_listing(
+    compiled_dir: &Path,
+    managed: &HashSet<PathBuf>,
+) -> Result<(usize, Option<Vec<PathBuf>>), ArtifactError> {
     if !compiled_dir.is_dir() {
-        return Ok(0);
+        return Ok((0, None));
     }
-    let (_, removed) = remove_stale_under(compiled_dir, managed)?;
+    let cleaned: Cleaned = remove_stale_under(compiled_dir, managed)?;
     let _ = std::fs::remove_dir(compiled_dir);
-    Ok(removed)
+    let listed: Option<Vec<PathBuf>> = cleaned.complete.then_some(cleaned.directories);
+    Ok((cleaned.removed, listed))
 }
 
-/// Clean one directory bottom-up as `os.walk(topdown=False)`; returns (gone, files removed).
+/// What cleaning one directory did, and whether every directory below it could be read.
+#[derive(Debug)]
+struct Cleaned {
+    gone: bool,
+    removed: usize,
+    directories: Vec<PathBuf>,
+    complete: bool,
+}
+
+/// Clean one directory bottom-up as `os.walk(topdown=False)`.
 fn remove_stale_under(
     directory: &Path,
     managed: &HashSet<PathBuf>,
-) -> Result<(bool, usize), ArtifactError> {
+) -> Result<Cleaned, ArtifactError> {
     let entries = match std::fs::read_dir(directory) {
         Ok(entries) => entries,
-        Err(_) => return Ok((false, 0)),
+        Err(_) => {
+            return Ok(Cleaned {
+                gone: false,
+                removed: 0,
+                directories: Vec::new(),
+                complete: false,
+            });
+        }
     };
+    let mut directories: Vec<PathBuf> = vec![directory.to_path_buf()];
+    let mut complete: bool = true;
     let mut removed: usize = 0;
     let mut kept_file: bool = false;
     let mut subdirectories_removed: bool = true;
@@ -176,9 +203,11 @@ fn remove_stale_under(
             subdirectories_removed = false;
             continue;
         }
-        let (gone, below) = remove_stale_under(&subdirectory, managed)?;
-        removed += below;
-        subdirectories_removed &= gone;
+        let below: Cleaned = remove_stale_under(&subdirectory, managed)?;
+        removed += below.removed;
+        subdirectories_removed &= below.gone;
+        complete &= below.complete;
+        directories.extend(below.directories);
     }
     for file in files {
         if managed.contains(&file) {
@@ -189,7 +218,15 @@ fn remove_stale_under(
         }
     }
     let gone: bool = !kept_file && subdirectories_removed && removed_directory(directory);
-    Ok((gone, removed))
+    if gone {
+        directories.clear();
+    }
+    Ok(Cleaned {
+        gone,
+        removed,
+        directories,
+        complete,
+    })
 }
 
 fn removed_directory(directory: &Path) -> bool {
diff --git a/crates/sqlbuild-cache/src/artifacts/_helpers/mod.rs b/crates/sqlbuild-cache/src/artifacts/_helpers/mod.rs
index c72064b61f..5a3c0fbdc0 100644
--- a/crates/sqlbuild-cache/src/artifacts/_helpers/mod.rs
+++ b/crates/sqlbuild-cache/src/artifacts/_helpers/mod.rs
@@ -1,2 +1,3 @@
 pub(crate) mod files;
 pub(crate) mod sql_test_records;
+pub(crate) mod tree_stamp;
diff --git a/crates/sqlbuild-cache/src/artifacts/_helpers/tree_stamp.rs b/crates/sqlbuild-cache/src/artifacts/_helpers/tree_stamp.rs
new file mode 100644
index 0000000000..c570b084b8
--- /dev/null
+++ b/crates/sqlbuild-cache/src/artifacts/_helpers/tree_stamp.rs
@@ -0,0 +1,162 @@
+//! The stamp of a compiled tree that held exactly its managed files when last cleaned.
+
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::digest::main::content_digest::content_digest;
+use crate::digest::types::ContentDigest;
+use crate::store::main::open_native_store::open_native_store;
+use crate::store::models::{NativeStore, RecordReader, RecordWriter};
+
+const STORE_KIND: &str = "compiled-tree";
+const STORE_ENVIRONMENT: &str = "compiled-tree-1";
+
+/// One folder's identity: its path, inode, and modification and change times.
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct FolderStamp {
+    path: String,
+    inode: u64,
+    mtime_ns: i64,
+    ctime_ns: i64,
+}
+
+fn entry_key() -> ContentDigest {
+    content_digest(&["tree"])
+}
+
+/// The digest of the compiled folder and its managed file set, in any order.
+pub(crate) fn managed_digest(compiled_dir: &Path, managed: &HashSet<PathBuf>) -> ContentDigest {
+    let mut paths: Vec<String> = managed
+        .iter()
+        .map(|path| path.to_string_lossy().into_owned())
+        .collect();
+    paths.sort_unstable();
+    let mut hasher: blake3::Hasher = blake3::Hasher::new();
+    for part in std::iter::once(compiled_dir.to_string_lossy().into_owned()).chain(paths) {
+        let _ = hasher.update(&(part.len() as u64).to_le_bytes());
+        let _ = hasher.update(part.as_bytes());
+    }
+    *hasher.finalize().as_bytes()
+}
+
+#[cfg(unix)]
+fn folder_stamp(path: &Path) -> Option<FolderStamp> {
+    use std::os::unix::fs::MetadataExt;
+
+    let metadata: std::fs::Metadata = match std::fs::symlink_metadata(path) {
+        Ok(metadata) => metadata,
+        Err(_absent) => return None,
+    };
+    if !metadata.is_dir() {
+        return None;
+    }
+    Some(FolderStamp {
+        path: path.to_string_lossy().into_owned(),
+        inode: metadata.ino(),
+        mtime_ns: metadata.mtime() * 1_000_000_000 + metadata.mtime_nsec(),
+        ctime_ns: metadata.ctime() * 1_000_000_000 + metadata.ctime_nsec(),
+    })
+}
+
+#[cfg(not(unix))]
+fn folder_stamp(_path: &Path) -> Option<FolderStamp> {
+    None
+}
+
+/// Whether the tree recorded at `stamp_path` for `digest` is unchanged on disk.
+pub(crate) fn tree_unchanged(stamp_path: &Path, digest: &ContentDigest) -> bool {
+    let Ok(mut store) = open_native_store(stamp_path, STORE_KIND, STORE_ENVIRONMENT) else {
+        return false;
+    };
+    let Some(bytes) = store.get(&entry_key()) else {
+        return false;
+    };
+    let Ok((recorded, folders)) = decode(bytes) else {
+        return false;
+    };
+    recorded == *digest
+        && !folders.is_empty()
+        && folders
+            .iter()
+            .all(|folder| folder_stamp(Path::new(&folder.path)).as_ref() == Some(folder))
+}
+
+/// Remove the recorded stamp, before the tree is changed.
+pub(crate) fn forget_tree(stamp_path: &Path) {
+    let _ = std::fs::remove_file(stamp_path);
+}
+
+/// The current time in nanoseconds since the epoch, as file stamps record it.
+pub(crate) fn now_ns() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |elapsed| {
+            i64::try_from(elapsed.as_nanos()).unwrap_or(i64::MAX)
+        })
+}
+
+/// Record `folders` as the tree `digest` names when each stamps and changed before `trusted_ns`.
+pub(crate) fn record_tree(
+    stamp_path: &Path,
+    digest: &ContentDigest,
+    folders: &[PathBuf],
+    trusted_ns: i64,
+) {
+    let Some(stamps) = folders
+        .iter()
+        .map(|path| folder_stamp(path))
+        .collect::<Option<Vec<FolderStamp>>>()
+    else {
+        return;
+    };
+    let settled: bool = !stamps.is_empty()
+        && stamps
+            .iter()
+            .all(|stamp| stamp.mtime_ns.max(stamp.ctime_ns) < trusted_ns);
+    if !settled {
+        return;
+    }
+    let mut store: NativeStore = NativeStore {
+        kind: STORE_KIND.to_owned(),
+        environment: STORE_ENVIRONMENT.to_owned(),
+        ..NativeStore::default()
+    };
+    store.put(entry_key(), encode(digest, &stamps));
+    let _ = store.save(stamp_path, &[]);
+}
+
+fn encode(digest: &ContentDigest, folders: &[FolderStamp]) -> Vec<u8> {
+    let mut writer: RecordWriter = RecordWriter::default();
+    writer.put_bytes(digest);
+    writer.put_u64(folders.len() as u64);
+    for folder in folders {
+        writer.put_str(&folder.path);
+        writer.put_u64(folder.inode);
+        writer.put_u64(folder.mtime_ns as u64);
+        writer.put_u64(folder.ctime_ns as u64);
+    }
+    writer.into_bytes()
+}
+
+fn decode(
+    bytes: &[u8],
+) -> Result<(ContentDigest, Vec<FolderStamp>), crate::store::errors::StoreDecodeError> {
+    let mut reader: RecordReader<'_> = RecordReader::new(bytes);
+    let digest: ContentDigest = reader
+        .bytes()?
+        .try_into()
+        .map_err(|_| crate::store::errors::StoreDecodeError)?;
+    let count: u64 = reader.u64()?;
+    let mut folders: Vec<FolderStamp> = Vec::new();
+    for _ in 0..count {
+        folders.push(FolderStamp {
+            path: reader.string()?,
+            inode: reader.u64()?,
+            mtime_ns: reader.u64()? as i64,
+            ctime_ns: reader.u64()? as i64,
+        });
+    }
+    reader.finish()?;
+    Ok((digest, folders))
+}
diff --git a/crates/sqlbuild-cache/src/artifacts/main/mod.rs b/crates/sqlbuild-cache/src/artifacts/main/mod.rs
index 090d08595b..1f3a4d7571 100644
--- a/crates/sqlbuild-cache/src/artifacts/main/mod.rs
+++ b/crates/sqlbuild-cache/src/artifacts/main/mod.rs
@@ -1,6 +1,7 @@
 pub mod new_sql_test_record;
 pub mod publish_staged;
 pub mod remove_stale;
+pub mod remove_stale_unless_unchanged;
 pub mod reusable_sql_test_record;
 pub mod staged_files;
 pub mod stored_sql_test_record;
diff --git a/crates/sqlbuild-cache/src/artifacts/main/remove_stale_unless_unchanged.rs b/crates/sqlbuild-cache/src/artifacts/main/remove_stale_unless_unchanged.rs
new file mode 100644
index 0000000000..42baf94284
--- /dev/null
+++ b/crates/sqlbuild-cache/src/artifacts/main/remove_stale_unless_unchanged.rs
@@ -0,0 +1,29 @@
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+
+use crate::artifacts::_helpers::files::remove_stale_listing;
+use crate::artifacts::_helpers::tree_stamp::{
+    forget_tree, managed_digest, now_ns, record_tree, tree_unchanged,
+};
+use crate::artifacts::errors::ArtifactError;
+use crate::digest::types::ContentDigest;
+
+/// `remove_stale`, skipped while the tree stamped at `stamp_path` still holds exactly `managed`.
+pub fn remove_stale_unless_unchanged(
+    compiled_dir: &Path,
+    managed: &HashSet<PathBuf>,
+    stamp_path: &Path,
+    racy_window_ns: i64,
+) -> Result<usize, ArtifactError> {
+    let digest: ContentDigest = managed_digest(compiled_dir, managed);
+    if tree_unchanged(stamp_path, &digest) {
+        return Ok(0);
+    }
+    forget_tree(stamp_path);
+    let trusted_ns: i64 = now_ns().saturating_sub(racy_window_ns);
+    let (removed, folders) = remove_stale_listing(compiled_dir, managed)?;
+    if let Some(folders) = folders {
+        record_tree(stamp_path, &digest, &folders, trusted_ns);
+    }
+    Ok(removed)
+}
diff --git a/crates/sqlbuild-cache/src/artifacts/tests.rs b/crates/sqlbuild-cache/src/artifacts/tests.rs
index 991ace2b43..6d280821bd 100644
--- a/crates/sqlbuild-cache/src/artifacts/tests.rs
+++ b/crates/sqlbuild-cache/src/artifacts/tests.rs
@@ -1,5 +1,8 @@
 #[path = "tests/helpers.rs"]
 mod helpers;
+#[cfg(unix)]
+#[path = "tests/test_remove_stale.rs"]
+mod remove_stale;
 #[path = "tests/test_types.rs"]
 mod test_types;
 #[path = "tests/test_write_artifacts.rs"]
diff --git a/crates/sqlbuild-cache/src/artifacts/tests/helpers.rs b/crates/sqlbuild-cache/src/artifacts/tests/helpers.rs
index febcafc534..114464f151 100644
--- a/crates/sqlbuild-cache/src/artifacts/tests/helpers.rs
+++ b/crates/sqlbuild-cache/src/artifacts/tests/helpers.rs
@@ -17,3 +17,33 @@ pub(crate) fn place(path: &Path, contents: &[u8]) {
     std::fs::create_dir_all(path.parent().expect("artifact parent")).expect("artifact directory");
     std::fs::write(path, contents).expect("existing artifact");
 }
+
+/// Every file below `root` as a sorted relative path.
+pub(crate) fn tree_files(root: &Path) -> Vec<String> {
+    let mut files: Vec<String> = walkdir::WalkDir::new(root)
+        .into_iter()
+        .flatten()
+        .filter(|entry| entry.file_type().is_file())
+        .filter_map(|entry| {
+            entry
+                .path()
+                .strip_prefix(root)
+                .ok()
+                .map(|relative| relative.to_string_lossy().into_owned())
+        })
+        .collect();
+    files.sort();
+    files
+}
+
+/// A compiled folder that is a plain directory.
+pub(crate) fn plain_compiled_root(target: &Path) {
+    std::fs::create_dir_all(target.join("compiled")).expect("compiled folder");
+}
+
+/// A compiled folder that is a link to a directory elsewhere in the target folder.
+#[cfg(unix)]
+pub(crate) fn linked_compiled_root(target: &Path) {
+    std::fs::create_dir_all(target.join("elsewhere")).expect("linked folder");
+    std::os::unix::fs::symlink(target.join("elsewhere"), target.join("compiled")).expect("link");
+}
diff --git a/crates/sqlbuild-cache/src/artifacts/tests/test_remove_stale.rs b/crates/sqlbuild-cache/src/artifacts/tests/test_remove_stale.rs
new file mode 100644
index 0000000000..8f4683dcfe
--- /dev/null
+++ b/crates/sqlbuild-cache/src/artifacts/tests/test_remove_stale.rs
@@ -0,0 +1,212 @@
+use std::collections::HashSet;
+use std::path::PathBuf;
+
+use crate::artifacts::_helpers::tree_stamp::{managed_digest, now_ns, record_tree};
+use crate::artifacts::main::remove_stale_unless_unchanged::remove_stale_unless_unchanged;
+use crate::artifacts::tests::helpers::{
+    linked_compiled_root, place, plain_compiled_root, tree_files,
+};
+use crate::artifacts::tests::test_types::{StampedTreeTestCase, WalkRaceTestCase};
+
+const MANAGED: [&str; 3] = [
+    "models/orders.sql",
+    "models/marts/customers.sql",
+    "tests/t.sql",
+];
+
+#[test]
+fn given_stamped_tree_when_cleaning_again_then_only_changed_trees_are_walked() {
+    let test_cases = [
+        StampedTreeTestCase {
+            description: "nothing changed",
+            change: |_| {},
+            managed_after: &MANAGED,
+            racy_window_ns: 0,
+            layout: plain_compiled_root,
+            expected_stamped: true,
+            expected_removed: 0,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+        StampedTreeTestCase {
+            description: "a stale file added to a stamped folder",
+            change: |root| place(&root.join("compiled/models/marts/old.sql"), b"select 0\n"),
+            managed_after: &MANAGED,
+            racy_window_ns: 0,
+            layout: plain_compiled_root,
+            expected_stamped: true,
+            expected_removed: 1,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+        StampedTreeTestCase {
+            description: "a stale file in a new folder",
+            change: |root| place(&root.join("compiled/models/new/old.sql"), b"select 0\n"),
+            managed_after: &MANAGED,
+            racy_window_ns: 0,
+            layout: plain_compiled_root,
+            expected_stamped: true,
+            expected_removed: 1,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+        StampedTreeTestCase {
+            description: "a managed file no longer managed",
+            change: |_| {},
+            managed_after: &["models/orders.sql", "models/marts/customers.sql"],
+            racy_window_ns: 0,
+            layout: plain_compiled_root,
+            expected_stamped: true,
+            expected_removed: 1,
+            expected_files: &["models/marts/customers.sql", "models/orders.sql"],
+        },
+        StampedTreeTestCase {
+            description: "the stamp file damaged",
+            change: |root| {
+                std::fs::write(root.join("cache/compiled-tree.bin"), b"garbage").expect("damaged");
+                place(&root.join("compiled/models/old.sql"), b"select 0\n");
+            },
+            managed_after: &MANAGED,
+            racy_window_ns: 0,
+            layout: plain_compiled_root,
+            expected_stamped: true,
+            expected_removed: 1,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+        StampedTreeTestCase {
+            description: "folders inside the racy window",
+            change: |root| place(&root.join("compiled/models/marts/old.sql"), b"select 0\n"),
+            managed_after: &MANAGED,
+            racy_window_ns: i64::MAX / 2,
+            layout: plain_compiled_root,
+            expected_stamped: false,
+            expected_removed: 1,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+        StampedTreeTestCase {
+            description: "a stale file directly in a linked compiled root",
+            change: |root| place(&root.join("compiled/stray.sql"), b"select 0\n"),
+            managed_after: &MANAGED,
+            racy_window_ns: 0,
+            layout: linked_compiled_root,
+            expected_stamped: false,
+            expected_removed: 1,
+            expected_files: &[
+                "models/marts/customers.sql",
+                "models/orders.sql",
+                "tests/t.sql",
+            ],
+        },
+    ];
+
+    for test_case in test_cases {
+        let folder = tempfile::tempdir().expect("target folder");
+        (test_case.layout)(folder.path());
+        let compiled: PathBuf = folder.path().join("compiled");
+        let stamp: PathBuf = folder.path().join("cache/compiled-tree.bin");
+        for path in MANAGED {
+            place(&compiled.join(path), b"select 1\n");
+        }
+        place(&compiled.join("models/stale.sql"), b"select 0\n");
+        let first: HashSet<PathBuf> = MANAGED.iter().map(|path| compiled.join(path)).collect();
+        let removed =
+            remove_stale_unless_unchanged(&compiled, &first, &stamp, test_case.racy_window_ns)
+                .expect("first pass");
+        std::thread::sleep(std::time::Duration::from_millis(30));
+        let settled =
+            remove_stale_unless_unchanged(&compiled, &first, &stamp, test_case.racy_window_ns)
+                .expect("settling pass");
+        let stamped: bool = stamp.is_file();
+        std::thread::sleep(std::time::Duration::from_millis(30));
+        (test_case.change)(folder.path());
+        let second: HashSet<PathBuf> = test_case
+            .managed_after
+            .iter()
+            .map(|path| compiled.join(path))
+            .collect();
+
+        let removed_again =
+            remove_stale_unless_unchanged(&compiled, &second, &stamp, test_case.racy_window_ns)
+                .expect("second pass");
+
+        assert_eq!(
+            (removed, settled, stamped),
+            (1, 0, test_case.expected_stamped),
+            "{}: first pass",
+            test_case.description
+        );
+        assert_eq!(
+            (removed_again, tree_files(&compiled)),
+            (
+                test_case.expected_removed,
+                test_case
+                    .expected_files
+                    .iter()
+                    .map(|path| (*path).to_owned())
+                    .collect::<Vec<String>>()
+            ),
+            "{}",
+            test_case.description
+        );
+    }
+}
+
+#[test]
+fn given_folder_changed_during_the_walk_when_recording_then_no_stamp_is_kept() {
+    let test_cases = [
+        WalkRaceTestCase {
+            description: "folder changed after the walk started",
+            change_during_walk: |compiled| place(&compiled.join("models/stray.sql"), b"select 0\n"),
+            expected_stamped: false,
+        },
+        WalkRaceTestCase {
+            description: "nothing changed during the walk",
+            change_during_walk: |_| {},
+            expected_stamped: true,
+        },
+    ];
+
+    for test_case in test_cases {
+        let folder = tempfile::tempdir().expect("target folder");
+        let compiled: PathBuf = folder.path().join("compiled");
+        let stamp: PathBuf = folder.path().join("cache/compiled-tree.bin");
+        place(&compiled.join("models/orders.sql"), b"select 1\n");
+        let managed: HashSet<PathBuf> = [compiled.join("models/orders.sql")].into_iter().collect();
+        std::thread::sleep(std::time::Duration::from_millis(30));
+        let since_ns: i64 = now_ns();
+        let folders: Vec<PathBuf> = vec![compiled.clone(), compiled.join("models")];
+        std::thread::sleep(std::time::Duration::from_millis(30));
+        (test_case.change_during_walk)(&compiled);
+
+        record_tree(
+            &stamp,
+            &managed_digest(&compiled, &managed),
+            &folders,
+            since_ns,
+        );
+
+        assert_eq!(
+            stamp.is_file(),
+            test_case.expected_stamped,
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-cache/src/artifacts/tests/test_types.rs b/crates/sqlbuild-cache/src/artifacts/tests/test_types.rs
index 0d130d0c77..a475d71bf5 100644
--- a/crates/sqlbuild-cache/src/artifacts/tests/test_types.rs
+++ b/crates/sqlbuild-cache/src/artifacts/tests/test_types.rs
@@ -24,3 +24,27 @@ pub(crate) struct RepeatedPathTestCase {
     pub(crate) repeated: usize,
     pub(crate) expected_contents: &'static [u8],
 }
+
+/// A change between two stale-file passes over a stamped compiled tree.
+#[cfg(unix)]
+pub(crate) struct StampedTreeTestCase {
+    pub(crate) description: &'static str,
+    /// Changes the target folder holding `compiled/` and `cache/` after the first pass.
+    pub(crate) change: fn(&std::path::Path),
+    /// The files the second pass manages, relative to the compiled folder.
+    pub(crate) managed_after: &'static [&'static str],
+    pub(crate) racy_window_ns: i64,
+    /// Creates the `compiled/` folder under the target folder.
+    pub(crate) layout: fn(&std::path::Path),
+    pub(crate) expected_stamped: bool,
+    pub(crate) expected_removed: usize,
+    pub(crate) expected_files: &'static [&'static str],
+}
+
+/// A change made to the compiled tree after its stale-file walk started.
+#[cfg(unix)]
+pub(crate) struct WalkRaceTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) change_during_walk: fn(&std::path::Path),
+    pub(crate) expected_stamped: bool,
+}
diff --git a/crates/sqlbuild-cache/src/digest/constants.rs b/crates/sqlbuild-cache/src/digest/constants.rs
index 3f14686585..b2f4bd45bc 100644
--- a/crates/sqlbuild-cache/src/digest/constants.rs
+++ b/crates/sqlbuild-cache/src/digest/constants.rs
@@ -48,3 +48,6 @@ pub(crate) const METADATA_DIRECTORY_SUFFIXES: [&str; 2] = [".dist-info", ".egg-i
 pub(crate) const METADATA_FILES: [&str; 2] = ["METADATA", "PKG-INFO"];
 /// The installer's record of where a distribution was installed from.
 pub(crate) const DIRECT_URL_FILE: &str = "direct_url.json";
+
+/// Inputs at least this large are hashed across the rayon pool; the digest is the same either way.
+pub(crate) const PARALLEL_HASH_BYTES: usize = 1 << 20;
diff --git a/crates/sqlbuild-cache/src/digest/main/bytes_digest.rs b/crates/sqlbuild-cache/src/digest/main/bytes_digest.rs
index 7b5ec8f290..5b66238fa9 100644
--- a/crates/sqlbuild-cache/src/digest/main/bytes_digest.rs
+++ b/crates/sqlbuild-cache/src/digest/main/bytes_digest.rs
@@ -1,8 +1,14 @@
 //! The BLAKE3 digest of bytes held in memory, equal to `digest_files` for the same contents.
 
+use crate::digest::constants::PARALLEL_HASH_BYTES;
 use crate::digest::types::ContentDigest;
 
 /// The digest `digest_files` gives a file holding exactly `contents`.
 pub fn bytes_digest(contents: &[u8]) -> ContentDigest {
-    *blake3::hash(contents).as_bytes()
+    if contents.len() < PARALLEL_HASH_BYTES {
+        return *blake3::hash(contents).as_bytes();
+    }
+    let mut hasher: blake3::Hasher = blake3::Hasher::new();
+    let _ = hasher.update_rayon(contents);
+    *hasher.finalize().as_bytes()
 }
diff --git a/crates/sqlbuild-cache/src/digest/main/fingerprint_project_files.rs b/crates/sqlbuild-cache/src/digest/main/fingerprint_project_files.rs
index 9c544f71cf..a36daf4e99 100644
--- a/crates/sqlbuild-cache/src/digest/main/fingerprint_project_files.rs
+++ b/crates/sqlbuild-cache/src/digest/main/fingerprint_project_files.rs
@@ -1,16 +1,24 @@
 //! Fingerprint the content of a project's files, leaving out the files a cache key covers itself.
 
 use rayon::prelude::{IntoParallelRefIterator, ParallelIterator};
-use sha2::{Digest, Sha256};
 use std::collections::HashSet;
 use std::path::Path;
 
 use crate::digest::_helpers::walk::{CoveredFile, covered_files};
 use crate::digest::constants::{CONTENT_TAG, PRESENCE_TAG};
 use crate::digest::errors::FingerprintError;
-use crate::digest::main::content_digest::content_digest;
 use crate::digest::types::ContentDigest;
 
+/// BLAKE3 digest of `parts`, each length-prefixed so part boundaries cannot shift.
+fn parts_digest(parts: &[&[u8]]) -> ContentDigest {
+    let mut hasher: blake3::Hasher = blake3::Hasher::new();
+    for part in parts {
+        let _ = hasher.update(&(part.len() as u64).to_le_bytes());
+        let _ = hasher.update(part);
+    }
+    *hasher.finalize().as_bytes()
+}
+
 /// Digest every covered file's project-relative path and bytes; fails if any file is unreadable.
 pub fn fingerprint_project_files(
     root: &Path,
@@ -21,18 +29,18 @@ pub fn fingerprint_project_files(
         .par_iter()
         .map(file_digest)
         .collect::<Result<Vec<ContentDigest>, FingerprintError>>()?;
-    let mut hasher: Sha256 = Sha256::new();
+    let mut hasher: blake3::Hasher = blake3::Hasher::new();
     for (file, digest) in files.iter().zip(digests) {
-        hasher.update(content_digest(&[file.relative_path.as_bytes(), &digest]));
+        let _ = hasher.update(&parts_digest(&[file.relative_path.as_bytes(), &digest]));
     }
-    Ok(hasher.finalize().into())
+    Ok(*hasher.finalize().as_bytes())
 }
 
 fn file_digest(file: &CoveredFile) -> Result<ContentDigest, FingerprintError> {
     if file.presence_only {
-        return Ok(content_digest(&[PRESENCE_TAG]));
+        return Ok(parts_digest(&[PRESENCE_TAG]));
     }
     let bytes: Vec<u8> = std::fs::read(&file.path)
         .map_err(|error| FingerprintError::Read(file.path.clone(), error))?;
-    Ok(content_digest(&[CONTENT_TAG, &bytes]))
+    Ok(parts_digest(&[CONTENT_TAG, &bytes]))
 }
diff --git a/crates/sqlbuild-cache/src/project_reuse/_helpers/attempt.rs b/crates/sqlbuild-cache/src/project_reuse/_helpers/attempt.rs
index 0459e17285..689056c288 100644
--- a/crates/sqlbuild-cache/src/project_reuse/_helpers/attempt.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/_helpers/attempt.rs
@@ -5,7 +5,7 @@ use std::path::Path;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use crate::project_reuse::_helpers::entry::{
-    read_entry, remove_entry, rewrite_inputs, write_entry,
+    read_output, read_slot, remove_entry, rewrite_inputs, write_entry,
 };
 use crate::project_reuse::_helpers::files::{
     carried_forward_digests, compare_project_files, needs_refresh, restamped_paths,
@@ -74,11 +74,14 @@ pub(crate) fn check(check: ReuseCheck<'_>) -> ReuseAttempt {
     attempt.search_path =
         search_path_stamps(&check.identity.search_path, &check.project_dir, rules);
     attempt.snapshot = snapshot(&check.project_dir, rules);
-    if let Some((inputs, output)) = read_entry(&check.store_path) {
-        let comparison: Comparison = checked(&attempt, &inputs, rules);
+    if let Some(slot) = read_slot(&check.store_path) {
+        let inputs: &StoredInputs = &slot.inputs;
+        let comparison: Comparison = checked(&attempt, inputs, rules);
         attempt.digested_files += comparison.digested_files;
         attempt.digests = comparison.digests;
-        if comparison.unchanged {
+        if comparison.unchanged
+            && let Some(output) = read_output(&check.store_path, &slot)
+        {
             if check.json_output && output.timings_span.is_none() {
                 attempt.replay_failed = true;
                 return attempt;
@@ -92,9 +95,9 @@ pub(crate) fn check(check: ReuseCheck<'_>) -> ReuseAttempt {
                         attempt.snapshot_ns,
                         rules.racy_window_ns,
                     ),
-                    ..inputs
+                    ..inputs.clone()
                 };
-                rewrite_inputs(&check.store_path, &refreshed);
+                rewrite_inputs(&check.store_path, &slot, &refreshed);
             }
             attempt.replay = Some(output);
             return attempt;
diff --git a/crates/sqlbuild-cache/src/project_reuse/_helpers/entry.rs b/crates/sqlbuild-cache/src/project_reuse/_helpers/entry.rs
index fdd365f3e7..4f6bf849e1 100644
--- a/crates/sqlbuild-cache/src/project_reuse/_helpers/entry.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/_helpers/entry.rs
@@ -1,50 +1,106 @@
-//! One stored compile per selected target, kept as one shared native store file.
+//! One stored compile per selected target: a small inputs file and the output file it names.
 
 use std::path::{Path, PathBuf};
+use std::time::{Duration, SystemTime};
 
-use crate::digest::main::content_digest::content_digest;
 use crate::digest::types::ContentDigest;
-use crate::project_reuse::constants::{
-    REUSE_STORE_ENVIRONMENT, REUSE_STORE_KIND, REUSE_STORE_SUFFIX,
+use crate::project_reuse::_helpers::sections::{
+    SectionFile, decode_sections, encode_sections, write_atomically,
 };
+use crate::project_reuse::constants::{REUSE_OUTPUT_SUFFIX, REUSE_STORE_SUFFIX};
 use crate::project_reuse::models::{SettingsInputs, StoredInputs, StoredOutput, StoredProjectFile};
 use crate::project_snapshot::models::PathStamp;
 use crate::store::errors::StoreDecodeError;
-use crate::store::main::open_native_store::open_native_store;
-use crate::store::models::{NativeStore, RecordReader, RecordWriter};
+use crate::store::models::{RecordReader, RecordWriter};
 
 const KINDS: [&str; 7] = ["f", "fl", "d", "dl", "bl", "s", "p"];
+const SLOT_ROLE: &str = "project-reuse-slot";
+const OUTPUT_ROLE: &str = "project-reuse-output";
+const INPUTS_SECTION: &str = "inputs";
+const OUTPUT_IDENTITY_SECTION: &str = "output";
+const OUTPUT_META_SECTION: &str = "meta";
+const STDOUT_SECTION: &str = "stdout";
+/// How long an output with no slot is kept, so a compile still writing its slot keeps it.
+const ORPHANED_OUTPUT_GRACE: Duration = Duration::from_secs(600);
 
-fn inputs_key() -> ContentDigest {
-    content_digest(&["inputs"])
+/// A slot's verified inputs, the output file it names, and every section it carries.
+#[derive(Debug)]
+pub(crate) struct StoredSlot {
+    pub(crate) inputs: StoredInputs,
+    output_identity: ContentDigest,
+    sections: Vec<(String, Vec<u8>)>,
 }
 
-fn output_key() -> ContentDigest {
-    content_digest(&["output"])
+/// The output file paired with the slot file at `path`.
+fn output_path(path: &Path) -> PathBuf {
+    path.with_extension(REUSE_OUTPUT_SUFFIX)
 }
 
-fn stdout_key(stdout: &str) -> ContentDigest {
-    content_digest(&["stdout", stdout])
-}
-
-/// The stored inputs and output, or `None` when the slot is empty, damaged or foreign.
-pub(crate) fn read_entry(path: &Path) -> Option<(StoredInputs, StoredOutput)> {
-    let mut store: NativeStore =
-        match open_native_store(path, REUSE_STORE_KIND, REUSE_STORE_ENVIRONMENT) {
-            Ok(store) => store,
-            Err(_) => return None,
-        };
-    let Ok(inputs) = decode_inputs(store.get(&inputs_key())?) else {
-        return None;
+/// The slot's inputs and output identity, or `None` when it is empty, damaged or foreign.
+pub(crate) fn read_slot(path: &Path) -> Option<StoredSlot> {
+    let bytes: Vec<u8> = match std::fs::read(path) {
+        Ok(bytes) => bytes,
+        Err(_unreadable) => return None,
     };
-    let Ok((mut output, digest)) = decode_output(store.get(&output_key())?) else {
+    let Ok(slot) = decoded_slot(&bytes) else {
         return None;
     };
-    let Ok(stdout) = String::from_utf8(store.get(&digest)?.to_vec()) else {
+    Some(slot)
+}
+
+fn decoded_slot(bytes: &[u8]) -> Result<StoredSlot, StoreDecodeError> {
+    let file: SectionFile<'_> = decode_sections(bytes, SLOT_ROLE)?;
+    let inputs: StoredInputs = decode_inputs(required(&file, INPUTS_SECTION)?)?;
+    let output_identity: ContentDigest = required(&file, OUTPUT_IDENTITY_SECTION)?
+        .try_into()
+        .map_err(|_| StoreDecodeError)?;
+    Ok(StoredSlot {
+        inputs,
+        output_identity,
+        sections: file
+            .sections
+            .into_iter()
+            .map(|(name, bytes)| (name, bytes.to_vec()))
+            .collect(),
+    })
+}
+
+fn required<'a>(file: &SectionFile<'a>, name: &str) -> Result<&'a [u8], StoreDecodeError> {
+    file.section(name).ok_or(StoreDecodeError)
+}
+
+/// The output `slot` names, or `None` when it is missing, damaged or another compile's.
+pub(crate) fn read_output(path: &Path, slot: &StoredSlot) -> Option<StoredOutput> {
+    let bytes: Vec<u8> = match std::fs::read(output_path(path)) {
+        Ok(bytes) => bytes,
+        Err(_unreadable) => return None,
+    };
+    let Ok(output) = decoded_output(&bytes, &slot.output_identity) else {
         return None;
     };
-    output.stdout = stdout;
-    Some((inputs, output))
+    Some(output)
+}
+
+fn decoded_output(
+    bytes: &[u8],
+    identity: &ContentDigest,
+) -> Result<StoredOutput, StoreDecodeError> {
+    let file: SectionFile<'_> = decode_sections(bytes, OUTPUT_ROLE)?;
+    if file.identity != *identity {
+        return Err(StoreDecodeError);
+    }
+    let mut output: StoredOutput = decode_output(required(&file, OUTPUT_META_SECTION)?)?;
+    output.stdout = String::from_utf8(required(&file, STDOUT_SECTION)?.to_vec())
+        .map_err(|_| StoreDecodeError)?;
+    Ok(output)
+}
+
+/// The stored inputs and output, or `None` when either cannot be read.
+#[cfg(test)]
+pub(crate) fn read_entry(path: &Path) -> Option<(StoredInputs, StoredOutput)> {
+    let slot: StoredSlot = read_slot(path)?;
+    let output: StoredOutput = read_output(path, &slot)?;
+    Some((slot.inputs, output))
 }
 
 /// Replace the slot with one stored compile, then keep at most `max_entries` slots.
@@ -54,31 +110,56 @@ pub(crate) fn write_entry(
     output: &StoredOutput,
     max_entries: usize,
 ) -> std::io::Result<()> {
-    let mut store: NativeStore = NativeStore {
-        kind: REUSE_STORE_KIND.to_owned(),
-        environment: REUSE_STORE_ENVIRONMENT.to_owned(),
-        ..NativeStore::default()
-    };
-    let digest: ContentDigest = stdout_key(&output.stdout);
-    store.put(inputs_key(), encode_inputs(inputs));
-    store.put(output_key(), encode_output(output, &digest));
-    store.put(digest, output.stdout.as_bytes().to_vec());
-    let _ = store.save(path, &[])?;
+    let meta: Vec<u8> = encode_output(output);
+    let (encoded, identity) = encode_sections(
+        OUTPUT_ROLE,
+        &[
+            (OUTPUT_META_SECTION, &meta),
+            (STDOUT_SECTION, output.stdout.as_bytes()),
+        ],
+    );
+    write_atomically(&output_path(path), &encoded)?;
+    drop(encoded);
+    write_slot(
+        path,
+        &[
+            (INPUTS_SECTION, &encode_inputs(inputs)),
+            (OUTPUT_IDENTITY_SECTION, &identity),
+        ],
+    )?;
     prune_entries(path, max_entries);
     Ok(())
 }
 
-/// Replace only the stored inputs, keeping the stored output.
-pub(crate) fn rewrite_inputs(path: &Path, inputs: &StoredInputs) {
-    if let Some((_, output)) = read_entry(path) {
-        let _ = write_entry(path, inputs, &output, usize::MAX);
-    }
+/// Replace only the stored inputs, keeping the output and every other section of `slot`.
+pub(crate) fn rewrite_inputs(path: &Path, slot: &StoredSlot, inputs: &StoredInputs) {
+    let encoded: Vec<u8> = encode_inputs(inputs);
+    let sections: Vec<(&str, &[u8])> = slot
+        .sections
+        .iter()
+        .map(|(name, bytes)| {
+            if name == INPUTS_SECTION {
+                (name.as_str(), encoded.as_slice())
+            } else {
+                (name.as_str(), bytes.as_slice())
+            }
+        })
+        .collect();
+    let _ = write_slot(path, &sections);
+}
+
+fn write_slot(path: &Path, sections: &[(&str, &[u8])]) -> std::io::Result<()> {
+    let (encoded, _) = encode_sections(SLOT_ROLE, sections);
+    write_atomically(path, &encoded)
 }
 
+/// Remove the slot and its output.
 pub(crate) fn remove_entry(path: &Path) {
     let _ = std::fs::remove_file(path);
+    let _ = std::fs::remove_file(output_path(path));
 }
 
+/// Keep at most `max_entries` slots, newest first, and remove outputs no slot names any more.
 fn prune_entries(keep: &Path, max_entries: usize) {
     let Some(directory) = keep.parent() else {
         return;
@@ -86,22 +167,35 @@ fn prune_entries(keep: &Path, max_entries: usize) {
     let Ok(entries) = std::fs::read_dir(directory) else {
         return;
     };
-    let mut stored: Vec<(std::time::SystemTime, PathBuf)> = entries
-        .flatten()
-        .map(|entry| entry.path())
-        .filter(|path| path != keep)
-        .filter(|path| {
-            path.extension()
-                .is_some_and(|suffix| suffix == REUSE_STORE_SUFFIX)
-        })
-        .filter_map(|path| {
-            let modified = std::fs::metadata(&path).and_then(|metadata| metadata.modified());
-            match modified {
-                Ok(modified) => Some((modified, path)),
-                Err(_) => None,
+    let mut stored: Vec<(SystemTime, PathBuf)> = Vec::new();
+    let mut outputs: Vec<(SystemTime, PathBuf)> = Vec::new();
+    for path in entries.flatten().map(|entry| entry.path()) {
+        let Ok(modified) = std::fs::metadata(&path).and_then(|metadata| metadata.modified()) else {
+            continue;
+        };
+        if path
+            .extension()
+            .is_some_and(|suffix| suffix == REUSE_STORE_SUFFIX)
+        {
+            if path != keep {
+                stored.push((modified, path));
             }
-        })
-        .collect();
+        } else if path
+            .extension()
+            .is_some_and(|suffix| suffix == REUSE_OUTPUT_SUFFIX)
+        {
+            outputs.push((modified, path));
+        }
+    }
+    let cutoff: SystemTime = SystemTime::now()
+        .checked_sub(ORPHANED_OUTPUT_GRACE)
+        .unwrap_or(SystemTime::UNIX_EPOCH);
+    for (modified, output) in outputs {
+        let orphaned: bool = !output.with_extension(REUSE_STORE_SUFFIX).exists();
+        if orphaned && modified < cutoff {
+            let _ = std::fs::remove_file(output);
+        }
+    }
     stored.sort_by(|left, right| right.cmp(left));
     for (_, stale) in stored.into_iter().skip(max_entries.saturating_sub(1)) {
         remove_entry(&stale);
@@ -303,7 +397,7 @@ fn decode_inputs(bytes: &[u8]) -> Result<StoredInputs, StoreDecodeError> {
     })
 }
 
-fn encode_output(output: &StoredOutput, stdout: &ContentDigest) -> Vec<u8> {
+fn encode_output(output: &StoredOutput) -> Vec<u8> {
     let mut writer: RecordWriter = RecordWriter::default();
     writer.put_strings(&output.stderr_lines);
     writer.put_i64(output.exit_code);
@@ -315,11 +409,10 @@ fn encode_output(output: &StoredOutput, stdout: &ContentDigest) -> Vec<u8> {
         }
         None => writer.put_u64(0),
     }
-    writer.put_bytes(stdout);
     writer.into_bytes()
 }
 
-fn decode_output(bytes: &[u8]) -> Result<(StoredOutput, ContentDigest), StoreDecodeError> {
+fn decode_output(bytes: &[u8]) -> Result<StoredOutput, StoreDecodeError> {
     let mut reader: RecordReader<'_> = RecordReader::new(bytes);
     let stderr_lines: Vec<String> = reader.strings()?;
     let exit_code: i64 = reader.read_i64()?;
@@ -330,15 +423,11 @@ fn decode_output(bytes: &[u8]) -> Result<(StoredOutput, ContentDigest), StoreDec
     } else {
         None
     };
-    let digest: ContentDigest = reader.bytes()?.try_into().map_err(|_| StoreDecodeError)?;
     reader.finish()?;
-    Ok((
-        StoredOutput {
-            stderr_lines,
-            exit_code,
-            timings_span,
-            stdout: String::new(),
-        },
-        digest,
-    ))
+    Ok(StoredOutput {
+        stderr_lines,
+        exit_code,
+        timings_span,
+        stdout: String::new(),
+    })
 }
diff --git a/crates/sqlbuild-cache/src/project_reuse/_helpers/mod.rs b/crates/sqlbuild-cache/src/project_reuse/_helpers/mod.rs
index a68068cda7..30f44c1302 100644
--- a/crates/sqlbuild-cache/src/project_reuse/_helpers/mod.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/_helpers/mod.rs
@@ -2,5 +2,6 @@ pub(crate) mod attempt;
 pub(crate) mod entry;
 pub(crate) mod files;
 pub(crate) mod identity;
+pub(crate) mod sections;
 pub(crate) mod target;
 pub(crate) mod timings;
diff --git a/crates/sqlbuild-cache/src/project_reuse/_helpers/sections.rs b/crates/sqlbuild-cache/src/project_reuse/_helpers/sections.rs
new file mode 100644
index 0000000000..9a00af3b97
--- /dev/null
+++ b/crates/sqlbuild-cache/src/project_reuse/_helpers/sections.rs
@@ -0,0 +1,118 @@
+//! Files of named, separately checksummed sections that readers may skip or carry forward.
+
+use std::io::Write;
+use std::path::Path;
+
+use crate::digest::main::bytes_digest::bytes_digest;
+use crate::digest::types::ContentDigest;
+use crate::store::errors::StoreDecodeError;
+use crate::store::models::{RecordReader, RecordWriter};
+
+const MAGIC: &[u8; 8] = b"SQBREUSE";
+const FORMAT_VERSION: u32 = 1;
+const DIGEST_BYTES: usize = 32;
+
+/// One file's sections in stored order and the digest identifying the file.
+#[derive(Debug)]
+pub(crate) struct SectionFile<'a> {
+    pub(crate) identity: ContentDigest,
+    pub(crate) sections: Vec<(String, &'a [u8])>,
+}
+
+impl<'a> SectionFile<'a> {
+    /// The bytes of the section called `name`.
+    pub(crate) fn section(&self, name: &str) -> Option<&'a [u8]> {
+        self.sections
+            .iter()
+            .find(|(section, _)| section == name)
+            .map(|(_, bytes)| *bytes)
+    }
+}
+
+/// Encode `sections` for `role`; returns the bytes and the file identity.
+pub(crate) fn encode_sections(role: &str, sections: &[(&str, &[u8])]) -> (Vec<u8>, ContentDigest) {
+    let mut header: RecordWriter = RecordWriter::default();
+    header.put_str(role);
+    header.put_u64(sections.len() as u64);
+    for (name, bytes) in sections {
+        header.put_str(name);
+        header.put_u64(bytes.len() as u64);
+        header.put_bytes(&bytes_digest(bytes));
+    }
+    let header: Vec<u8> = header.into_bytes();
+    let identity: ContentDigest = bytes_digest(&header);
+    let payload: usize = sections.iter().map(|(_, bytes)| bytes.len()).sum();
+    let mut encoded: Vec<u8> =
+        Vec::with_capacity(MAGIC.len() + 12 + header.len() + DIGEST_BYTES + payload);
+    encoded.extend_from_slice(MAGIC);
+    encoded.extend_from_slice(&FORMAT_VERSION.to_le_bytes());
+    encoded.extend_from_slice(&(header.len() as u64).to_le_bytes());
+    encoded.extend_from_slice(&header);
+    encoded.extend_from_slice(&identity);
+    for (_, bytes) in sections {
+        encoded.extend_from_slice(bytes);
+    }
+    (encoded, identity)
+}
+
+/// The verified sections of a `role` file; any other format, role or damage is a decode error.
+pub(crate) fn decode_sections<'a>(
+    bytes: &'a [u8],
+    role: &str,
+) -> Result<SectionFile<'a>, StoreDecodeError> {
+    let rest: &[u8] = bytes
+        .strip_prefix(MAGIC.as_slice())
+        .ok_or(StoreDecodeError)?;
+    let (version, rest) = rest.split_first_chunk::<4>().ok_or(StoreDecodeError)?;
+    if u32::from_le_bytes(*version) != FORMAT_VERSION {
+        return Err(StoreDecodeError);
+    }
+    let (length, rest) = rest.split_first_chunk::<8>().ok_or(StoreDecodeError)?;
+    let length: usize =
+        usize::try_from(u64::from_le_bytes(*length)).map_err(|_| StoreDecodeError)?;
+    let header: &[u8] = rest.get(..length).ok_or(StoreDecodeError)?;
+    let (identity, mut payload) = rest[length..]
+        .split_first_chunk::<DIGEST_BYTES>()
+        .ok_or(StoreDecodeError)?;
+    if bytes_digest(header) != *identity {
+        return Err(StoreDecodeError);
+    }
+    let mut reader: RecordReader<'_> = RecordReader::new(header);
+    if reader.string()? != role {
+        return Err(StoreDecodeError);
+    }
+    let count: u64 = reader.u64()?;
+    let mut sections: Vec<(String, &'a [u8])> = Vec::new();
+    for _ in 0..count {
+        let name: String = reader.string()?;
+        let size: usize = usize::try_from(reader.u64()?).map_err(|_| StoreDecodeError)?;
+        let digest: &[u8] = reader.bytes()?;
+        if size > payload.len() {
+            return Err(StoreDecodeError);
+        }
+        let (section, remaining) = payload.split_at(size);
+        if bytes_digest(section).as_slice() != digest {
+            return Err(StoreDecodeError);
+        }
+        sections.push((name, section));
+        payload = remaining;
+    }
+    reader.finish()?;
+    if !payload.is_empty() {
+        return Err(StoreDecodeError);
+    }
+    Ok(SectionFile {
+        identity: *identity,
+        sections,
+    })
+}
+
+/// Replace `path` with `bytes` through a temporary file in the same folder.
+pub(crate) fn write_atomically(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
+    let directory: &Path = path.parent().unwrap_or_else(|| Path::new("."));
+    std::fs::create_dir_all(directory)?;
+    let mut file: tempfile::NamedTempFile = tempfile::NamedTempFile::new_in(directory)?;
+    file.write_all(bytes)?;
+    let _ = file.persist(path).map_err(|error| error.error)?;
+    Ok(())
+}
diff --git a/crates/sqlbuild-cache/src/project_reuse/_helpers/target.rs b/crates/sqlbuild-cache/src/project_reuse/_helpers/target.rs
index ef404d8c4e..edfecaebac 100644
--- a/crates/sqlbuild-cache/src/project_reuse/_helpers/target.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/_helpers/target.rs
@@ -75,19 +75,23 @@ fn compiled_prefix(project_dir: &Path, rules: &ReuseRules) -> String {
     )
 }
 
+/// Whether `stamp` shows what this compile left there, and whether that read its content.
 fn matches_recorded(
     stamp: &PathStamp,
     artifact: &RecordedArtifact,
     since_ns: i64,
     rules: &ReuseRules,
-) -> bool {
+) -> (bool, bool) {
     match artifact {
         RecordedArtifact::Kept { size, mtime_ns } => {
-            stamp.size == *size && stamp.mtime_ns == *mtime_ns
+            (stamp.size == *size && stamp.mtime_ns == *mtime_ns, false)
         }
         RecordedArtifact::Written { digest } => {
-            stamp.mtime_ns.max(stamp.ctime_ns) < since_ns - rules.racy_window_ns
-                || file_digest(Path::new(&stamp.relative_path)).as_deref() == Some(digest)
+            if stamp.mtime_ns.max(stamp.ctime_ns) < since_ns - rules.racy_window_ns {
+                return (true, false);
+            }
+            let read: Option<String> = file_digest(Path::new(&stamp.relative_path));
+            (read.as_deref() == Some(digest), true)
         }
     }
 }
@@ -114,15 +118,24 @@ pub(crate) fn verified_target_files(
         && stamps
             .keys()
             .all(|path| expected.contains_key(path.as_str()));
-    if !keys_match
-        || !stamps
-            .iter()
-            .all(|(path, stamp)| matches_recorded(stamp, expected[path.as_str()], since_ns, rules))
-    {
+    if !keys_match {
         return None;
     }
-    if target_stamps(project_dir, &paths, written, rules) != stamps {
-        return None;
+    let mut read_paths: Vec<&str> = Vec::new();
+    for (path, stamp) in &stamps {
+        let (matched, read) = matches_recorded(stamp, expected[path.as_str()], since_ns, rules);
+        if !matched {
+            return None;
+        }
+        if read {
+            read_paths.push(path.as_str());
+        }
+    }
+    for path in read_paths {
+        let restamped: Option<PathStamp> = path_stamp(path);
+        if restamped.as_ref() != stamps.get(path) {
+            return None;
+        }
     }
     Some(stamps.into_values().collect())
 }
diff --git a/crates/sqlbuild-cache/src/project_reuse/constants.rs b/crates/sqlbuild-cache/src/project_reuse/constants.rs
index 1c61e3745a..c5407b5c90 100644
--- a/crates/sqlbuild-cache/src/project_reuse/constants.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/constants.rs
@@ -1,8 +1,6 @@
-/// The native store kind holding one stored compile per selected target.
-pub const REUSE_STORE_KIND: &str = "project-reuse";
-/// The store environment; bumped when the stored records change shape.
-pub const REUSE_STORE_ENVIRONMENT: &str = "project-reuse-records-1";
-/// Suffix of every reuse store file.
+/// Suffix of every stored compile's slot file, which holds its inputs.
 pub const REUSE_STORE_SUFFIX: &str = "store";
+/// Suffix of the file beside each slot file holding the stored compile's output.
+pub const REUSE_OUTPUT_SUFFIX: &str = "output";
 /// Prefix of digests computed natively, so they never equal a Python digest.
 pub const NATIVE_DIGEST_PREFIX: &str = "b3:";
diff --git a/crates/sqlbuild-cache/src/project_reuse/tests/helpers.rs b/crates/sqlbuild-cache/src/project_reuse/tests/helpers.rs
index 3bc8626078..14b3da8483 100644
--- a/crates/sqlbuild-cache/src/project_reuse/tests/helpers.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/tests/helpers.rs
@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
 use std::path::Path;
 
 use crate::project_reuse::_helpers::entry::write_entry;
+use crate::project_reuse::_helpers::sections::{decode_sections, encode_sections};
 use crate::project_reuse::models::{SettingsInputs, StoredInputs, StoredOutput, StoredProjectFile};
 use crate::project_snapshot::models::PathStamp;
 
@@ -73,6 +74,21 @@ pub(crate) fn store_aged_slots(directory: &Path, slots: u64, max_entries: usize)
     }
 }
 
+/// Leave output files with no slot in `directory`, each modified `age` seconds ago.
+pub(crate) fn leave_orphaned_outputs(directory: &Path, outputs: &[(&str, u64)]) {
+    std::fs::create_dir_all(directory).expect("store folder");
+    for (name, age) in outputs {
+        let path = directory.join(name);
+        std::fs::write(&path, b"output").expect("orphaned output");
+        let modified = std::time::SystemTime::now() - std::time::Duration::from_secs(*age);
+        std::fs::File::options()
+            .write(true)
+            .open(&path)
+            .and_then(|file| file.set_modified(modified))
+            .expect("aged output");
+    }
+}
+
 /// The slot file names left in `directory`, sorted.
 pub(crate) fn slot_names(directory: &Path) -> Vec<String> {
     let mut names: Vec<String> = std::fs::read_dir(directory)
@@ -89,6 +105,58 @@ pub(crate) fn slot_names(directory: &Path) -> Vec<String> {
     names
 }
 
+/// The output file stored beside the slot file at `path`.
+pub(crate) fn output_file(path: &Path) -> std::path::PathBuf {
+    path.with_extension("output")
+}
+
+/// Keep only the first `length` bytes of the file.
+pub(crate) fn truncate(path: &Path, length: usize) {
+    let bytes = std::fs::read(path).expect("file bytes");
+    std::fs::write(path, &bytes[..length]).expect("truncated file");
+}
+
+/// Swap in the intact output of a compile that printed something else.
+pub(crate) fn replace_output(path: &Path) {
+    let other = path.with_file_name("other.store");
+    let output = StoredOutput {
+        stdout: "{\"command\": \"build\"}\n".to_owned(),
+        ..stored_output()
+    };
+    write_entry(&other, &stored_inputs(), &output, usize::MAX).expect("other compile");
+    std::fs::rename(output_file(&other), output_file(path)).expect("swapped output");
+}
+
+/// A slot file as the previous single-file native store format wrote it.
+pub(crate) fn previous_format_slot() -> Vec<u8> {
+    let mut bytes: Vec<u8> = b"SQBSTORE".to_vec();
+    bytes.extend_from_slice(&1_u32.to_le_bytes());
+    bytes.extend_from_slice(&[0_u8; 32]);
+    bytes.extend_from_slice(b"project-reuse");
+    bytes
+}
+
+/// Append a section named `name` to the slot file, as a later release might.
+pub(crate) fn add_slot_section(path: &Path, name: &str, bytes: &[u8]) {
+    let stored = std::fs::read(path).expect("slot bytes");
+    let file = decode_sections(&stored, "project-reuse-slot").expect("slot sections");
+    let mut sections: Vec<(&str, &[u8])> = file
+        .sections
+        .iter()
+        .map(|(name, bytes)| (name.as_str(), *bytes))
+        .collect();
+    sections.push((name, bytes));
+    let (encoded, _) = encode_sections("project-reuse-slot", &sections);
+    std::fs::write(path, encoded).expect("extended slot");
+}
+
+/// The bytes of the slot file's section named `name`.
+pub(crate) fn slot_section(path: &Path, name: &str) -> Option<Vec<u8>> {
+    let stored = std::fs::read(path).expect("slot bytes");
+    let file = decode_sections(&stored, "project-reuse-slot").expect("slot sections");
+    file.section(name).map(<[u8]>::to_vec)
+}
+
 /// Flip one byte near the end of the file.
 pub(crate) fn damage(path: &Path) {
     let mut bytes = std::fs::read(path).expect("slot bytes");
diff --git a/crates/sqlbuild-cache/src/project_reuse/tests/test_project_reuse.rs b/crates/sqlbuild-cache/src/project_reuse/tests/test_project_reuse.rs
index 13296a9d8b..ff85acfbe6 100644
--- a/crates/sqlbuild-cache/src/project_reuse/tests/test_project_reuse.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/tests/test_project_reuse.rs
@@ -1,12 +1,15 @@
-use crate::project_reuse::_helpers::entry::{read_entry, write_entry};
+use crate::project_reuse::_helpers::entry::{read_entry, read_slot, rewrite_inputs, write_entry};
 use crate::project_reuse::_helpers::identity::search_path_stamps;
 use crate::project_reuse::_helpers::timings::{compile_timings_span, replace_compile_timings};
-use crate::project_reuse::models::ReuseRules;
+use crate::project_reuse::models::{ReuseRules, StoredInputs};
 use crate::project_reuse::tests::helpers::{
-    damage, slot_names, store_aged_slots, stored_inputs, stored_output,
+    add_slot_section, damage, leave_orphaned_outputs, output_file, previous_format_slot,
+    replace_output, slot_names, slot_section, store_aged_slots, stored_inputs, stored_output,
+    truncate,
 };
 use crate::project_reuse::tests::test_types::{
-    EntryReadTestCase, NonUnicodeNameTestCase, SlotPruneTestCase, TimingsSpanTestCase,
+    EntryReadTestCase, NonUnicodeNameTestCase, SlotPruneTestCase, SlotRefreshTestCase,
+    TimingsSpanTestCase,
 };
 use crate::project_snapshot::main::path_text::path_text;
 use crate::project_snapshot::main::snapshot_project_files::snapshot_project_files;
@@ -28,7 +31,7 @@ fn given_stored_compile_when_reading_then_only_an_intact_slot_round_trips() {
         },
         EntryReadTestCase {
             description: "truncated slot",
-            edit: |path| std::fs::write(path, b"SQBSTORE").expect("truncated"),
+            edit: |path| truncate(path, 40),
             expected_read: false,
         },
         EntryReadTestCase {
@@ -36,6 +39,31 @@ fn given_stored_compile_when_reading_then_only_an_intact_slot_round_trips() {
             edit: |path| std::fs::remove_file(path).expect("removed"),
             expected_read: false,
         },
+        EntryReadTestCase {
+            description: "damaged output",
+            edit: |path| damage(&output_file(path)),
+            expected_read: false,
+        },
+        EntryReadTestCase {
+            description: "truncated output",
+            edit: |path| truncate(&output_file(path), 60),
+            expected_read: false,
+        },
+        EntryReadTestCase {
+            description: "missing output",
+            edit: |path| std::fs::remove_file(output_file(path)).expect("removed"),
+            expected_read: false,
+        },
+        EntryReadTestCase {
+            description: "output of another compile",
+            edit: replace_output,
+            expected_read: false,
+        },
+        EntryReadTestCase {
+            description: "slot of the previous single-file format",
+            edit: |path| std::fs::write(path, previous_format_slot()).expect("old slot"),
+            expected_read: false,
+        },
     ];
 
     for test_case in test_cases {
@@ -64,19 +92,35 @@ fn given_more_slots_than_retained_when_storing_then_the_oldest_are_pruned() {
             description: "within the limit",
             slots: 2,
             max_entries: 8,
-            expected_kept: &["slot0.store", "slot1.store"],
+            orphaned_outputs: &[],
+            expected_kept: &["slot0.output", "slot0.store", "slot1.output", "slot1.store"],
         },
         SlotPruneTestCase {
             description: "over the limit",
             slots: 4,
             max_entries: 2,
-            expected_kept: &["slot2.store", "slot3.store"],
+            orphaned_outputs: &[],
+            expected_kept: &["slot2.output", "slot2.store", "slot3.output", "slot3.store"],
+        },
+        SlotPruneTestCase {
+            description: "outputs left without a slot",
+            slots: 2,
+            max_entries: 8,
+            orphaned_outputs: &[("crashed.output", 3_600), ("in_flight.output", 0)],
+            expected_kept: &[
+                "in_flight.output",
+                "slot0.output",
+                "slot0.store",
+                "slot1.output",
+                "slot1.store",
+            ],
         },
     ];
 
     for test_case in test_cases {
         let folder = tempfile::tempdir().expect("store folder");
         let directory = folder.path().join("reuse");
+        leave_orphaned_outputs(&directory, test_case.orphaned_outputs);
 
         store_aged_slots(&directory, test_case.slots, test_case.max_entries);
 
@@ -213,3 +257,47 @@ fn given_search_path_entry_that_is_not_utf8_when_stamping_then_its_real_folder_i
         assert!(stamps[0].1 > 0, "{}", test_case.description);
     }
 }
+
+#[test]
+fn given_stored_compile_when_refreshing_inputs_then_only_the_slot_file_is_rewritten() {
+    let test_cases = [SlotRefreshTestCase {
+        description: "slot carrying a section this release does not read",
+        extra_section: ("project-image", b"orders index"),
+        expected_output_rewritten: false,
+    }];
+
+    for test_case in test_cases {
+        let folder = tempfile::tempdir().expect("store folder");
+        let path = folder.path().join("reuse/slot.store");
+        write_entry(&path, &stored_inputs(), &stored_output(), 8).expect("stored compile");
+        let (name, bytes) = test_case.extra_section;
+        add_slot_section(&path, name, bytes);
+        let output_before = std::fs::read(output_file(&path)).expect("output bytes");
+        let slot = read_slot(&path).expect("slot");
+        let refreshed = StoredInputs {
+            invocation_digest: "b3:refreshed".to_owned(),
+            ..stored_inputs()
+        };
+
+        rewrite_inputs(&path, &slot, &refreshed);
+
+        assert_eq!(
+            read_entry(&path),
+            Some((refreshed, stored_output())),
+            "{}: refreshed inputs read back with the kept output",
+            test_case.description
+        );
+        assert_eq!(
+            std::fs::read(output_file(&path)).expect("output bytes") != output_before,
+            test_case.expected_output_rewritten,
+            "{}: output file",
+            test_case.description
+        );
+        assert_eq!(
+            slot_section(&path, name),
+            Some(bytes.to_vec()),
+            "{}: unread sections are carried forward",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-cache/src/project_reuse/tests/test_types.rs b/crates/sqlbuild-cache/src/project_reuse/tests/test_types.rs
index b1369a1620..9e2f58b920 100644
--- a/crates/sqlbuild-cache/src/project_reuse/tests/test_types.rs
+++ b/crates/sqlbuild-cache/src/project_reuse/tests/test_types.rs
@@ -12,9 +12,18 @@ pub(crate) struct SlotPruneTestCase {
     pub(crate) description: &'static str,
     pub(crate) slots: u64,
     pub(crate) max_entries: usize,
+    /// Output files left without a slot before storing, each with its age in seconds.
+    pub(crate) orphaned_outputs: &'static [(&'static str, u64)],
     pub(crate) expected_kept: &'static [&'static str],
 }
 
+/// A stored compile whose slot carries an extra section when its inputs are refreshed.
+pub(crate) struct SlotRefreshTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) extra_section: (&'static str, &'static [u8]),
+    pub(crate) expected_output_rewritten: bool,
+}
+
 /// One report and the compile_timings span expected inside it.
 pub(crate) struct TimingsSpanTestCase {
     pub(crate) description: &'static str,
diff --git a/crates/sqlbuild-cache/src/store/_helpers/frame.rs b/crates/sqlbuild-cache/src/store/_helpers/frame.rs
index 91e8028df9..be7d536ea8 100644
--- a/crates/sqlbuild-cache/src/store/_helpers/frame.rs
+++ b/crates/sqlbuild-cache/src/store/_helpers/frame.rs
@@ -1,6 +1,6 @@
 //! Encode and verify whole store files.
 
-use crate::digest::main::content_digest::content_digest;
+use crate::digest::main::bytes_digest::bytes_digest;
 use crate::digest::types::ContentDigest;
 use crate::store::constants::{STORE_FORMAT_VERSION, STORE_MAGIC};
 use crate::store::errors::StoreDecodeError;
@@ -15,7 +15,7 @@ pub(crate) struct StoreFile {
     pub(crate) entries: Vec<(ContentDigest, u64, Vec<u8>)>,
 }
 
-/// Magic, format version, a digest of the body, then the body.
+/// Magic, format version, the BLAKE3 digest of the body, then the body.
 pub(crate) fn encode_store(file: &StoreFile) -> Vec<u8> {
     let mut body: RecordWriter = RecordWriter::default();
     body.put_str(&file.kind);
@@ -32,7 +32,7 @@ pub(crate) fn encode_store(file: &StoreFile) -> Vec<u8> {
     let mut bytes: Vec<u8> = Vec::with_capacity(body.len() + 44);
     bytes.extend_from_slice(STORE_MAGIC);
     bytes.extend_from_slice(&STORE_FORMAT_VERSION.to_le_bytes());
-    bytes.extend_from_slice(&content_digest(&[&body]));
+    bytes.extend_from_slice(&bytes_digest(&body));
     bytes.extend_from_slice(&body);
     bytes
 }
@@ -47,7 +47,7 @@ pub(crate) fn decode_store(bytes: &[u8]) -> Result<StoreFile, StoreDecodeError>
         return Err(StoreDecodeError);
     }
     let (checksum, body) = rest.split_first_chunk::<32>().ok_or(StoreDecodeError)?;
-    if content_digest(&[body]) != *checksum {
+    if bytes_digest(body) != *checksum {
         return Err(StoreDecodeError);
     }
     let mut reader: RecordReader<'_> = RecordReader::new(body);
diff --git a/crates/sqlbuild-cache/src/store/constants.rs b/crates/sqlbuild-cache/src/store/constants.rs
index ffda20c06d..58db216cc9 100644
--- a/crates/sqlbuild-cache/src/store/constants.rs
+++ b/crates/sqlbuild-cache/src/store/constants.rs
@@ -3,6 +3,6 @@
 /// The first bytes of every store file.
 pub const STORE_MAGIC: &[u8; 8] = b"SQBSTORE";
 /// Bumped whenever the file layout changes; older files are discarded whole.
-pub const STORE_FORMAT_VERSION: u32 = 1;
+pub const STORE_FORMAT_VERSION: u32 = 2;
 /// Saves an unused entry survives while the store's environment is unchanged.
 pub const RETAINED_GENERATIONS: u64 = 64;
diff --git a/crates/sqlbuild-cache/src/store/tests/helpers.rs b/crates/sqlbuild-cache/src/store/tests/helpers.rs
index 044e993772..cc0f7d15fc 100644
--- a/crates/sqlbuild-cache/src/store/tests/helpers.rs
+++ b/crates/sqlbuild-cache/src/store/tests/helpers.rs
@@ -2,6 +2,7 @@ use std::path::Path;
 
 use crate::digest::main::content_digest::content_digest;
 use crate::digest::types::ContentDigest;
+use crate::store::constants::STORE_FORMAT_VERSION;
 use crate::store::errors::StoreDecodeError;
 use crate::store::main::open_native_store::open_native_store;
 use crate::store::models::{NativeStore, RecordReader, RecordWriter};
@@ -34,6 +35,16 @@ pub(crate) fn damage(path: &Path) {
     std::fs::write(path, bytes).expect("damaged store");
 }
 
+/// Relabel the file as the previous format version, whose body digest was SHA-256.
+pub(crate) fn rewrite_as_previous_format(path: &Path) {
+    let mut bytes = std::fs::read(path).expect("store bytes");
+    let previous: u32 = STORE_FORMAT_VERSION - 1;
+    bytes[8..12].copy_from_slice(&previous.to_le_bytes());
+    let digest = content_digest(&[&bytes[44..]]);
+    bytes[12..44].copy_from_slice(&digest);
+    std::fs::write(path, bytes).expect("previous format store");
+}
+
 /// A record holding a string and a number.
 pub(crate) fn record() -> Vec<u8> {
     let mut writer = RecordWriter::default();
diff --git a/crates/sqlbuild-cache/src/store/tests/test_native_store.rs b/crates/sqlbuild-cache/src/store/tests/test_native_store.rs
index bd6128a871..c12e136466 100644
--- a/crates/sqlbuild-cache/src/store/tests/test_native_store.rs
+++ b/crates/sqlbuild-cache/src/store/tests/test_native_store.rs
@@ -1,7 +1,8 @@
 use crate::store::errors::StoreDecodeError;
 use crate::store::main::open_native_store::open_native_store;
 use crate::store::tests::helpers::{
-    ENVIRONMENT, KIND, METADATA, VALUE, damage, key, open, read_record, record, saved_store,
+    ENVIRONMENT, KIND, METADATA, VALUE, damage, key, open, read_record, record,
+    rewrite_as_previous_format, saved_store,
 };
 use crate::store::tests::test_types::{
     NativeStoreDiscardTestCase, NativeStoreOpenTestCase, NativeStoreRetentionTestCase,
@@ -51,6 +52,14 @@ fn given_saved_store_when_opening_then_only_the_same_kind_and_environment_load()
             expected_value: None,
             expected_metadata: b"",
         },
+        NativeStoreOpenTestCase {
+            description: "file of the previous format version",
+            kind: KIND,
+            environment: ENVIRONMENT,
+            edit: rewrite_as_previous_format,
+            expected_value: None,
+            expected_metadata: b"",
+        },
         NativeStoreOpenTestCase {
             description: "missing file",
             kind: KIND,
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/constants.rs b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/constants.rs
index 649bcd6d24..212576852e 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/constants.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/constants.rs
@@ -5,9 +5,10 @@ use sqlbuild_core::constants::PROJECT_CONFIG_FILENAME;
 use crate::bindings::_helpers::frontier::stage_capture::CAPTURE_DIR_ENV_VAR;
 
 /// The directory under the compiler cache directory holding the native reuse store.
-pub(crate) const NATIVE_REUSE_DIRECTORY_NAME: &str = "project-reuse-v2";
-/// The reuse directory an older release kept stored compiles in.
-pub(crate) const RETIRED_REUSE_DIRECTORY_NAME: &str = "project-reuse-v1";
+pub(crate) const NATIVE_REUSE_DIRECTORY_NAME: &str = "project-reuse-v3";
+/// The reuse directories older releases kept stored compiles in.
+pub(crate) const RETIRED_REUSE_DIRECTORY_NAMES: [&str; 2] =
+    ["project-reuse-v1", "project-reuse-v2"];
 pub(crate) const REUSE_MAX_ENTRY_BYTES: i64 = 512 * 1024 * 1024;
 pub(crate) const REUSE_MAX_STORED_ENTRIES: i64 = 8;
 pub(crate) const REUSE_DISABLE_ENV_VAR: &str = "SQLBUILD_DISABLE_COMPILE_REUSE";
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/reuse_attempt.rs b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/reuse_attempt.rs
index dd54e6847b..38fff9bfc4 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/reuse_attempt.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/reuse_attempt.rs
@@ -21,7 +21,7 @@ use crate::bindings::_helpers::compile_command::constants::{
     MISSING_ENVIRONMENT_VALUE, MISSING_FILE_DIGEST, MISSING_PATH_MTIME_NS,
     NATIVE_REUSE_DIRECTORY_NAME, PRESENCE_ONLY_FILE_SUFFIXES, PROJECT_CONFIG_FILENAMES,
     PROJECT_ROOT_PATH_MTIME_NS, RACY_WINDOW_ATTRIBUTE, RACY_WINDOW_NS,
-    RETIRED_REUSE_DIRECTORY_NAME, REUSE_BYPASS_TIMING, REUSE_CHECK_TIMING, REUSE_DISABLE_ENV_VAR,
+    RETIRED_REUSE_DIRECTORY_NAMES, REUSE_BYPASS_TIMING, REUSE_CHECK_TIMING, REUSE_DISABLE_ENV_VAR,
     REUSE_DISABLE_VALUE, REUSE_HIT_MESSAGE, REUSE_HIT_TIMING, REUSE_LOGGER_NAME,
     REUSE_MAX_ENTRY_BYTES, REUSE_MAX_STORED_ENTRIES, REUSE_MISS_TIMING, TARGET_DIRECTORY_NAME,
     TOTAL_TIMING, TRACKED_ENVIRONMENT_PREFIXES, UNTRACKED_ENVIRONMENT_NAMES,
@@ -129,7 +129,7 @@ pub(crate) fn reuse_rules(py: Python<'_>, with_output_files: bool) -> PyResult<B
     )?;
     rules.set_item(
         "retired_directories",
-        PyList::new(py, [RETIRED_REUSE_DIRECTORY_NAME])?,
+        PyList::new(py, RETIRED_REUSE_DIRECTORY_NAMES)?,
     )?;
     Ok(rules)
 }
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/target_writer.rs b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/target_writer.rs
index 51a6ac15b8..20a731b1df 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/target_writer.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/target_writer.rs
@@ -15,12 +15,13 @@ use sqlbuild_cache::artifacts::models::SqlTestArtifactRecord;
 
 use crate::bindings::_helpers::analysis::sql_test_comparisons::single_comparison_sql;
 use crate::bindings::_helpers::compile_command::artifact_batch::ArtifactBatch;
+use crate::bindings::_helpers::compile_command::constants::RACY_WINDOW_NS;
 use crate::bindings::_helpers::compile_command::reuse_recording::{
     ArtifactRecorder, CompileArtifactWrites, artifact_kept, artifact_moved, artifact_tree_moved,
 };
 use crate::bindings::_helpers::compile_command::test_artifact_cache::{
-    IdentityContext, artifact_identity, artifact_record_key, identity_context, read_cache,
-    write_cache,
+    IdentityContext, artifact_identity_payload, artifact_record_key_payload, identity_context,
+    payload_digests, read_cache, write_cache, yield_interpreter,
 };
 use crate::bindings::_helpers::compile_driver::compile_timings::record_compile_timing;
 use crate::bindings::_helpers::compile_driver::python_calls::{
@@ -54,6 +55,8 @@ const GENERIC_DIR: &str = "generic";
 const SINGULAR_DIR: &str = "singular";
 const TESTS_DIR: &str = "tests";
 const MANIFEST_FILE: &str = "manifest.json";
+/// The compile cache file stamping the compiled tree the last stale-file walk left.
+const COMPILED_TREE_STAMP_FILE: &str = "compiled-tree.bin";
 const SQL_FILE_SUFFIX: &str = ".sql";
 /// `functions/<language>` path components that already lead a function's path.
 const LANGUAGE_DIR_DEPTH: usize = 2;
@@ -173,12 +176,18 @@ fn function_sql(adapter: &Bound<'_, PyAny>, function: &Bound<'_, PyAny>) -> PyRe
     Ok(statements.join(";\n\n"))
 }
 
-/// Delete compiled files no write group manages.
-fn remove_stale(target_dir: &Bound<'_, PyAny>, managed: Vec<PathBuf>) -> PyResult<()> {
+/// Delete compiled files no write group manages, skipping the walk while the tree is unchanged.
+fn remove_stale(
+    target_dir: &Bound<'_, PyAny>,
+    managed: Vec<PathBuf>,
+    cache_dir: Option<PathBuf>,
+) -> PyResult<()> {
     let py: Python<'_> = target_dir.py();
     let compiled: PathBuf = joined(target_dir, &[COMPILED_DIR])?.extract()?;
+    let stamp: Option<(PathBuf, i64)> =
+        cache_dir.map(|directory| (directory.join(COMPILED_TREE_STAMP_FILE), RACY_WINDOW_NS));
     record_compile_timing("stale_traversal_ms", || {
-        remove_stale_artifacts(py, compiled, managed).map(|_| ())
+        remove_stale_artifacts(py, compiled, managed, stamp).map(|_| ())
     })
 }
 
@@ -307,7 +316,7 @@ fn write_planned_target<'py>(
     }
     batch.flush(py)?;
     if remove_stale_files {
-        remove_stale(target_dir, extracted(&managed)?)?;
+        remove_stale(target_dir, extracted(&managed)?, None)?;
     }
     write_manifest(target_dir, manifest, None)?;
     written_target(
@@ -413,6 +422,32 @@ fn planned_artifacts<'py>(
     })
 }
 
+/// Each test's cache slot and artifact identity, hashed together without holding the interpreter.
+fn test_identities<'py>(
+    tests: &[Bound<'py, PyAny>],
+    context: &IdentityContext<'py>,
+    chains: &Bound<'py, PyAny>,
+) -> PyResult<Vec<(String, String)>> {
+    let Some(first) = tests.first() else {
+        return Ok(Vec::new());
+    };
+    let py: Python<'py> = first.py();
+    let empty: Bound<'py, PyTuple> = PyTuple::empty(py);
+    let mut payloads: Vec<String> = Vec::with_capacity(tests.len() * 2);
+    for (index, test) in tests.iter().enumerate() {
+        let chain: Bound<'py, PyAny> = mapping_get_or(chains, native_field(test, "key")?, &empty)?;
+        payloads.push(artifact_record_key_payload(test)?);
+        payloads.push(artifact_identity_payload(test, &chain, context)?);
+        yield_interpreter(py, index);
+    }
+    let mut digests = payload_digests(py, &payloads).into_iter();
+    let mut identities: Vec<(String, String)> = Vec::with_capacity(tests.len());
+    while let (Some(record_key), Some(identity)) = (digests.next(), digests.next()) {
+        identities.push((record_key, identity));
+    }
+    Ok(identities)
+}
+
 /// Plan every SQL test whose cached artifact under `target_dir` cannot be reused.
 fn plan_tests<'py>(
     target_dir: &Bound<'py, PyAny>,
@@ -434,19 +469,22 @@ fn plan_tests<'py>(
             sql_test_model_chain_names_by_key(project, &tests)?.into_any();
         Some((context, chains))
     };
+    let tests: Vec<Bound<'py, PyAny>> = tests.try_iter()?.collect::<PyResult<_>>()?;
+    let identities: Vec<Option<(String, String)>> = match &caching {
+        None => vec![None; tests.len()],
+        Some((context, chains)) => test_identities(&tests, context, chains)?
+            .into_iter()
+            .map(Some)
+            .collect(),
+    };
     let mut pending: Vec<PendingTest> = Vec::new();
     let mut cached_paths: Vec<Py<PyAny>> = Vec::new();
     let mut cached_records: Vec<(String, SqlTestArtifactRecord)> = Vec::new();
-    let empty: Bound<'py, PyTuple> = PyTuple::empty(py);
-    for test in tests.try_iter()? {
-        let test: Bound<'py, PyAny> = test?;
-        let Some((context, chains)) = &caching else {
+    for (index, (test, identity)) in tests.into_iter().zip(identities).enumerate() {
+        let Some((record_key, identity)) = identity else {
             pending.push((test.unbind(), None, None));
             continue;
         };
-        let record_key: String = artifact_record_key(&test)?;
-        let chain: Bound<'py, PyAny> = mapping_get_or(chains, native_field(&test, "key")?, &empty)?;
-        let identity: String = artifact_identity(&test, &chain, context)?;
         let reusable: Option<SqlTestArtifactRecord> = cached
             .get(&record_key)
             .and_then(|record| reusable_sql_test_record(&tests_root_path, record, &identity));
@@ -455,9 +493,10 @@ fn plan_tests<'py>(
             artifact_kept(recorder, &path, (record.size, record.mtime_ns))?;
             cached_paths.push(path.unbind());
             cached_records.push((record_key, record));
-            continue;
+        } else {
+            pending.push((test.unbind(), Some(record_key), Some(identity)));
         }
-        pending.push((test.unbind(), Some(record_key), Some(identity)));
+        yield_interpreter(py, index);
     }
     let pending_tests: Vec<Bound<'py, PyAny>> = pending
         .iter()
@@ -648,7 +687,13 @@ pub(crate) fn write_static_target<'py>(
     if remove_stale_files {
         let mut paths: Vec<PathBuf> = managed_models;
         paths.extend(extracted(&managed)?);
-        remove_stale(target_dir, paths)?;
+        let cache_dir: Bound<'py, PyAny> = native_field(project, "compile_cache_dir")?;
+        let cache_dir: Option<PathBuf> = if cache_dir.is_none() {
+            None
+        } else {
+            Some(cache_dir.extract()?)
+        };
+        remove_stale(target_dir, paths, cache_dir)?;
     }
     write_manifest(target_dir, manifest, prepared.recorder)?;
     written_target(
@@ -720,7 +765,7 @@ pub(crate) fn publish_staged_target<'py>(
             managed.push(path_class(py)?.call1((path,))?);
         }
         if existed {
-            remove_stale(target_dir, extracted(&managed)?)?;
+            remove_stale(target_dir, extracted(&managed)?, None)?;
         }
     }
     write_manifest(target_dir, manifest, recorder)?;
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/test_artifact_cache.rs b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/test_artifact_cache.rs
index 254504c00f..97c3b49e45 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compile_command/test_artifact_cache.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compile_command/test_artifact_cache.rs
@@ -6,17 +6,21 @@ use std::path::PathBuf;
 use pyo3::IntoPyObjectExt;
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult, Python};
 use pyo3::types::{PyBool, PyDict, PyDictMethods, PyInt, PyList, PyListMethods, PyString, PyTuple};
+use rayon::prelude::{IntoParallelRefIterator, ParallelIterator};
 use sqlbuild_cache::artifacts::main::stored_sql_test_record::stored_sql_test_record;
 use sqlbuild_cache::artifacts::models::{SqlTestArtifactRecord, StoredSqlTestRecord};
-use sqlbuild_cache::digest::main::sha256_hex::sha256_hex;
+use sqlbuild_cache::digest::main::bytes_digest::bytes_digest;
+use sqlbuild_cache::digest::main::hex_digest::hex_digest;
 
 use crate::bindings::_helpers::compile_driver::python_calls::imported;
 use crate::bindings::_helpers::compile_outputs::json_report::{compact_sorted_json, orjson_sorted};
 use crate::bindings::_helpers::records::construct::record_class;
 use crate::bindings::_helpers::records::ids::RecordId;
 
-const CACHE_VERSION: i64 = 4;
-const ALGORITHM_FINGERPRINT: &str = "sql-test-artifact-v4-orjson-identity";
+const CACHE_VERSION: i64 = 5;
+const ALGORITHM_FINGERPRINT: &str = "sql-test-artifact-v5-orjson-blake3-identity";
+/// Items a loop holding the interpreter handles before letting other threads run.
+const INTERPRETER_YIELD_INTERVAL: usize = 64;
 const CACHE_FILE_NAME: &str = "sql-test-artifacts.json";
 const MAX_CACHE_BYTES: u64 = 10_000_000;
 const MAX_CACHE_RECORDS: usize = 100_000;
@@ -28,10 +32,26 @@ pub(crate) struct IdentityContext<'py> {
     model_identities: Bound<'py, PyDict>,
 }
 
-/// `hashlib.sha256(orjson.dumps(payload, option=OPT_SORT_KEYS)).hexdigest()`.
-fn payload_digest(payload: &Bound<'_, PyAny>) -> PyResult<String> {
-    let encoded: String = orjson_sorted(payload)?;
-    Ok(sha256_hex(encoded.as_bytes()))
+/// `orjson.dumps(payload, option=OPT_SORT_KEYS)`, the text an identity digest is taken of.
+fn payload_text(payload: &Bound<'_, PyAny>) -> PyResult<String> {
+    orjson_sorted(payload)
+}
+
+/// The hexadecimal BLAKE3 digest of each payload text, hashed without holding the interpreter.
+pub(crate) fn payload_digests(py: Python<'_>, payloads: &[String]) -> Vec<String> {
+    py.detach(|| {
+        payloads
+            .par_iter()
+            .map(|payload| hex_digest(&bytes_digest(payload.as_bytes())))
+            .collect()
+    })
+}
+
+/// Let other threads take the interpreter every few items of a long loop that holds it.
+pub(crate) fn yield_interpreter(py: Python<'_>, index: usize) {
+    if index % INTERPRETER_YIELD_INTERVAL == INTERPRETER_YIELD_INTERVAL - 1 {
+        py.detach(|| ());
+    }
 }
 
 fn dict<'py>(
@@ -77,10 +97,10 @@ fn sorted_deps<'py>(deps: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
     sorted(pairs)
 }
 
-fn model_identity(model: &Bound<'_, PyAny>) -> PyResult<String> {
+fn model_identity_payload(model: &Bound<'_, PyAny>) -> PyResult<String> {
     let py: Python<'_> = model.py();
     let values: Bound<'_, PyAny> = model.getattr("config")?.getattr("values")?;
-    payload_digest(
+    payload_text(
         dict(
             py,
             vec![
@@ -170,13 +190,23 @@ pub(crate) fn identity_context<'py>(
             ("function_locations", sorted(functions)?),
         ],
     )?;
-    let model_identities: Bound<'py, PyDict> = PyDict::new(py);
-    for model in project.getattr("models")?.try_iter()? {
+    let mut names: Vec<Bound<'py, PyAny>> = Vec::new();
+    let mut payloads: Vec<String> = Vec::new();
+    for (index, model) in project.getattr("models")?.try_iter()?.enumerate() {
         let model: Bound<'py, PyAny> = model?;
-        model_identities.set_item(model.getattr("name")?, model_identity(&model)?)?;
+        names.push(model.getattr("name")?);
+        payloads.push(model_identity_payload(&model)?);
+        yield_interpreter(py, index);
+    }
+    payloads.push(payload_text(common.as_any())?);
+    let mut digests: Vec<String> = payload_digests(py, &payloads);
+    let common_identity: String = digests.pop().unwrap_or_default();
+    let model_identities: Bound<'py, PyDict> = PyDict::new(py);
+    for (name, digest) in names.into_iter().zip(digests) {
+        model_identities.set_item(name, digest)?;
     }
     Ok(IdentityContext {
-        common_identity: payload_digest(common.as_any())?,
+        common_identity,
         model_identities,
     })
 }
@@ -287,8 +317,8 @@ fn test_payload<'py>(test: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyDict>> {
     Ok(payload)
 }
 
-/// The complete identity of one rendered comparison artifact.
-pub(crate) fn artifact_identity(
+/// The text whose digest is the complete identity of one rendered comparison artifact.
+pub(crate) fn artifact_identity_payload(
     test: &Bound<'_, PyAny>,
     model_chain_names: &Bound<'_, PyAny>,
     context: &IdentityContext<'_>,
@@ -300,7 +330,7 @@ pub(crate) fn artifact_identity(
         let identity: Option<Bound<'_, PyAny>> = context.model_identities.get_item(&name)?;
         chain.append((name, identity))?;
     }
-    payload_digest(
+    payload_text(
         dict(
             py,
             vec![
@@ -316,11 +346,11 @@ pub(crate) fn artifact_identity(
     )
 }
 
-/// The stable cache slot of one source test case.
-pub(crate) fn artifact_record_key(test: &Bound<'_, PyAny>) -> PyResult<String> {
+/// The text whose digest is the stable cache slot of one source test case.
+pub(crate) fn artifact_record_key_payload(test: &Bound<'_, PyAny>) -> PyResult<String> {
     let py: Python<'_> = test.py();
     let source_path: Bound<'_, PyAny> = test.getattr("source_path")?;
-    payload_digest(
+    payload_text(
         dict(
             py,
             vec![
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compile_outputs/artifacts.rs b/crates/sqlbuild-python/src/bindings/_helpers/compile_outputs/artifacts.rs
index 8dcc012f25..b392c3b15c 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/compile_outputs/artifacts.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/compile_outputs/artifacts.rs
@@ -15,6 +15,7 @@ use pyo3::types::PyBytes;
 use sqlbuild_cache::artifacts::errors::ArtifactError;
 use sqlbuild_cache::artifacts::main::publish_staged::publish_staged;
 use sqlbuild_cache::artifacts::main::remove_stale::remove_stale;
+use sqlbuild_cache::artifacts::main::remove_stale_unless_unchanged::remove_stale_unless_unchanged;
 use sqlbuild_cache::artifacts::main::staged_files::staged_files;
 use sqlbuild_cache::artifacts::main::write_artifacts::write_artifacts;
 use sqlbuild_cache::artifacts::models::{Publication, WrittenArtifacts};
@@ -187,11 +188,17 @@ pub(crate) fn remove_stale_artifacts(
     py: Python<'_>,
     compiled_dir: PathBuf,
     managed: Vec<PathBuf>,
+    stamp: Option<(PathBuf, i64)>,
 ) -> PyResult<usize> {
     compiler_guard(|| {
         let managed: HashSet<PathBuf> = managed.into_iter().collect();
-        py.detach(|| remove_stale(&compiled_dir, &managed))
-            .map_err(|error| artifact_error(py, error))
+        py.detach(|| match &stamp {
+            Some((stamp_path, racy_window_ns)) => {
+                remove_stale_unless_unchanged(&compiled_dir, &managed, stamp_path, *racy_window_ns)
+            }
+            None => remove_stale(&compiled_dir, &managed),
+        })
+        .map_err(|error| artifact_error(py, error))
     })
 }
 
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/render/macro_call_memo.rs b/crates/sqlbuild-python/src/bindings/_helpers/render/macro_call_memo.rs
index e07eeb085f..6e4cd932ca 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/render/macro_call_memo.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/render/macro_call_memo.rs
@@ -76,6 +76,16 @@ impl MacroCallMemo {
         Ok(())
     }
 
+    /// Whether this compile recorded calls the attached store does not hold yet.
+    fn store_needs_save(&self) -> PyResult<bool> {
+        Ok(self
+            .inner
+            .lock()
+            .map_err(poisoned)?
+            .store_mut()
+            .is_some_and(|store| store.needs_save()))
+    }
+
     /// Calls found in the store and calls recorded into it.
     fn store_stats(&self) -> PyResult<(usize, usize)> {
         Ok(self.inner.lock().map_err(poisoned)?.store_stats())
diff --git a/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs b/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
index fe8990db6d..5c06143522 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/rules/lint_pipeline.rs
@@ -31,6 +31,7 @@ use sqlbuild_rules::lint_text::models::{
     RetainedViolation, SuppressibleViolation, SuppressionFinding,
 };
 use sqlbuild_rules::project_tree::main::project_real_path::project_real_path;
+use sqlbuild_rules::project_tree::main::project_real_paths::project_real_paths;
 use sqlbuild_rules::sql_lint::main::preparation::prepare;
 use sqlbuild_rules::sql_lint::types::PreparedSql;
 
@@ -461,9 +462,18 @@ pub(crate) fn lint_sources(
     given: Option<Vec<LintFile>>,
 ) -> PyResult<Vec<LintFile>> {
     if let Some(given) = given {
+        let Some(paths) = selected else {
+            return Ok(given);
+        };
+        let files: Vec<&Path> = given.iter().map(|file| file.path.as_path()).collect();
+        let keep: Vec<bool> = project_real_paths(&files)
+            .iter()
+            .map(|path| paths.contains(path))
+            .collect();
         return Ok(given
             .into_iter()
-            .filter(|file| selected.is_none_or(|paths| paths.contains(&resolved(&file.path))))
+            .zip(keep)
+            .filter_map(|(file, kept)| kept.then_some(file))
             .collect());
     }
     let collected: Vec<(PathBuf, String)> =
diff --git a/crates/sqlbuild-rules/src/project_tree/main/mod.rs b/crates/sqlbuild-rules/src/project_tree/main/mod.rs
index 2e0808bc3b..2e651abd47 100644
--- a/crates/sqlbuild-rules/src/project_tree/main/mod.rs
+++ b/crates/sqlbuild-rules/src/project_tree/main/mod.rs
@@ -1,5 +1,6 @@
 pub mod project_path_escapes;
 pub mod project_real_path;
+pub mod project_real_paths;
 pub mod project_relative_parts;
 pub mod project_tree_children;
 pub mod project_tree_descendants;
diff --git a/crates/sqlbuild-rules/src/project_tree/main/project_real_paths.rs b/crates/sqlbuild-rules/src/project_tree/main/project_real_paths.rs
new file mode 100644
index 0000000000..9729c11932
--- /dev/null
+++ b/crates/sqlbuild-rules/src/project_tree/main/project_real_paths.rs
@@ -0,0 +1,59 @@
+use std::collections::HashMap;
+use std::path::{Component, Path, PathBuf};
+
+use crate::project_tree::main::project_real_path::project_real_path;
+
+/// `project_real_path` of every path, resolving each shared parent folder once.
+pub fn project_real_paths(paths: &[&Path]) -> Vec<PathBuf> {
+    let plain: Vec<Option<(&Path, &std::ffi::OsStr)>> =
+        paths.iter().map(|path| plain_file(path)).collect();
+    let mut folders: HashMap<&Path, Option<PathBuf>> = HashMap::new();
+    for (folder, _) in plain.iter().flatten() {
+        let _ = folders
+            .entry(folder)
+            .or_insert_with(|| resolved_folder(folder));
+    }
+    paths
+        .iter()
+        .zip(plain)
+        .map(|(path, plain)| match plain {
+            Some((folder, name)) => match &folders[folder] {
+                Some(resolved) => resolved.join(name),
+                None => project_real_path(path),
+            },
+            None => project_real_path(path),
+        })
+        .collect()
+}
+
+/// The folder and last name of `path` when that name is spelled last and is not a link.
+fn plain_file(path: &Path) -> Option<(&Path, &std::ffi::OsStr)> {
+    if !cfg!(target_os = "linux") {
+        return None;
+    }
+    let (Some(folder), Some(Component::Normal(name))) =
+        (path.parent(), path.components().next_back())
+    else {
+        return None;
+    };
+    let spelled_last: bool = path
+        .as_os_str()
+        .as_encoded_bytes()
+        .strip_suffix(name.as_encoded_bytes())
+        .is_some_and(|rest| rest.ends_with(b"/"));
+    if folder.as_os_str().is_empty() || !spelled_last {
+        return None;
+    }
+    match std::fs::symlink_metadata(path) {
+        Ok(metadata) if !metadata.file_type().is_symlink() => Some((folder, name)),
+        Ok(_link) => None,
+        Err(_absent) => None,
+    }
+}
+
+fn resolved_folder(folder: &Path) -> Option<PathBuf> {
+    let Ok(resolved) = std::fs::canonicalize(folder) else {
+        return None;
+    };
+    Some(resolved)
+}
diff --git a/crates/sqlbuild-rules/src/project_tree/tests.rs b/crates/sqlbuild-rules/src/project_tree/tests.rs
index 30903ec9e3..f0253b863e 100644
--- a/crates/sqlbuild-rules/src/project_tree/tests.rs
+++ b/crates/sqlbuild-rules/src/project_tree/tests.rs
@@ -10,6 +10,9 @@ mod path_queries;
 #[path = "tests/test_project_tree.rs"]
 mod project_tree;
 #[cfg(unix)]
+#[path = "tests/test_real_paths.rs"]
+mod real_paths;
+#[cfg(unix)]
 #[path = "tests/test_relative_parts.rs"]
 mod relative_parts;
 #[path = "tests/test_types.rs"]
diff --git a/crates/sqlbuild-rules/src/project_tree/tests/test_real_paths.rs b/crates/sqlbuild-rules/src/project_tree/tests/test_real_paths.rs
new file mode 100644
index 0000000000..204780617a
--- /dev/null
+++ b/crates/sqlbuild-rules/src/project_tree/tests/test_real_paths.rs
@@ -0,0 +1,89 @@
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+
+use crate::project_tree::main::project_real_paths::project_real_paths;
+use crate::project_tree::tests::helpers::linked_project;
+use crate::project_tree::tests::test_types::RealPathsTestCase;
+
+#[test]
+fn given_paths_resolved_together_when_resolving_then_each_resolves_as_alone() {
+    let test_cases = [
+        RealPathsTestCase {
+            description: "plain file",
+            path: "models/marts/orders.sql",
+            expected_resolved: "models/marts/orders.sql",
+        },
+        RealPathsTestCase {
+            description: "linked file",
+            path: "models/link.sql",
+            expected_resolved: "models/marts/orders.sql",
+        },
+        RealPathsTestCase {
+            description: "file below a linked folder",
+            path: "models/linkdir/orders.sql",
+            expected_resolved: "models/marts/orders.sql",
+        },
+        RealPathsTestCase {
+            description: "file linked out of its folder",
+            path: "models/readme.sql",
+            expected_resolved: "readme.md",
+        },
+        RealPathsTestCase {
+            description: "missing file",
+            path: "models/missing.sql",
+            expected_resolved: "models/missing.sql",
+        },
+        RealPathsTestCase {
+            description: "parent part before a link",
+            path: "models/marts/../link.sql",
+            expected_resolved: "models/marts/orders.sql",
+        },
+        RealPathsTestCase {
+            description: "trailing current part",
+            path: "models/marts/.",
+            expected_resolved: "models/marts",
+        },
+        RealPathsTestCase {
+            description: "trailing current part after a linked folder",
+            path: "models/linkdir/.",
+            expected_resolved: "models/marts",
+        },
+        RealPathsTestCase {
+            description: "linked folder with a trailing slash",
+            path: "models/linkdir/",
+            expected_resolved: "models/marts",
+        },
+        RealPathsTestCase {
+            description: "parent part after a linked folder",
+            path: "models/linkdir/../readme.sql",
+            expected_resolved: "readme.md",
+        },
+        RealPathsTestCase {
+            description: "file below a linked project root",
+            path: "sub/root_link/models/marts/orders.sql",
+            expected_resolved: "models/marts/orders.sql",
+        },
+    ];
+    let (_directory, root) = linked_project();
+    std::os::unix::fs::symlink(&root, root.join("sub/root_link")).expect("root link");
+    let paths: Vec<PathBuf> = test_cases
+        .iter()
+        .map(|test_case| root.join(test_case.path))
+        .collect();
+    let borrowed: Vec<&Path> = paths.iter().map(PathBuf::as_path).collect();
+
+    let resolved: HashMap<&str, PathBuf> = test_cases
+        .iter()
+        .map(|test_case| test_case.path)
+        .zip(project_real_paths(&borrowed))
+        .collect();
+
+    for test_case in test_cases {
+        assert_eq!(
+            resolved[test_case.path],
+            root.join(test_case.expected_resolved),
+            "{}",
+            test_case.description
+        );
+    }
+}
diff --git a/crates/sqlbuild-rules/src/project_tree/tests/test_types.rs b/crates/sqlbuild-rules/src/project_tree/tests/test_types.rs
index 7918aa2a67..300e29d7be 100644
--- a/crates/sqlbuild-rules/src/project_tree/tests/test_types.rs
+++ b/crates/sqlbuild-rules/src/project_tree/tests/test_types.rs
@@ -44,3 +44,11 @@ pub(crate) struct EscapesTestCase {
     pub(crate) path: &'static str,
     pub(crate) expected_escapes: bool,
 }
+
+/// One project path resolved in a batch and the project path it resolves to.
+#[cfg(unix)]
+pub(crate) struct RealPathsTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) path: &'static str,
+    pub(crate) expected_resolved: &'static str,
+}
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/build.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/build.md
index e925792cf9..9e7e4e3d7e 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/build.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/build.md
@@ -59,6 +59,7 @@ own option. `--version` works only directly after `sqb`, as in `sqb --version`:
 | `--defer-sources-to` | Read managed source data from another target |
 | `--fail-fast` | Stop on first failure and skip remaining nodes |
 | `--concurrency` | Number of worker connections (default: 1) |
+| `--test-timeout <duration>` | Time limit for each unit test the build runs, such as `30s`, or `disabled`; see [Time limits](../concepts/testing.md#time-limits) |
 | `--verbose`, `-v` | Show lifecycle SQL inline after each model |
 | `--start-cursor` | Override the start cursor, typed from the selected incremental models' cursor type; fails with `S303` if the selection mixes timestamp and integer cursors. With no cursor model selected, an integer value is read as an integer cursor and anything else as an ISO timestamp |
 | `--end-cursor` | Override the end cursor, typed the same way as `--start-cursor` |
@@ -155,6 +156,16 @@ When a model fails:
 - Staging/delta tables are retained for inspection
 - Failure details show the model name, failed phase, and error message
 
+A model that exceeds its [`max_duration`](../concepts/models/configuration.md#execution-time-limit)
+fails with `C417`. The failure states the elapsed time, the limit and where it was set, whether
+cancelling its running statement was confirmed, and the query IDs SQLBuild saw; downstream models are
+skipped as for any other failure. Its row shows `TIMEOUT`. In the summary line, timed-out models and unit tests are counted under
+`TIMEOUT=` and excluded from `FAIL=`
+(`PASS=1  WARN=0  FAIL=0  TIMEOUT=1  INSUFFICIENT=0  SKIP=1  TOTAL=3`). In `--json` output,
+`summary.failure_count` still includes them, and `summary.timeout_count` reports how many of those
+failures were timeouts. A timeout is a failure for the exit code: the build exits non-zero and its
+JSON `status` is `failed`.
+
 ## Fingerprints
 
 After a successful build, SQLBuild writes version identities to `_sqlbuild_fingerprints` in the target schema. These are used on subsequent runs to detect changes and skip unchanged work. See [Planning and Change Detection](../concepts/planning.md) for details.
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md
index 45ce79f9d1..7fec47b9f4 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md
@@ -120,8 +120,10 @@ and when a provider's settings use sources SQLBuild cannot list exactly, such as
 `settings_customise_sources` or command-line parsing. The reason is logged at debug level.
 
 SQLBuild keeps one stored result per project and target under
-`target/cache/compiler-native-v1/project-reuse-v2/`. It is replaced atomically; a damaged or unreadable entry is
-ignored and compile runs in full. Storing a result normally takes a fraction of a second after the
+`target/cache/compiler-native-v1/project-reuse-v3/`: a small `.store` file holding what the result
+depends on, and a `.output` file beside it holding the report. A check reads only the `.store` file;
+the report is read only when nothing changed. Each file is replaced atomically; a damaged, missing or
+unreadable file is ignored and compile runs in full. Storing a result normally takes a fraction of a second after the
 report is ready. When it has to hash a lot of data or the project has very many files, compile
 prints `Recording compile for reuse...` and a completion line to stderr before the report.
 
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/scenario.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/scenario.md
index 05c25fa021..644c73e468 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/scenario.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/scenario.md
@@ -33,6 +33,7 @@ sqb scenario test [flags]
 | `--local` | Run locally against DuckDB using captured JSONL snapshots |
 | `--strict` | Treat missing/stale local snapshots as errors instead of skips |
 | `--concurrency N` | Run up to `N` scenarios at once (default: the project `concurrency` setting) |
+| `--scenario-timeout <duration>` | Time limit for each scenario in this warehouse run, such as `10m`, or `disabled`; overrides `SCENARIO(timeout ...)`, the target and `[settings] scenario_timeout`. Not applied with `--local` |
 | `--sync-snapshots` | Capture missing/stale snapshots before local run (requires `--local`) |
 | `--refresh` | Recapture all selected snapshots before local run (requires `--local`) |
 | `--force` | Bypass snapshot capture safety limits |
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/test.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/test.md
index ff2664ca2c..5aeb59c0fe 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/test.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/test.md
@@ -26,6 +26,7 @@ Positional `SELECTOR` arguments work like `--select`, and both forms can be comb
 | `--select`, `-s` | Select tests by name (`order_status_rules` or `test:order_status_rules`), or the tests targeting specific models |
 | `--exclude` | Exclude tests by name, or the tests targeting specific models |
 | `--warehouse <name>` | Snowflake warehouse for this invocation; overrides the target's `query` [warehouse group](../concepts/project-configuration.md#command-group-warehouses) and the connection warehouse |
+| `--test-timeout <duration>` | Time limit for each test in this run, such as `30s` or `2m`, or `disabled`; overrides `TEST(timeout ...)`, the target and `[settings] test_timeout` |
 
 Test names are globally unique, so a bare test name selects exactly that test. `test:<name>` is the
 explicit form and accepts the same glob patterns as names, such as `test:orders_*`. A bare pattern
@@ -43,6 +44,13 @@ declared parameter types and nullability, typed values, and a content fingerprin
 are strings in JSON so their scale is preserved. Each test check in JSON output reports its
 `duration_ms`.
 
+Each test row shows how many real (unmocked) models the test executes, such as `3 real models`. A
+test that runs past its [time limit](../concepts/testing.md#time-limits) is reported as `TIMEOUT`,
+counted under `TIMEOUT=` and excluded from `FAIL=` in the summary line
+(`PASS=4  FAIL=0  TIMEOUT=1  TOTAL=5`), and the run continues with the remaining tests. In `--json`
+output, `summary.fail_count` still includes timed-out tests, and `summary.timeout_count` reports how
+many of them timed out. A timeout still fails the run: `sqb test` exits non-zero.
+
 ## Examples
 
 ```bash
@@ -57,4 +65,7 @@ sqb test order_status_rules
 
 # Run one case of a parameterized test
 sqb test test:order_status_rules --case open_case
+
+# Allow each test two minutes in this run
+sqb test --test-timeout 2m
 ```
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md
index b471f7fe13..ab4ac91e34 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md
@@ -101,6 +101,8 @@ Set `sql_analysis_dialect_name` to the SQL dialect name that matches your engine
 
 For full control with no inherited defaults, subclass `StrictAdapter` instead. Every method is abstract and must be implemented explicitly. SQLBuild raises a clear error listing any unimplemented methods.
 
+A few `StrictAdapter` methods are optional and have a documented default. `create_sql_test_output` and `drop_sql_test_output` let an adapter store a unit test's actual output once so every comparison reads it (see [Single-read comparison](testing.md#single-read-comparison)). By default `create_sql_test_output` returns `None`, and SQLBuild runs the test's inline comparison query instead, so adapters that do not implement them keep working unchanged.
+
 ### Discovery rules
 
 - SQLBuild discovers all `.py` files under `adapters/` recursively (excluding `__init__.py` and files starting with `_`)
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/models/configuration.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/models/configuration.md
index bde6b83c91..c5f6b291c4 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/models/configuration.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/models/configuration.md
@@ -16,6 +16,7 @@ Online: https://sqlbuild.com/docs/concepts/models/configuration/
 - Snapshot fields
 - Custom materialization fields
 - Diff fields
+- Execution time limit
 
 Every `MODEL()` key must be one of the fields on this page. Any other key fails compilation with
 `D002`, naming the key and its file and line and suggesting the nearest supported key. SQL comments
@@ -52,6 +53,7 @@ which accepts any keys.
 | `sql_analysis` | Per-model SQL-analysis override |
 | `dynamic_columns` | Named pivot column families with `pivot_column`, `value_column`, `aggregate`, `type`, and `name_pattern` |
 | `audit_factories` | Audit factories that generate audit cases for this model; see [Audits](../audits.md) |
+| `max_duration` | Time limit for one build attempt of this model, or `"disabled"`; see [Execution time limit](#execution-time-limit) |
 
 ## Descriptions
 
@@ -224,3 +226,42 @@ See [Snapshots](../snapshots.md) for strategy combinations, defaults, and safety
 | `row_diff_tolerances` | Numeric comparison tolerances |
 | `row_diff_sample_rows` | Deterministic unique-key sample size; `0` disables inherited sampling |
 | `row_diff_sample_seed` | Integer seed used for deterministic key hashing |
+
+## Execution time limit
+
+`max_duration` bounds how long one build attempt of a model may run. It has no default: a model
+without it, or with `max_duration "disabled"`, runs until it finishes. Set it in the usual layers,
+with the `MODEL()` header taking precedence over the nearest path default, and the path default over
+`[defaults]`:
+
+```toml
+[defaults]
+max_duration = "30m"
+
+[path_defaults."marts/order_history"]
+max_duration = "2h"
+```
+
+```sql
+MODEL (
+  description 'Daily customer order counts.',
+  materialized table,
+  max_duration "10m"
+);
+```
+
+The limit is a positive fixed duration such as `"90s"`, `"15m"` or `"1h30m"`. One deadline covers the
+whole attempt, starting when the model begins executing: its hooks, statements, attached audits and
+every microbatch. Time spent waiting for upstream models or for a free worker does not count.
+
+At the deadline SQLBuild cancels the model's running statements through the adapter (see the
+[adapter table](../project-configuration.md#duration-limit)), runs no further statements
+for the attempt, and reports the model as timed out with the elapsed time, the limit, where it was
+set, the cancellation outcome and the query IDs it saw. Statements that already completed are not
+undone, but the attempt stops before any later promotion or swap, so a staged table is not
+published. Models that depend on it are skipped as for any failure, and independent models continue.
+A model that finishes successfully just after its limit keeps its successful result.
+
+The per-model limit is separate from the target's whole-build
+[`max_duration`](../project-configuration.md#duration-limit). When both apply, each
+statement is bounded by whichever deadline comes first.
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/project-configuration.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/project-configuration.md
index d425a95333..6a01ea8476 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/project-configuration.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/project-configuration.md
@@ -368,9 +368,9 @@ Both policies default to `false`. `sqb clone --from prod --to dev` requires `all
 
 ### Execution limits
 
-Named targets can set optional limits for `sqb build`. Limits are useful for development targets
-where an unexpectedly broad selector or long-running command should stop instead of consuming an
-unbounded amount of warehouse compute.
+Named targets can set optional limits for `sqb build`, and per-item time limits for unit tests and
+scenarios. Limits are useful for development targets where an unexpectedly broad selector or
+long-running command should stop instead of consuming an unbounded amount of warehouse compute.
 
 ```toml
 [targets.dev]
@@ -381,6 +381,8 @@ schema = "dev_alice"
 [targets.dev.execution_limits]
 max_models = 125
 max_duration = "45m"
+test_timeout = "60s"
+scenario_timeout = "10m"
 remediation = """
 Review the expanded plan and confirm that the selected scope is intentional.
 Automated tools must request approval before changing this policy.
@@ -403,17 +405,67 @@ key, and confirms that no warehouse changes were made.
 
 `max_duration` is a positive fixed duration such as `30m`, `2h`, or `90s`. It covers planning and
 warehouse-mutating build phases after target resolution. SQLBuild stops dispatching work at the
-deadline and passes the remaining duration to each warehouse statement so in-flight work is
-cancelled by the adapter.
-
-Safe in-flight cancellation is currently supported by the Snowflake adapter, whose maximum accepted
-duration is seven days. SQLBuild rejects an unsupported duration or an adapter that cannot guarantee
-active statement cancellation rather than presenting a limit that may leave warehouse work running.
+deadline, bounds each warehouse statement by the remaining time where the warehouse supports a
+statement limit, and cancels statements still running at the deadline.
+
+Every built-in adapter enforces the limit:
+
+| Adapter | Warehouse statement limit | Cancellation at the deadline | Longest accepted duration |
+|---------|---------------------------|------------------------------|---------------------------|
+| Snowflake | `STATEMENT_TIMEOUT_IN_SECONDS` per statement | Driver cancel request for the running query | 7 days |
+| DuckDB, MotherDuck | none | Connection interrupt | about 24 days |
+| PostgreSQL | `SET LOCAL statement_timeout` inside transactions | Driver cancel (`cancel_safe`) | about 24 days |
+| BigQuery | Job timeout (`job_timeout_ms`) on each query job | `jobs.cancel` (asynchronous) | about 24 days |
+| Databricks | `STATEMENT_TIMEOUT` session setting, capped at 48 hours | Cursor cancel (asynchronous) | about 24 days |
+| SQL Server | none | Driver cancel | about 24 days |
+
+BigQuery and Databricks stop cancelled work asynchronously, so SQLBuild reports their cancellation
+as requested but unconfirmed. SQL Server has no per-statement limit; the driver's query timeout is
+process-wide, so SQLBuild relies on cancelling the running statement instead. A project adapter that
+cannot cancel a running statement is rejected for `max_duration` rather than presenting a limit that
+may leave warehouse work running. Under a per-item timeout, such an adapter's items stop before their
+next statement and report the cancellation as not supported.
+
+PostgreSQL never sets `statement_timeout` for the whole session, so a limit cannot leak to another
+client through a transaction-mode connection pooler; statements outside a transaction are bounded by
+cancellation alone.
 
 A duration failure may occur after earlier models have completed. Its error therefore distinguishes
 the timeout from a pre-execution model-limit failure and states that warehouse changes may already
 have occurred.
 
+Upgrade note: earlier releases accepted `max_duration` only on Snowflake and rejected it on other
+adapters. It is now accepted on every built-in adapter, so a target that sets it for a non-Snowflake
+warehouse is enforced instead of failing at startup.
+
+#### Per-item time limits
+
+Unit test, scenario and model limits (`test_timeout`, `scenario_timeout` and a model's
+`max_duration`) count running time only. The clock starts at the item's first warehouse statement,
+so rendering and setup before it do not count, and it stops during rollback and cleanup, which always
+run even after the limit. A timeout is reported only when the limit passed before the item's work
+finished. Where a warehouse reports queueing separately, queued time is excluded:
+
+| Adapter | How the item limit is enforced | Queued time |
+|---------|--------------------------------|-------------|
+| Snowflake | `STATEMENT_TIMEOUT_IN_SECONDS` per statement, with a client-side backstop after a grace margin of half the limit (at least 30 seconds) | Excluded by the warehouse limit; the backstop counts it |
+| DuckDB, MotherDuck | Connection interrupt at the limit | Not applicable |
+| PostgreSQL | `SET LOCAL statement_timeout` inside transactions; cancellation at the limit | Lock waits count |
+| BigQuery | `jobs.cancel` at the limit; no job timeout, because a job timeout counts from job creation | Excluded: the clock stops while a job is pending |
+| Databricks | `STATEMENT_TIMEOUT` session setting and cursor cancel | May count |
+| SQL Server | Driver cancel at the limit | Lock waits count |
+
+When a model contains unit tests or nested work with its own limit, the earliest limit applies and
+cancels whatever is running under it. After a timeout interrupted a statement, SQLBuild checks the
+connection before giving it to the next item and replaces it if it is no longer usable.
+
+#### Unit test and scenario timeouts
+
+`test_timeout` and `scenario_timeout` set the per-test and per-scenario time limit on this target,
+replacing `[settings] test_timeout` and `[settings] scenario_timeout`. They follow the
+[timeout precedence](#unit-test-and-scenario-limits) described under Settings and accept a positive
+fixed duration or `"disabled"`.
+
 #### Custom remediation
 
 `remediation` is optional project-authored text. SQLBuild appends it to model-count, duration, and
@@ -490,6 +542,7 @@ instead.
 - Microbatch: `microbatch_strategy`, `batch_size`, `batch_concurrency`, `max_microbatches`
 - Row diff: `row_diff_exclude_columns`, `row_diff_tolerances`, `row_diff_sample_rows`,
   `row_diff_sample_seed`
+- Execution: `max_duration` (see [Execution time limit](models/configuration.md#execution-time-limit))
 
 ## Constants
 
@@ -594,6 +647,9 @@ auto_load_sources = true
 table_promotion_mode = "staged"
 default_audit_severity = "error"
 default_audit_run_scope = "delta_and_final"
+unit_test_max_real_models = 10
+test_timeout = "30s"
+scenario_timeout = "15m"
 ```
 
 | Field | Default | Description |
@@ -608,6 +664,33 @@ default_audit_run_scope = "delta_and_final"
 | `table_promotion_mode` | adapter default | `staged` (CTAS to staging, audit, then promote) or `immediate` (CTAS directly to target, then audit) |
 | `default_audit_severity` | `error` | Default severity for audits: `warn` or `error` |
 | `default_audit_run_scope` | `delta_and_final` | Default run scope for audits: `final` or `delta_and_final` |
+| `unit_test_max_real_models` | `10` | Most real (unmocked) models one SQL unit test may execute; a larger test fails compilation. See [Unit test and scenario limits](#unit-test-and-scenario-limits). |
+| `test_timeout` | `30s` | Time limit for each SQL unit test, or `"disabled"` |
+| `scenario_timeout` | none | Time limit for each scenario, or `"disabled"` |
+
+### Unit test and scenario limits
+
+A SQL unit test runs the real SQL of every unmocked model between its mocks and the model under
+test as one warehouse statement. `unit_test_max_real_models` stops that statement from silently
+growing into a whole-chain integration query: a test that executes more real models fails
+compilation and names every real model, the dependency path that pulls it in, and the inputs to
+mock. See [Testing](testing.md#cost-guards) for an example.
+
+`test_timeout` bounds each test's warehouse work, measured from the test's first statement. The
+limit that applies to a test is, from highest precedence:
+
+1. `--test-timeout` on `sqb test` or `sqb build`
+2. `TEST(timeout "...")` in the test file
+3. `[targets.<target>.execution_limits] test_timeout`
+4. `[settings] test_timeout`, then the 30-second default
+
+`scenario_timeout` works the same way with `--scenario-timeout`, `SCENARIO(timeout "...")`,
+`[targets.<target>.execution_limits] scenario_timeout` and `[settings] scenario_timeout`; it has
+no default. At the limit, SQLBuild cancels the item's running statement, reports the item as
+`TIMEOUT` with the elapsed time, the limit, where the limit was set, the cancellation outcome and
+the query IDs it saw, and continues with independent items. Every value accepts a positive fixed
+duration such as `"45s"` or `"10m"`, or `"disabled"`. `sqlbuild_local.toml` accepts all three
+settings.
 
 ### Table promotion mode
 
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/scenarios.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/scenarios.md
index 25df731c5d..6ef0c847bf 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/scenarios.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/scenarios.md
@@ -30,6 +30,13 @@ Scenarios materialize physical relations in the warehouse and build the actual m
 
 Use unit tests for transformation logic. Use scenarios for coherent end-to-end validation.
 
+Cost is a second reason to move a check into a scenario. A unit test inlines every real model in its
+chain into one SQL statement, so the warehouse compiles and optimises the whole chain at once, and
+that cost can grow much faster than the query's size. A scenario materialises each model step by
+step, so each statement is compiled separately, as in a real build. SQLBuild limits how many real
+models one unit test may execute (see [Cost guards](testing.md#cost-guards)); when a test
+reaches that limit because it checks behaviour across several models, it may belong here.
+
 ## Authoring scenarios
 
 Scenario files live under `tests/scenarios/` and use one file per scenario:
@@ -128,12 +135,17 @@ available only to the model are not included. See
 
 ### SCENARIO() header
 
-The header is metadata only:
+The header holds metadata and an optional time limit:
 
 | Field | Description |
 |-------|-------------|
 | `description` | Required description of the behaviour the scenario proves ([Descriptions](models/configuration.md#descriptions)) |
 | `tags` | Optional list of string tags |
+| `timeout` | Optional time limit for the scenario's warehouse run, such as `"10m"`, or `"disabled"`; overrides `[settings] scenario_timeout` and the target's `scenario_timeout` |
+
+At the limit SQLBuild cancels the scenario's running statement, still drops the relations the
+scenario created unless `--retain` is set, and reports the scenario as failed with `C417`, the
+elapsed time, the limit and where it was set.
 
 ## How scenarios execute
 
diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/testing.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/testing.md
index 61b9d0922d..c78426471a 100644
--- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/testing.md
+++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/testing.md
@@ -24,6 +24,7 @@ Online: https://sqlbuild.com/docs/concepts/testing/
 - Multiple tests per file
 - Repeating test logic
 - Test placement
+- Cost guards
 - Skipping SQL analysis
 - Running tests
 
@@ -788,6 +789,102 @@ tests/
     ...
 ```
 
+## Cost guards
+
+A unit test runs the real SQL of every unmocked model between its mocks and the model under test as
+one warehouse statement. That is cheap for one or two models, but the warehouse compiles and
+optimises the whole chain at once, and that cost can grow much faster than the query's size.
+SQLBuild therefore bounds each unit test in two ways.
+
+### Real-model limit
+
+`[settings] unit_test_max_real_models` (default `10`, also accepted in `sqlbuild_local.toml`) caps
+how many real models one unit test may execute, counting the model under test. A test over the limit
+fails compilation before any warehouse connection opens:
+
+```text
+error[S000]: Invalid SQL test fixtures:
+- SQL unit test 'customer_totals_rollup' executes 3 real models (limit 2).
+  Test file:        tests/unit/customer_totals_rollup.sql
+  Model under test: customer_totals
+
+Real models executed, and the dependency path that pulls each one in:
+  customer_totals  model under test
+  order_totals     customer_totals -> order_totals
+  stg_orders       customer_totals -> order_totals -> stg_orders
+
+Options:
+  1. Mock the direct inputs of the model under test to cut the chain:
+       __ref__order_totals AS (SELECT ...)
+  2. If this test checks behaviour across several models, consider a scenario
+     (tests/scenarios/). A unit test inlines every real model into one statement,
+     so the warehouse compiles and optimises the whole chain at once; a scenario
+     builds each model step by step, compiling each statement separately.
+  3. Raise the limit in sqlbuild_project.toml (or sqlbuild_local.toml):
+       [settings]
+       unit_test_max_real_models = 3
+```
+
+Upgrade note: the limit is new, so after upgrading, a selection containing a unit test that executes
+more than 10 real models fails to compile (`sqb test`, and `sqb build` when it runs tests) until the
+test is mocked or the limit is raised.
+
+`sqb test` shows each test's real-model count on its row, such as `3 real models`, so a chain that
+is growing towards the limit is visible early. Direct macro, UDF and table function tests execute
+no models and are not counted.
+
+### Time limits
+
+Each unit test has a time limit, `30s` by default, measured from its first warehouse statement.
+Rendering, warehouse queueing where the warehouse reports it, and cleanup after the comparison do not
+count; see [per-item time limits](project-configuration.md#per-item-time-limits) for
+each adapter.
+Set it, from highest precedence, with `--test-timeout` on `sqb test` or `sqb build`, with
+`TEST(timeout "45s")` in the test file, with `[targets.<target>.execution_limits] test_timeout`, or
+with `[settings] test_timeout`. Each accepts a positive fixed duration or `"disabled"`.
+
+At the limit SQLBuild cancels the test's running statement, reports the test as `TIMEOUT` with the
+elapsed time, the limit, where it was set and the cancellation outcome, and continues with the
+remaining tests. When the test inlines real models beyond the model under test, the message also
+lists them and the same options as the real-model limit:
+
+```text
+SQL unit test 'order_scans_kept' timed out after 1.0s (limit 1s).
+  Limit set by:  TEST() timeout in tests/unit/order_scans_kept.sql
+  Cancellation:  confirmed; the running statement stopped
+
+This test inlines 1 real model beyond the model under test into one statement:
+  stg_orders
+
+Options:
+  1. Mock the inlined models closest to the model under test to cut the chain,
+     for example __ref__stg_orders AS (SELECT ...).
+  2. If this test checks behaviour across several models, consider a scenario
+     (tests/scenarios/). A unit test inlines every real model into one statement,
+     so the warehouse compiles and optimises the whole chain at once; a scenario
+     builds each model step by step, compiling each statement separately.
+  3. To allow more time, set TEST(timeout "3s") on this test, [settings] test_timeout = "3s"
+     in sqlbuild_project.toml (or sqlbuild_local.toml), or --test-timeout for one run.
+```
+
+A test that runs only the model under test reports the timeout, the limit, its source and how to
+change it, without the scenario advice. See
+[Duration limit](project-configuration.md#duration-limit) for how each adapter cancels
+a running statement.
+
+### Single-read comparison
+
+A test compares its actual output with the expected rows several ways: both row counts, rows
+missing from each side, the expected-column check and, on failure, row samples. Where the warehouse
+supports it, SQLBuild evaluates each compared model's actual output once into a temporary relation
+and runs every comparison against it, so the chain is not recomputed for each read. Results,
+messages and samples are unchanged. DuckDB and MotherDuck use a session temporary table that
+SQLBuild drops after the test, including after a failure or timeout. PostgreSQL runs the test in
+one transaction with an `ON COMMIT DROP` temporary table that the transaction's end removes, so the
+reads stay on one server session behind a transaction-mode connection pooler. Snowflake reads the
+first query's persisted result through `RESULT_SCAN`. If a stored output can no longer be read,
+the test falls back to the inline comparison. Other adapters compare the inline query as before. `target/run/tests/` keeps the self-contained inline comparison query for each test.
+
 ## Skipping SQL analysis
 
 Compile checks SQL test fixtures and assertions like model SQL. For SQL the parser cannot read,
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index 26335e717b..831b01efc3 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -5282,6 +5282,7 @@ class MacroCallMemo:
     def stats(self) -> tuple[int, int, int]: ...
     def attach_store(self, path: str, environment: str) -> tuple[bytes, int]: ...
     def discard_store(self) -> None: ...
+    def store_needs_save(self) -> bool: ...
     def store_stats(self) -> tuple[int, int]: ...
     def save_store(self, path: str, metadata: bytes) -> int | None: ...
 
diff --git a/src/sqlbuild/compiler/compile/_helpers/macro_bridge/call_store.py b/src/sqlbuild/compiler/compile/_helpers/macro_bridge/call_store.py
index 297577b5ab..0904ebc79d 100644
--- a/src/sqlbuild/compiler/compile/_helpers/macro_bridge/call_store.py
+++ b/src/sqlbuild/compiler/compile/_helpers/macro_bridge/call_store.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 from sqlbuild._native import (
@@ -48,10 +49,17 @@ def target_context_with_macro_call_store(
 
 
 def _model_paths(*, discovered_inputs: DiscoveredProjectInputs, project_dir: Path) -> list[str]:
+    root: str = str(project_dir)
+    prefix: str = root if root.endswith(os.sep) else root + os.sep
     paths: list[str] = []
     for model_file in discovered_inputs.model_files:
+        file_path: Path = model_file.file_path
+        text: str = str(file_path)
+        if text.startswith(prefix):
+            paths.append(text[len(prefix) :].replace(os.sep, "/"))
+            continue
         try:
-            paths.append(model_file.file_path.relative_to(project_dir).as_posix())
+            paths.append(file_path.relative_to(project_dir).as_posix())
         except ValueError:
             continue
     return paths
diff --git a/src/sqlbuild/compiler/macro_bridge/classes/macro_bridge.py b/src/sqlbuild/compiler/macro_bridge/classes/macro_bridge.py
index 76e2821122..262903156b 100644
--- a/src/sqlbuild/compiler/macro_bridge/classes/macro_bridge.py
+++ b/src/sqlbuild/compiler/macro_bridge/classes/macro_bridge.py
@@ -87,6 +87,8 @@ def save_store(self) -> None:
         if not observed or digest_module_files(self._module_digests) != self._module_digests:
             logging.getLogger(__name__).debug("Macro call store not saved: code or files changed")
             return
+        if not self.memo.store_needs_save():
+            return
         metadata: bytes = module_digests_metadata(self._module_digests)
         try:
             if project_fingerprint(project_dir=project_dir, model_paths=model_paths) != fingerprint:
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
index 7ba3c3653d..dbbb6d7464 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
@@ -619,6 +619,30 @@ class CompileReuseBypassTestCase:
     expected_stored_entries: int = 0
 
 
+def keep_compiled_root(_project_dir: Path) -> None:
+    """Leave the compiled folder as compile creates it."""
+
+
+@dataclass(frozen=True)
+class CompiledTreeStampTestCase:
+    """An unmanaged file added under a compiled tree the previous compile stamped."""
+
+    description: str
+    edit: Callable[[Path], None]
+    expected_removed_path: str
+    prepare: Callable[[Path], None] = keep_compiled_root
+    expected_stamped: bool = True
+
+
+@dataclass(frozen=True)
+class CompileReuseRacyRefreshTestCase:
+    """A stored compile whose inputs were racy when stored, refreshed by the next hit."""
+
+    description: str
+    expected_reused: tuple[bool, bool, bool]
+    expected_output_rewritten: bool
+
+
 @dataclass(frozen=True)
 class CompileReuseCorruptEntryTestCase:
     """One way a stored compile can be damaged on disk."""
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
index 284f856294..4d584bdfcf 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
@@ -68,8 +68,9 @@
 COMPILER_CACHE_DIRECTORY_NAME: str = "compiler"
 RETIRED_CACHE_NAMESPACE_SUFFIXES: tuple[str, ...] = ("", "-native-v1", "-native-preview-v1")
 
-NATIVE_REUSE_DIRECTORY_NAME: str = "project-reuse-v2"
+NATIVE_REUSE_DIRECTORY_NAME: str = "project-reuse-v3"
 NATIVE_REUSE_SUFFIX: str = ".store"
+NATIVE_REUSE_OUTPUT_SUFFIX: str = ".output"
 RETIRED_REUSE_DIRECTORY_NAME: str = "project-reuse-v1"
 REUSE_LOGGER_NAME: str = "sqlbuild.compile.reuse"
 _DBT_SHAPED_SQL_SIZE_PROFILE: tuple[tuple[float, int], ...] = (
@@ -3254,6 +3255,35 @@ def add_stale_compiled_file(project_dir: Path) -> None:
     write_project_file(project_dir, "target/compiled/models/stale_model.sql", "SELECT 1\n")
 
 
+COMPILED_TREE_STAMP_FILE_NAME: str = "compiled-tree.bin"
+
+
+def compiled_tree_stamp_path(*, project_dir: Path) -> Path:
+    """Return the compile cache file stamping the compiled tree the last stale-file walk left."""
+
+    return compiler_cache_directory(project_dir) / COMPILED_TREE_STAMP_FILE_NAME
+
+
+def add_stale_compiled_file_beside_models(project_dir: Path) -> None:
+    """Add an unmanaged compiled file to a folder that already holds compiled models."""
+
+    write_project_file(project_dir, "target/compiled/models/marts/stale_mart.sql", "SELECT 1\n")
+
+
+def link_compiled_root(project_dir: Path) -> None:
+    """Make `target/compiled` a link to another folder under `target/`."""
+
+    linked: Path = project_dir / "target/compiled-linked"
+    linked.mkdir(parents=True, exist_ok=True)
+    (project_dir / "target/compiled").symlink_to(linked, target_is_directory=True)
+
+
+def add_stale_file_in_compiled_root(project_dir: Path) -> None:
+    """Add an unmanaged file directly in the compiled folder."""
+
+    write_project_file(project_dir, "target/compiled/stray.sql", "SELECT 1\n")
+
+
 def truncate_file(path: Path) -> None:
     """Drop the last bytes of a stored entry."""
 
@@ -3273,6 +3303,42 @@ def flip_header_byte(path: Path) -> None:
     path.write_bytes(contents[:40] + b"!" + contents[41:])
 
 
+def stored_output_path(entry: Path) -> Path:
+    """Return the output file stored beside one reuse slot file."""
+
+    return entry.with_suffix(NATIVE_REUSE_OUTPUT_SUFFIX)
+
+
+def truncate_stored_output(entry: Path) -> None:
+    """Drop the last bytes of the slot's stored output."""
+
+    truncate_file(stored_output_path(entry))
+
+
+def flip_stored_stdout_bytes(entry: Path) -> None:
+    """Corrupt the stored report inside the slot's output without changing its length."""
+
+    flip_trailing_bytes(stored_output_path(entry))
+
+
+def remove_stored_output(entry: Path) -> None:
+    """Delete the slot's stored output, leaving its inputs."""
+
+    stored_output_path(entry).unlink()
+
+
+def garbage_stored_output(entry: Path) -> None:
+    """Replace the slot's stored output with unrelated bytes."""
+
+    garbage_file(stored_output_path(entry))
+
+
+def previous_format_slot(entry: Path) -> None:
+    """Leave a slot as the previous single-file store format wrote it."""
+
+    entry.write_bytes(b"SQBSTORE" + (1).to_bytes(4, "little") + bytes(32) + b"project-reuse")
+
+
 def empty_file(path: Path) -> None:
     """Leave an empty stored entry, as an interrupted copy would."""
 
@@ -3385,6 +3451,13 @@ def __exit__(self, *exception: object) -> None:
     monkeypatch.setattr(native_module, "PreparedCompileArtifacts", WrittenThenChanged)
 
 
+def edit_project_model_during_compile(project_dir: Path) -> None:
+    """Change one model's bytes after the compile read it and before it stores."""
+
+    path: Path = project_dir / "models/staging/stg_orders.sql"
+    path.write_bytes(path.read_bytes() + b"\n")
+
+
 def rewrite_compiled_model_unchanged(project_dir: Path) -> None:
     """Rewrite one compiled artifact with the bytes it already holds."""
 
@@ -4203,13 +4276,24 @@ def change(root: Path, _monkeypatch: pytest.MonkeyPatch) -> None:
         rewritten: bytes = b"".join(
             (*map(_framed, fields), generation, _framed(metadata), count.to_bytes(8, "little"))
         ) + b"".join(entries)
-        checksum: bytes = hashlib.sha256(_framed(rewritten)).digest()
+        _ = path.write_bytes(rewritten)
+        checksum: bytes = bytes.fromhex(_native_file_digest(path))
         _ = path.write_bytes(header + checksum + rewritten)
 
     return change
 
 
 _STORE_PREFIX_LENGTH: int = 12
+
+
+def _native_file_digest(path: Path) -> str:
+    """The BLAKE3 digest native stores checksum their bodies with, of the file at `path`."""
+
+    digest: str | None = native_module.digest_files([str(path)])[0]
+    assert digest is not None
+    return digest
+
+
 _STORE_CHECKSUM_LENGTH: int = 32
 
 
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_reuse.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_reuse.py
index e4ecab5325..a02b0d13a9 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_reuse.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_reuse.py
@@ -4,12 +4,15 @@
 
 import json
 import subprocess
+import time
 from pathlib import Path
 from typing import cast
 
 import pytest
 
+from sqlbuild._native import COMPILE_REUSE_DISABLE_ENV_VAR
 from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
+    CompiledTreeStampTestCase,
     CompileReuseBypassTestCase,
     CompileReuseConcurrentWriterTestCase,
     CompileReuseCorruptEntryTestCase,
@@ -18,6 +21,7 @@
     CompileReuseInvalidationTestCase,
     CompileReuseLargeFileTestCase,
     CompileReuseProviderSettingsTestCase,
+    CompileReuseRacyRefreshTestCase,
     CompileReuseRedirectTestCase,
     CompileReuseReplayTestCase,
     CompileReuseStoreFailureTestCase,
@@ -31,6 +35,8 @@
     ORDERS_API_TOKEN_ENV_VAR,
     CompileReuseRun,
     add_stale_compiled_file,
+    add_stale_compiled_file_beside_models,
+    add_stale_file_in_compiled_root,
     compile_in_process,
     compile_in_process_output,
     compile_in_process_reused,
@@ -38,18 +44,25 @@
     compile_reuse_entry_paths,
     compile_reuse_hit_count,
     compiled_artifacts,
+    compiled_tree_stamp_path,
     delete_compiled_model,
     edit_compiled_model,
+    edit_project_model_during_compile,
     empty_file,
     enable_compile_reuse,
     fail_reuse_store,
     flip_header_byte,
+    flip_stored_stdout_bytes,
     flip_trailing_bytes,
     garbage_file,
+    garbage_stored_output,
     in_process_compile_reads,
+    link_compiled_root,
+    previous_format_slot,
     progress_only,
     record_digested_files,
     recorded_events,
+    remove_stored_output,
     replace_project_text,
     rewrite_compiled_model_unchanged,
     rewrite_macro_without_change,
@@ -57,9 +70,11 @@
     run_reuse_compile_into_file,
     run_reuse_text_compile,
     settle_racy_window,
+    stored_output_path,
     touch_back,
     touch_model_without_change,
     truncate_file,
+    truncate_stored_output,
     write_before_reuse_store,
     write_custom_source_provider,
     write_large_file,
@@ -70,6 +85,8 @@
     write_retired_compiler_cache_files,
 )
 
+# Longer than the racy window, so folders the first compile created are trusted when stamped.
+_SETTLED_TREE_SECONDS: float = 2.2
 _CONTRACT_ERROR_MODEL: str = (
     "MODEL (description 'Order contract.', columns (missing_col (type INTEGER)));\n\n"
     'SELECT order_id FROM __ref("stg_orders")\n'
@@ -427,6 +444,21 @@ def test_given_reuse_bypass_when_compiling_twice_then_both_compiles_run_in_full(
         CompileReuseCorruptEntryTestCase(description="header_bytes", corrupt=flip_header_byte),
         CompileReuseCorruptEntryTestCase(description="empty", corrupt=empty_file),
         CompileReuseCorruptEntryTestCase(description="garbage", corrupt=garbage_file),
+        CompileReuseCorruptEntryTestCase(
+            description="previous_store_format", corrupt=previous_format_slot
+        ),
+        CompileReuseCorruptEntryTestCase(
+            description="output_truncated", corrupt=truncate_stored_output
+        ),
+        CompileReuseCorruptEntryTestCase(
+            description="output_report_bytes", corrupt=flip_stored_stdout_bytes
+        ),
+        CompileReuseCorruptEntryTestCase(
+            description="output_missing", corrupt=remove_stored_output
+        ),
+        CompileReuseCorruptEntryTestCase(
+            description="output_garbage", corrupt=garbage_stored_output
+        ),
     ],
     ids=lambda case: case.description,
 )
@@ -456,6 +488,43 @@ def test_given_corrupt_stored_compile_when_compiling_then_it_falls_back_and_rest
     )
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CompileReuseRacyRefreshTestCase(
+            description="project_written_just_before_the_first_compile",
+            expected_reused=(False, True, True),
+            expected_output_rewritten=False,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_racy_stored_inputs_when_reusing_then_only_the_slot_file_is_refreshed(
+    compile_reuse_project: Path, test_case: CompileReuseRacyRefreshTestCase
+) -> None:
+    project_dir: Path = compile_reuse_project
+    cold: CompileReuseRun = run_reuse_compile(project_dir=project_dir)
+    (entry,) = compile_reuse_entry_paths(project_dir=project_dir)
+    output: Path = stored_output_path(entry)
+    stored_output: tuple[bytes, int] = (output.read_bytes(), output.stat().st_mtime_ns)
+    slot_mtime_ns: int = entry.stat().st_mtime_ns
+
+    refreshed: CompileReuseRun = run_reuse_compile(project_dir=project_dir)
+    rerun: CompileReuseRun = run_reuse_compile(project_dir=project_dir)
+
+    assert (cold.reused, refreshed.reused, rerun.reused) == test_case.expected_reused
+    assert entry.stat().st_mtime_ns != slot_mtime_ns, "the hit refreshed the racy stamps"
+    assert (
+        (output.read_bytes(), output.stat().st_mtime_ns) != stored_output
+    ) is test_case.expected_output_rewritten
+    for run in (refreshed, rerun):
+        assert (run.returncode, run.report, run.compiled) == (
+            cold.returncode,
+            cold.report,
+            cold.compiled,
+        )
+
+
 @pytest.mark.parametrize(
     "test_case",
     [
@@ -664,6 +733,12 @@ def test_given_provider_settings_when_their_inputs_change_then_reuse_misses_like
             expected_stored_entries=0,
             expected_rerun_reused=False,
         ),
+        CompileReuseConcurrentWriterTestCase(
+            description="project_model_edited_during_compile",
+            other_write=edit_project_model_during_compile,
+            expected_stored_entries=0,
+            expected_rerun_reused=False,
+        ),
         CompileReuseConcurrentWriterTestCase(
             description="other_compile_rewrites_same_bytes",
             other_write=rewrite_compiled_model_unchanged,
@@ -834,3 +909,50 @@ def test_given_cache_files_of_older_releases_when_compiling_then_cached_compile_
 
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CompiledTreeStampTestCase(
+            description="stale_file_beside_compiled_models",
+            edit=add_stale_compiled_file_beside_models,
+            expected_removed_path="target/compiled/models/marts/stale_mart.sql",
+        ),
+        CompiledTreeStampTestCase(
+            description="stale_file_in_compiled_models_root",
+            edit=add_stale_compiled_file,
+            expected_removed_path="target/compiled/models/stale_model.sql",
+        ),
+        CompiledTreeStampTestCase(
+            description="stale_file_in_linked_compiled_root",
+            edit=add_stale_file_in_compiled_root,
+            expected_removed_path="target/compiled/stray.sql",
+            prepare=link_compiled_root,
+            expected_stamped=False,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_stamped_compiled_tree_when_a_stale_file_appears_then_the_next_compile_removes_it(
+    compile_reuse_project: Path, test_case: CompiledTreeStampTestCase
+) -> None:
+    project_dir: Path = compile_reuse_project
+    uncached: dict[str, str] = {COMPILE_REUSE_DISABLE_ENV_VAR: "1"}
+    test_case.prepare(project_dir)
+    cold: CompileReuseRun = run_reuse_compile(project_dir=project_dir, env=uncached)
+    time.sleep(_SETTLED_TREE_SECONDS)
+    stamped: CompileReuseRun = run_reuse_compile(project_dir=project_dir, env=uncached)
+    assert compiled_tree_stamp_path(project_dir=project_dir).is_file() is test_case.expected_stamped
+    unchanged: CompileReuseRun = run_reuse_compile(project_dir=project_dir, env=uncached)
+    test_case.edit(project_dir)
+
+    cleaned: CompileReuseRun = run_reuse_compile(project_dir=project_dir, env=uncached)
+
+    assert not (project_dir / test_case.expected_removed_path).exists()
+    for run in (stamped, unchanged, cleaned):
+        assert (run.returncode, run.report, run.compiled) == (
+            cold.returncode,
+            cold.report,
+            cold.compiled,
+        )
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/compile_outputs/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/compile_outputs/helpers.py
index f8e6902028..87b09cc628 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/compile_outputs/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/compile_outputs/helpers.py
@@ -21,7 +21,7 @@
 from sqlbuild.cli.compile_reuse._helpers import provider_settings as provider_settings_module
 
 STAGING_PREFIX: str = ".sqlbuild-staging-"
-NATIVE_REUSE_DIRECTORY_NAME: str = "project-reuse-v2"
+NATIVE_REUSE_DIRECTORY_NAME: str = "project-reuse-v3"
 NATIVE_REUSE_SUFFIX: str = ".store"
 
 
diff --git a/website/src/content/docs/docs/cli/compile.mdx b/website/src/content/docs/docs/cli/compile.mdx
index a08b7cbc86..978f0493ee 100644
--- a/website/src/content/docs/docs/cli/compile.mdx
+++ b/website/src/content/docs/docs/cli/compile.mdx
@@ -111,8 +111,10 @@ and when a provider's settings use sources SQLBuild cannot list exactly, such as
 `settings_customise_sources` or command-line parsing. The reason is logged at debug level.
 
 SQLBuild keeps one stored result per project and target under
-`target/cache/compiler-native-v1/project-reuse-v2/`. It is replaced atomically; a damaged or unreadable entry is
-ignored and compile runs in full. Storing a result normally takes a fraction of a second after the
+`target/cache/compiler-native-v1/project-reuse-v3/`: a small `.store` file holding what the result
+depends on, and a `.output` file beside it holding the report. A check reads only the `.store` file;
+the report is read only when nothing changed. Each file is replaced atomically; a damaged, missing or
+unreadable file is ignored and compile runs in full. Storing a result normally takes a fraction of a second after the
 report is ready. When it has to hash a lot of data or the project has very many files, compile
 prints `Recording compile for reuse...` and a completion line to stderr before the report.