diff --git a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
index a1a8f3a8db..8a894db89b 100644
--- a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
+++ b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
@@ -22,7 +22,9 @@ use fensu_policy::lifecycle::models::{
     RuntimeIdentity, ScopedIgnore,
 };
 use fensu_policy::{apply_suppressions, evaluate_batch, run_custom_host};
-use rayon::iter::{IntoParallelIterator, ParallelIterator};
+use rayon::iter::{
+    IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
+};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path, PathBuf};
@@ -98,20 +100,20 @@ fn evaluate_native_models(
     let mut result = NativeModelEvaluation::default();
     let mut model_results: Vec<Option<Vec<Fault>>> = vec![None; model_limit];
     let mut pending_models: Vec<PendingModelRule<'_>> = Vec::new();
+    let mut identities = if cache.is_some() {
+        model_cache_identities(
+            &models[..model_limit],
+            ruleset_fingerprint,
+            project_fingerprint,
+        )?
+        .into_iter()
+        .map(Some)
+        .collect()
+    } else {
+        vec![None; model_limit]
+    };
     for (index, model) in models.iter().take(model_limit).enumerate() {
-        let identity = cache
-            .map(|_| {
-                model_cache_identity(
-                    model,
-                    ruleset_fingerprint,
-                    if index == 0 {
-                        project_fingerprint
-                    } else {
-                        None
-                    },
-                )
-            })
-            .transpose()?;
+        let identity = identities[index].take();
         if let Some(identity) = &identity
             && let Some(entry) = bucket
                 .entries
@@ -546,20 +548,28 @@ fn custom_local_identity(model_identity: &str, rule: &str, project: Option<&str>
 fn custom_model_fact_identities(
     models: &[crate::models::Model],
 ) -> Result<BTreeMap<String, String>, String> {
-    models
-        .iter()
-        .map(|model| {
-            let mut digest = Sha256::new();
-            digest.update(b"custom-rule-model-facts-v1");
-            digest.update(ANALYSIS_BATCH_SCHEMA_VERSION.to_le_bytes());
-            digest.update(env!("CARGO_PKG_VERSION").as_bytes());
-            digest.update(serde_json::to_vec(model).map_err(|error| error.to_string())?);
-            Ok((
-                model.relative_path.clone(),
-                format!("{:x}", digest.finalize()),
-            ))
-        })
-        .collect()
+    let identity = |model: &crate::models::Model| -> Result<(String, String), String> {
+        let mut digest = Sha256::new();
+        digest.update(b"custom-rule-model-facts-v1");
+        digest.update(ANALYSIS_BATCH_SCHEMA_VERSION.to_le_bytes());
+        digest.update(env!("CARGO_PKG_VERSION").as_bytes());
+        digest.update(serde_json::to_vec(model).map_err(|error| error.to_string())?);
+        Ok((
+            model.relative_path.clone(),
+            format!("{:x}", digest.finalize()),
+        ))
+    };
+    let results: Vec<Result<(String, String), String>> = if models.len() <= 1 {
+        models.iter().map(identity).collect()
+    } else {
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(NATIVE_RULE_WORKERS.min(models.len()))
+            .stack_size(NATIVE_RULE_STACK_BYTES)
+            .build()
+            .map_err(|error| error.to_string())?;
+        pool.install(|| models.par_iter().map(identity).collect())
+    };
+    results.into_iter().collect()
 }
 
 fn custom_rule_identity(rule: &RuleMetadata, request: &EvaluateRequest) -> Result<String, String> {
@@ -581,6 +591,27 @@ fn custom_rule_identity(rule: &RuleMetadata, request: &EvaluateRequest) -> Resul
     Ok(format!("{:x}", digest.finalize()))
 }
 
+fn model_cache_identities(
+    models: &[&crate::models::Model],
+    ruleset: &str,
+    project: Option<&str>,
+) -> Result<Vec<String>, String> {
+    let identity = |(index, model): (usize, &&crate::models::Model)| {
+        model_cache_identity(model, ruleset, if index == 0 { project } else { None })
+    };
+    let results: Vec<Result<String, String>> = if models.len() <= 1 {
+        models.iter().enumerate().map(identity).collect()
+    } else {
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(NATIVE_RULE_WORKERS.min(models.len()))
+            .stack_size(NATIVE_RULE_STACK_BYTES)
+            .build()
+            .map_err(|error| error.to_string())?;
+        pool.install(|| models.par_iter().enumerate().map(identity).collect())
+    };
+    results.into_iter().collect()
+}
+
 fn model_cache_identity(
     model: &crate::models::Model,
     ruleset: &str,
diff --git a/fensu.toml b/fensu.toml
index b8872f79dc..55dc1cd71d 100644
--- a/fensu.toml
+++ b/fensu.toml
@@ -159,6 +159,12 @@ path = "src/sqlbuild/runtime/event_exporting/main/command_output_sink.py"
 symbols = ["command_output_sink"]
 reason = "Bare command-output sink decorators are evaluated as decorator(function), requiring positional function calling."
 
+[[targets.python.rule_exceptions]]
+rule = "FFS120"
+path = "src/sqlbuild/compiler/fact_cache/_helpers/restricted_pickle.py"
+symbols = ["_RestrictedUnpickler.find_class"]
+reason = "This override preserves pickle.Unpickler's positional find_class(module, name) protocol."
+
 [[targets.python.rule_exceptions]]
 rule = "FFR402"
 path = "src/sqlbuild/integrations/dagster/__init__.py"
diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
index e497a1f14a..cede009293 100644
--- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
+++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py
@@ -42,7 +42,7 @@
 from sqlbuild.compiler.contracts.main.validate import evaluate_model_contracts
 from sqlbuild.compiler.contracts.models import ContractValidationResult
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
-from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveryCacheRequest
 from sqlbuild.compiler.lineage.models import ProjectColumnLineage
 from sqlbuild.compiler.pipeline.main.selected_graph import (
     build_project_graph_with_analysis_selection,
@@ -116,6 +116,9 @@ def _analyze_compile_project(
         sql_analysis_enabled_override=(
             False if profile_flags.skip_discovery_sql_analysis else None
         ),
+        cache_request=DiscoveryCacheRequest(
+            selected_target=selected_target, no_cache=analysis_selection.no_cache
+        ),
     )
     discover_ms: int = elapsed_ms(discover_start)
     _ = complete_compile_phase(
diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/declaration_scope.py b/src/sqlbuild/compiler/compile/_helpers/attachment/declaration_scope.py
index 8f671bb97f..ff3c048668 100644
--- a/src/sqlbuild/compiler/compile/_helpers/attachment/declaration_scope.py
+++ b/src/sqlbuild/compiler/compile/_helpers/attachment/declaration_scope.py
@@ -5,6 +5,7 @@
 import inspect
 from collections.abc import Mapping
 from dataclasses import replace
+from pathlib import Path
 from types import CodeType
 
 from sqlbuild.compiler.compile._helpers.attachment.scope_relationships import (
@@ -28,7 +29,10 @@
 
 
 def build_declaration_scope(
-    *, discovered_inputs: DiscoveredProjectInputs, loaded_macros: dict[str, LoadedMacro]
+    *,
+    discovered_inputs: DiscoveredProjectInputs,
+    loaded_macros: dict[str, LoadedMacro],
+    compile_cache_dir: Path | None = None,
 ) -> DeclarationScopeBuild:
     """Build one canonical index and validate it before SQL expansion."""
 
@@ -46,7 +50,11 @@ def build_declaration_scope(
         for declaration in index.declarations
     )
     relationships: ScopeRelationshipBuild = (
-        build_scope_relationship_grants(discovered_inputs=discovered_inputs, index=index)
+        build_scope_relationship_grants(
+            discovered_inputs=discovered_inputs,
+            index=index,
+            compile_cache_dir=compile_cache_dir,
+        )
         if has_scoped_relationship_declarations
         and (discovered_inputs.test_files or discovered_inputs.scenario_files)
         else ScopeRelationshipBuild()
diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/scope_relationships.py b/src/sqlbuild/compiler/compile/_helpers/attachment/scope_relationships.py
index 369e300c65..32818872f0 100644
--- a/src/sqlbuild/compiler/compile/_helpers/attachment/scope_relationships.py
+++ b/src/sqlbuild/compiler/compile/_helpers/attachment/scope_relationships.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import cast
 
 from sqlbuild.compiler.compile._helpers.render.macros import (
     find_macro_call_names,
@@ -15,14 +16,19 @@
     extract_sql_test_expected_model_names,
     extract_unclassified_sql_test_ctes,
 )
-from sqlbuild.compiler.compile.constants import MACRO_ACTUAL_TEST_CTE_NAME
+from sqlbuild.compiler.compile.constants import (
+    MACRO_ACTUAL_TEST_CTE_NAME,
+    SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM,
+    SQL_TEST_FACT_CACHE_NAMESPACE,
+)
 from sqlbuild.compiler.compile.models import (
     CompileSqlTestCte,
     ScopeRelationshipBuild,
     ScopeRelationshipFault,
 )
 from sqlbuild.compiler.compile.types import SqlTestMode
-from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveredSqlTestFile
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 from sqlbuild.compiler.scopes.main._resolve_scope_path_visibility import (
     resolve_scope_path_visibility,
 )
@@ -44,17 +50,26 @@
 
 
 def build_scope_relationship_grants(
-    *, discovered_inputs: DiscoveredProjectInputs, index: ScopeIndex
+    *,
+    discovered_inputs: DiscoveredProjectInputs,
+    index: ScopeIndex,
+    compile_cache_dir: Path | None = None,
 ) -> ScopeRelationshipBuild:
     """Return expected-model grants while retaining independent extraction faults."""
 
     lookup: ScopeLookup = build_scope_lookup(index=index)
     shared_declarations: _SharedDeclarations = _shared_declarations_by_directory(lookup=lookup)
-    test_grants, test_faults = _test_relationship_grants(
-        discovered_inputs=discovered_inputs,
-        lookup=lookup,
-        shared_declarations=shared_declarations,
-    )
+    with FactCacheStore(
+        root=compile_cache_dir,
+        namespace=SQL_TEST_FACT_CACHE_NAMESPACE,
+        algorithm=SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM,
+    ) as fact_cache:
+        test_grants, test_faults = _test_relationship_grants(
+            discovered_inputs=discovered_inputs,
+            lookup=lookup,
+            shared_declarations=shared_declarations,
+            fact_cache=fact_cache,
+        )
     scenario_grants, scenario_faults = _scenario_relationship_grants(
         discovered_inputs=discovered_inputs,
         lookup=lookup,
@@ -71,17 +86,43 @@ def _test_relationship_grants(
     discovered_inputs: DiscoveredProjectInputs,
     lookup: ScopeLookup,
     shared_declarations: _SharedDeclarations,
+    fact_cache: FactCacheStore,
 ) -> tuple[tuple[GrantRecord, ...], tuple[ScopeRelationshipFault, ...]]:
     grants: list[GrantRecord] = []
     faults: list[ScopeRelationshipFault] = []
-    for test_file in discovered_inputs.test_files:
-        for block in test_file.blocks:
+    cache_keys: dict[int, str] = (
+        {
+            file_index: _expected_models_fact_key(test_file=test_file, fact_cache=fact_cache)
+            for file_index, test_file in enumerate(discovered_inputs.test_files)
+        }
+        if fact_cache.enabled
+        else {}
+    )
+    cached_names: dict[str, object] = fact_cache.read_many(
+        tuple(
+            (_expected_models_fact_slot(discovered_inputs.test_files[file_index]), cache_key)
+            for file_index, cache_key in cache_keys.items()
+        )
+    )
+    for file_index, test_file in enumerate(discovered_inputs.test_files):
+        cache_key: str | None = cache_keys.get(file_index)
+        cached_file_names: tuple[tuple[str, ...], ...] | None = _cached_expected_names(
+            value=None if cache_key is None else cached_names.get(cache_key),
+            block_count=len(test_file.blocks),
+        )
+        extracted_names: list[tuple[str, ...]] = []
+        for block_index, block in enumerate(test_file.blocks):
             try:
-                expected_names: tuple[str, ...] = extract_sql_test_expected_model_names(
-                    sql=block.sql_body,
-                    file_label=str(test_file.relative_path),
-                    mode=block.mode,
+                expected_names: tuple[str, ...] = (
+                    cached_file_names[block_index]
+                    if cached_file_names is not None
+                    else extract_sql_test_expected_model_names(
+                        sql=block.sql_body,
+                        file_label=str(test_file.relative_path),
+                        mode=block.mode,
+                    )
                 )
+                extracted_names.append(expected_names)
                 grants.extend(
                     _expected_model_grants(
                         lookup=lookup,
@@ -106,9 +147,43 @@ def _test_relationship_grants(
                     )
             except Exception as error:
                 faults.append(ScopeRelationshipFault(test_file.relative_path, str(error)))
+        if (
+            cache_key is not None
+            and cached_file_names is None
+            and len(extracted_names) == len(test_file.blocks)
+        ):
+            fact_cache.stage(
+                key=cache_key,
+                slot=_expected_models_fact_slot(test_file),
+                value=tuple(extracted_names),
+            )
     return tuple(grants), tuple(faults)
 
 
+def _expected_models_fact_key(
+    *, test_file: DiscoveredSqlTestFile, fact_cache: FactCacheStore
+) -> str:
+    parts: list[str] = [str(test_file.relative_path)]
+    for block in test_file.blocks:
+        parts.extend((block.sql_body, block.mode.value))
+    return fact_cache.key(*parts)
+
+
+def _cached_expected_names(
+    *, value: object, block_count: int
+) -> tuple[tuple[str, ...], ...] | None:
+    if not isinstance(value, tuple) or len(value) != block_count:
+        return None
+    for names in value:
+        if not isinstance(names, tuple) or not all(isinstance(name, str) for name in names):
+            return None
+    return cast(tuple[tuple[str, ...], ...], value)
+
+
+def _expected_models_fact_slot(test_file: DiscoveredSqlTestFile) -> str:
+    return f"expected:{test_file.relative_path}"
+
+
 def _scenario_relationship_grants(
     *,
     discovered_inputs: DiscoveredProjectInputs,
diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py b/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py
index f99cabbe12..6482191a90 100644
--- a/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py
+++ b/src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py
@@ -32,7 +32,9 @@
     extract_assertion_target_model_names,
     extract_sql_test_ctes,
 )
-from sqlbuild.compiler.compile._helpers.sql_tests.native import extract_expanded_sql_tests
+from sqlbuild.compiler.compile._helpers.sql_tests.native import (
+    extract_expanded_sql_tests_cached,
+)
 from sqlbuild.compiler.compile.exceptions import CompileInputError
 from sqlbuild.compiler.compile.models import (
     AuthoredSqlExpansionResult,
@@ -102,7 +104,6 @@ def build_test_inputs_with_cache(
     """Build SQL test inputs inside the compile timing boundary."""
 
     with record_compile_timing("test_input_compile_ms"):
-        del compile_cache_dir
         return build_test_inputs(
             discovered_inputs=discovered_inputs,
             effective_vars=effective_vars,
@@ -111,6 +112,7 @@ def build_test_inputs_with_cache(
             declaration_expansion=declaration_expansion,
             external_sql_reference_resolver=external_sql_reference_resolver,
             sql_function_inputs=sql_function_inputs,
+            compile_cache_dir=compile_cache_dir,
         )
 
 
@@ -123,6 +125,7 @@ def build_test_inputs(
     declaration_expansion: DeclarationExpansionContext,
     external_sql_reference_resolver: ExternalSqlReferenceResolver | None = None,
     sql_function_inputs: tuple[CompileSqlFunctionInput, ...] = (),
+    compile_cache_dir: Path | None = None,
 ) -> tuple[CompileSqlTestInput, ...]:
     """Build compile-time test inputs from discovered SQL-native test blocks."""
 
@@ -276,10 +279,11 @@ def build_test_inputs(
                         test_case=test_case,
                     )
                 )
-    test_ctes_batch: tuple[CompileSqlTestCtes, ...] = extract_expanded_sql_tests(
-        tuple(
+    test_ctes_batch: tuple[CompileSqlTestCtes, ...] = extract_expanded_sql_tests_cached(
+        tests=tuple(
             (test.sql_body, str(test.test_file.relative_path), test.mode) for test in expanded_tests
-        )
+        ),
+        cache_root=compile_cache_dir,
     )
     for test, test_ctes in zip(expanded_tests, test_ctes_batch, strict=True):
         validate_test_ctes(
diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/target.py b/src/sqlbuild/compiler/compile/_helpers/attachment/target.py
index dbbc49ea39..3d7b954235 100644
--- a/src/sqlbuild/compiler/compile/_helpers/attachment/target.py
+++ b/src/sqlbuild/compiler/compile/_helpers/attachment/target.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 
 from sqlbuild.compiler.compile._helpers.attachment.core import build_effective_vars
@@ -10,11 +9,8 @@
     resolve_early_model_templates,
     resolve_run_id,
 )
-from sqlbuild.compiler.compile.constants import (
-    COMPILE_CACHE_DISABLE_ENV_VAR,
-    COMPILE_CACHE_DISABLE_VALUE,
-)
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.fact_cache.main._compile_cache_root import compile_cache_root
 from sqlbuild.spec.contracts.main.resolve_target_config import resolve_target_config
 from sqlbuild.spec.contracts.main.resolve_target_name import resolve_target_name
 from sqlbuild.spec.contracts.models import TargetConfig
@@ -42,16 +38,10 @@ def build_compile_target_context(
         if target_name is not None
         else None
     )
-    cache_disabled: bool = (
-        no_cache
-        or (target_config is not None and target_config.compile_cache is False)
-        or os.environ.get(COMPILE_CACHE_DISABLE_ENV_VAR) == COMPILE_CACHE_DISABLE_VALUE
-    )
-    project_dir: Path | None = discovered_inputs.project_dir
-    cache_dir: Path | None = (
-        None
-        if cache_disabled or project_dir is None
-        else project_dir / "target" / "cache" / "compiler"
+    cache_dir: Path | None = compile_cache_root(
+        project_dir=discovered_inputs.project_dir,
+        target_config=target_config,
+        no_cache=no_cache,
     )
     return target_name, target_config, cache_dir
 
diff --git a/src/sqlbuild/compiler/compile/_helpers/sql_tests/native.py b/src/sqlbuild/compiler/compile/_helpers/sql_tests/native.py
index 144c3fdd4f..648fb30043 100644
--- a/src/sqlbuild/compiler/compile/_helpers/sql_tests/native.py
+++ b/src/sqlbuild/compiler/compile/_helpers/sql_tests/native.py
@@ -2,11 +2,16 @@
 
 from __future__ import annotations
 
+from pathlib import Path
 from typing import Protocol, cast
 
 import orjson
 
 import sqlbuild._native as _native
+from sqlbuild.compiler.compile.constants import (
+    SQL_TEST_FACT_CACHE_ALGORITHM,
+    SQL_TEST_FACT_CACHE_NAMESPACE,
+)
 from sqlbuild.compiler.compile.exceptions import CompileInputError, NativeSqlTestResponseError
 from sqlbuild.compiler.compile.models import (
     CompileDirectLogicSqlTestCtes,
@@ -15,6 +20,7 @@
     CompileSqlTestCtes,
 )
 from sqlbuild.compiler.compile.types import SqlTestMode
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 
 
 class _NativeSqlTestModule(Protocol):
@@ -26,6 +32,81 @@ def extract_sql_tests_json(self, request_json: str) -> str: ...
 _PAIR_LENGTH: int = 2
 
 
+def extract_expanded_sql_tests_cached(
+    *,
+    tests: tuple[tuple[str, str, SqlTestMode], ...],
+    cache_root: Path | None,
+) -> tuple[CompileSqlTestCtes, ...]:
+    """Reuse exact per-file extraction facts and extract only tests of changed files natively."""
+
+    with FactCacheStore(
+        root=cache_root,
+        namespace=SQL_TEST_FACT_CACHE_NAMESPACE,
+        algorithm=SQL_TEST_FACT_CACHE_ALGORITHM,
+    ) as fact_cache:
+        if not fact_cache.enabled:
+            return extract_expanded_sql_tests(tests)
+        indexes_by_file: dict[str, list[int]] = {}
+        for index, (_sql, file_label, _mode) in enumerate(tests):
+            indexes_by_file.setdefault(file_label, []).append(index)
+        keys_by_file: dict[str, str] = {
+            file_label: fact_cache.key(
+                file_label, *_file_test_key_parts(tests=tests, indexes=indexes)
+            )
+            for file_label, indexes in indexes_by_file.items()
+        }
+        cached: dict[str, object] = fact_cache.read_many(tuple(keys_by_file.items()))
+        results: list[CompileSqlTestCtes | None] = [None] * len(tests)
+        missing_files: list[str] = []
+        for file_label, indexes in indexes_by_file.items():
+            file_results: tuple[CompileSqlTestCtes, ...] | None = _cached_file_tests(
+                value=cached.get(keys_by_file[file_label]),
+                modes=tuple(tests[index][2] for index in indexes),
+            )
+            if file_results is None:
+                missing_files.append(file_label)
+                continue
+            for index, test_ctes in zip(indexes, file_results, strict=True):
+                results[index] = test_ctes
+        missing_indexes: list[int] = []
+        for file_label in missing_files:
+            missing_indexes.extend(indexes_by_file[file_label])
+        if missing_indexes:
+            extracted: tuple[CompileSqlTestCtes, ...] = extract_expanded_sql_tests(
+                tuple(tests[index] for index in missing_indexes)
+            )
+            for index, test_ctes in zip(missing_indexes, extracted, strict=True):
+                results[index] = test_ctes
+            for file_label in missing_files:
+                fact_cache.stage(
+                    key=keys_by_file[file_label],
+                    slot=file_label,
+                    value=tuple(results[index] for index in indexes_by_file[file_label]),
+                )
+        return tuple(result for result in results if result is not None)
+
+
+def _file_test_key_parts(
+    *, tests: tuple[tuple[str, str, SqlTestMode], ...], indexes: list[int]
+) -> list[str]:
+    parts: list[str] = []
+    for index in indexes:
+        sql, _file_label, mode = tests[index]
+        parts.extend((sql, mode.value))
+    return parts
+
+
+def _cached_file_tests(
+    *, value: object, modes: tuple[SqlTestMode, ...]
+) -> tuple[CompileSqlTestCtes, ...] | None:
+    if not isinstance(value, tuple) or len(value) != len(modes):
+        return None
+    for test_ctes, mode in zip(value, modes, strict=True):
+        if not isinstance(test_ctes, CompileSqlTestCtes) or test_ctes.mode is not mode:
+            return None
+    return cast(tuple[CompileSqlTestCtes, ...], value)
+
+
 def extract_expanded_sql_tests(
     tests: tuple[tuple[str, str, SqlTestMode], ...],
 ) -> tuple[CompileSqlTestCtes, ...]:
diff --git a/src/sqlbuild/compiler/compile/constants.py b/src/sqlbuild/compiler/compile/constants.py
index dba6ddb660..e1d88f29fe 100644
--- a/src/sqlbuild/compiler/compile/constants.py
+++ b/src/sqlbuild/compiler/compile/constants.py
@@ -132,6 +132,9 @@
 )
 COMPILE_CACHE_DISABLE_ENV_VAR: str = "SQLBUILD_DISABLE_COMPILE_CACHE"
 COMPILE_CACHE_DISABLE_VALUE: str = "1"
+SQL_TEST_FACT_CACHE_NAMESPACE: str = "sql-tests"
+SQL_TEST_FACT_CACHE_ALGORITHM: str = "expanded-sql-test-ctes-v2"
+SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM: str = "sql-test-expected-models-v2"
 COMPACT_ANALYSIS_RESPONSE_LENGTH: int = 2
 COMPACT_RELATION_STUB_PREFIX: str = "__sqlbuild_project_input_"
 COMPACT_REFERENCE_MARKER_PATTERN: re.Pattern[str] = re.compile(
diff --git a/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py b/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py
index c9b972a175..35bbaca3d6 100644
--- a/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py
+++ b/src/sqlbuild/compiler/compile/main/_build_compile_inputs.py
@@ -124,7 +124,9 @@ def build_compile_inputs(
     resolved_run_id: str = resolve_run_id(selected_run_id=run_id)
     loaded_macros: dict[str, LoadedMacro] = load_project_macros(discovered_inputs.macro_files)
     declaration_scope: DeclarationScopeBuild = build_declaration_scope(
-        discovered_inputs=discovered_inputs, loaded_macros=loaded_macros
+        discovered_inputs=discovered_inputs,
+        loaded_macros=loaded_macros,
+        compile_cache_dir=compile_cache_dir,
     )
     model_context: ModelInputBuildContext = ModelInputBuildContext(
         effective_vars=effective_vars,
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py
index 908db8b327..d8026940b2 100644
--- a/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/aggregation.py
@@ -65,6 +65,7 @@
     DiscoveryFileFault,
     TolerantScopeDiscovery,
 )
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
 from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig
 
@@ -76,6 +77,7 @@ def build_discovered_project_inputs(
     local_config: LocalConfig,
     sql_analysis_enabled: bool,
     extract_output_column_locations: bool = True,
+    fact_cache: FactCacheStore | None = None,
 ) -> DiscoveredProjectInputs:
     """Discover all project files and functions into one inputs bundle."""
 
@@ -83,7 +85,7 @@ def build_discovered_project_inputs(
         operation_kind="project", operation_name="discovery_declaration_parse"
     ) as declaration_lifecycle:
         source_files: tuple[DiscoveredSourceFile, ...] = discover_source_files(
-            project_dir=project_dir
+            project_dir=project_dir, fact_cache=fact_cache
         )
         model_files: tuple[DiscoveredSqlModelFile, ...] = discover_model_files(
             project_dir=project_dir,
@@ -110,7 +112,9 @@ def build_discovered_project_inputs(
             project_dir=project_dir
         )
         seed_files: tuple[DiscoveredSeedFile, ...] = discover_seed_files(project_dir=project_dir)
-        test_files: tuple[DiscoveredSqlTestFile, ...] = discover_test_files(project_dir=project_dir)
+        test_files: tuple[DiscoveredSqlTestFile, ...] = discover_test_files(
+            project_dir=project_dir, fact_cache=fact_cache
+        )
         scenario_files: tuple[DiscoveredSqlScenarioFile, ...] = discover_scenario_files(
             project_dir=project_dir
         )
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/cache_root.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/cache_root.py
new file mode 100644
index 0000000000..b177f131b8
--- /dev/null
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/cache_root.py
@@ -0,0 +1,45 @@
+"""Compile-cache root resolution for discovery facts."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from sqlbuild.compiler.discovery.models import DiscoveryCacheRequest
+from sqlbuild.compiler.fact_cache.main._compile_cache_root import compile_cache_root
+from sqlbuild.spec.contracts.exceptions import SpecConfigError
+from sqlbuild.spec.contracts.main.resolve_target_config import resolve_target_config
+from sqlbuild.spec.contracts.main.resolve_target_name import resolve_target_name
+from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig, TargetConfig
+
+
+def discovery_cache_root(
+    *,
+    project_dir: Path,
+    project_config: ProjectConfig,
+    local_config: LocalConfig,
+    cache_request: DiscoveryCacheRequest | None,
+) -> Path | None:
+    """Return the compile-cache root for discovery facts, or None when caching is off."""
+
+    if cache_request is None or cache_request.no_cache:
+        return None
+    try:
+        target_name: str | None = resolve_target_name(
+            project_config=project_config,
+            local_config=local_config,
+            selected_target=cache_request.selected_target,
+        )
+        target_config: TargetConfig | None = (
+            resolve_target_config(
+                project_config=project_config,
+                local_config=local_config,
+                target_name=target_name,
+            )
+            if target_name is not None
+            else None
+        )
+    except SpecConfigError:
+        return None
+    return compile_cache_root(
+        project_dir=project_dir, target_config=target_config, no_cache=cache_request.no_cache
+    )
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/cached_files.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/cached_files.py
new file mode 100644
index 0000000000..1b363b92ab
--- /dev/null
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/cached_files.py
@@ -0,0 +1,81 @@
+"""Compact fact-cache codecs for discovered files whose raw contents are re-read live."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+
+from sqlbuild.compiler.discovery._helpers.yml.sources import parse_sources_yml
+from sqlbuild.compiler.discovery.constants import DISCOVERY_SOURCE_FACT_KIND
+from sqlbuild.compiler.discovery.models import (
+    DiscoveredSourceFile,
+    DiscoveredSqlTestFile,
+)
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
+
+
+def parse_source_file_with_cache(
+    *,
+    project_dir: Path,
+    file_path: Path,
+    relative_path: Path,
+    fact_cache: FactCacheStore | None,
+) -> DiscoveredSourceFile:
+    """Parse one source YAML file, reusing its exact cached fact when the contents match."""
+
+    contents: str = file_path.read_text(encoding="utf-8")
+    cache_key: str | None = (
+        fact_cache.key(DISCOVERY_SOURCE_FACT_KIND, str(project_dir), str(file_path), contents)
+        if fact_cache is not None and fact_cache.enabled
+        else None
+    )
+    slot: str = f"{DISCOVERY_SOURCE_FACT_KIND}:{file_path}"
+    if fact_cache is not None and cache_key is not None:
+        cached: DiscoveredSourceFile | None = decode_cached_source_file(
+            fact=fact_cache.read_many(((slot, cache_key),)).get(cache_key),
+            file_path=file_path,
+            contents=contents,
+        )
+        if cached is not None:
+            return cached
+    source_file: DiscoveredSourceFile = DiscoveredSourceFile(
+        file_path=file_path,
+        relative_path=relative_path,
+        contents=contents,
+        source_entries=parse_sources_yml(contents=contents, file_path=file_path),
+    )
+    if fact_cache is not None and cache_key is not None:
+        fact_cache.stage(key=cache_key, slot=slot, value=encode_cached_source_file(source_file))
+    return source_file
+
+
+def encode_cached_source_file(source_file: DiscoveredSourceFile) -> object:
+    """Return a source fact without raw YAML."""
+
+    return replace(source_file, contents="")
+
+
+def decode_cached_source_file(
+    *, fact: object, file_path: Path, contents: str
+) -> DiscoveredSourceFile | None:
+    """Rebuild one cached source file from its fact and exact live YAML."""
+
+    if not isinstance(fact, DiscoveredSourceFile) or fact.file_path != file_path:
+        return None
+    return replace(fact, contents=contents)
+
+
+def encode_cached_sql_test_file(test_file: DiscoveredSqlTestFile) -> object:
+    """Return a SQL test fact without raw file contents."""
+
+    return replace(test_file, contents="")
+
+
+def decode_cached_sql_test_file(
+    *, fact: object, file_path: Path, contents: str
+) -> DiscoveredSqlTestFile | None:
+    """Rebuild one cached SQL test file from its fact and exact live contents."""
+
+    if not isinstance(fact, DiscoveredSqlTestFile) or fact.file_path != file_path:
+        return None
+    return replace(fact, contents=contents)
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
index 3cf840d014..da75d7a2b3 100644
--- a/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/core.py
@@ -5,6 +5,7 @@
 import ast
 import importlib.util
 import inspect
+import re
 import sys
 from collections.abc import Callable, Iterable, Iterator
 from dataclasses import dataclass, field
@@ -15,6 +16,12 @@
 
 from pydantic import ValidationError
 
+from sqlbuild.compiler.discovery._helpers.filesystem.cached_files import (
+    parse_source_file_with_cache,
+)
+from sqlbuild.compiler.discovery._helpers.filesystem.model_files import (
+    discover_matched_model_file,
+)
 from sqlbuild.compiler.discovery._helpers.filesystem.named_declarations import (
     named_declaration_files,
     named_declaration_roots,
@@ -25,21 +32,16 @@
 from sqlbuild.compiler.discovery._helpers.sql.declarations import (
     parse_constant_declaration_file,
     parse_enum_declaration_file,
-    parse_model_constant_declarations,
-    parse_model_enum_declarations,
     parse_model_schema_declaration_file,
 )
 from sqlbuild.compiler.discovery._helpers.sql.functions import parse_function_sql
 from sqlbuild.compiler.discovery._helpers.sql.hooks import parse_sql_hook_file
 from sqlbuild.compiler.discovery._helpers.sql.model_files import (
-    model_header_column_locations,
-    model_output_column_locations,
-    parse_model_sql,
-    prepare_model_file_headers,
+    match_model_header,
+    prepare_matched_model_file_headers,
 )
 from sqlbuild.compiler.discovery._helpers.sql.scenarios import parse_sql_scenario_file
 from sqlbuild.compiler.discovery._helpers.yml.schema import parse_schema_yml
-from sqlbuild.compiler.discovery._helpers.yml.sources import parse_sources_yml
 from sqlbuild.compiler.discovery.constants import (
     CANONICAL_AUTHORED_ROOTS,
     PYTHON_INIT_MODULE_STEM,
@@ -90,6 +92,7 @@
     NamedDeclarationRoot,
 )
 from sqlbuild.compiler.discovery.types import ScopedDeclarationFile
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 from sqlbuild.compiler.resource_names.main._validate_resource_identity import (
     validate_resource_identity,
 )
@@ -375,17 +378,18 @@ def discover_model_files(
         except (OSError, UnicodeError, ValueError, SyntaxError) as error:
             loaded_model_files.append((file_path, None, error))
 
-    prepare_model_file_headers(
-        [
-            contents
-            for _path, contents, error in loaded_model_files
-            if contents is not None and error is None
-        ]
-    )
+    header_matches: list[re.Match[str] | None] = [
+        match_model_header(contents) if contents is not None and error is None else None
+        for _path, contents, error in loaded_model_files
+    ]
+    prepare_matched_model_file_headers(header_matches)
     discovered_model_files: list[DiscoveredSqlModelFile] = []
     contents: str | None
     read_error: Exception | None
-    for file_path, contents, read_error in loaded_model_files:
+    header_match: re.Match[str] | None
+    for (file_path, contents, read_error), header_match in zip(
+        loaded_model_files, header_matches, strict=True
+    ):
         if read_error is not None:
             if on_fault is None:
                 raise read_error
@@ -395,10 +399,11 @@ def discover_model_files(
             raise ModelSqlParseError("Model file read returned neither contents nor an error")
         try:
             discovered_model_files.append(
-                _discover_model_file(
-                    project_dir=project_dir,
+                discover_matched_model_file(
                     file_path=file_path,
+                    relative_path=_project_relative_path(path=file_path, project_dir=project_dir),
                     contents=contents,
+                    header_match=header_match,
                     extract_implicit_alias_columns=extract_implicit_alias_columns,
                     extract_output_column_locations=extract_output_column_locations,
                 )
@@ -411,48 +416,6 @@ def discover_model_files(
     return tuple(discovered_model_files)
 
 
-def _discover_model_file(
-    *,
-    project_dir: Path,
-    file_path: Path,
-    contents: str,
-    extract_implicit_alias_columns: bool,
-    extract_output_column_locations: bool,
-) -> DiscoveredSqlModelFile:
-    header_values, query_sql = parse_model_sql(contents=contents, file_path=file_path)
-    relative_path: Path = _project_relative_path(path=file_path, project_dir=project_dir)
-    return DiscoveredSqlModelFile(
-        file_path=file_path,
-        relative_path=relative_path,
-        contents=contents,
-        header_values=header_values,
-        header_column_locations=model_header_column_locations(
-            contents=contents, relative_path=relative_path
-        ),
-        output_column_locations=(
-            model_output_column_locations(
-                contents=contents,
-                relative_path=relative_path,
-                extract_implicit_alias_columns=extract_implicit_alias_columns,
-            )
-            if extract_output_column_locations
-            else {}
-        ),
-        query_sql=query_sql,
-        enum_declarations=parse_model_enum_declarations(
-            raw_value=header_values.get("enums"),
-            model_name=file_path.stem,
-            relative_path=relative_path,
-        ),
-        constant_declarations=parse_model_constant_declarations(
-            raw_value=header_values.get("constants"),
-            model_name=file_path.stem,
-            relative_path=relative_path,
-        ),
-        extract_implicit_alias_columns=extract_implicit_alias_columns,
-    )
-
-
 def discover_enum_files(
     *,
     project_dir: Path,
@@ -734,7 +697,10 @@ def discover_schema_files(*, project_dir: Path) -> tuple[DiscoveredSchemaFile, .
 
 
 def discover_source_files(
-    *, project_dir: Path, on_fault: Callable[[DiscoveryFileFault], None] | None = None
+    *,
+    project_dir: Path,
+    on_fault: Callable[[DiscoveryFileFault], None] | None = None,
+    fact_cache: FactCacheStore | None = None,
 ) -> tuple[DiscoveredSourceFile, ...]:
     """Discover source declaration YAML files under sources/."""
 
@@ -747,12 +713,11 @@ def discover_source_files(
     )
 
     def parse(file_path: Path) -> DiscoveredSourceFile:
-        contents: str = file_path.read_text(encoding="utf-8")
-        return DiscoveredSourceFile(
+        return parse_source_file_with_cache(
+            project_dir=project_dir,
             file_path=file_path,
             relative_path=_project_relative_path(path=file_path, project_dir=project_dir),
-            contents=contents,
-            source_entries=parse_sources_yml(contents=contents, file_path=file_path),
+            fact_cache=fact_cache,
         )
 
     return _parse_discovered_paths(
@@ -782,6 +747,7 @@ def discover_test_files(
     project_dir: Path,
     selected_paths: frozenset[Path] | None = None,
     on_fault: Callable[[DiscoveryFileFault], None] | None = None,
+    fact_cache: FactCacheStore | None = None,
 ) -> tuple[DiscoveredSqlTestFile, ...]:
     """Discover SQL-native unit test files under tests/unit/."""
 
@@ -799,6 +765,7 @@ def discover_test_files(
         project_dir=project_dir,
         file_paths=file_paths,
         on_fault=on_fault,
+        fact_cache=fact_cache,
     )
 
 
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/model_files.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/model_files.py
new file mode 100644
index 0000000000..701ce4b7e7
--- /dev/null
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/model_files.py
@@ -0,0 +1,64 @@
+"""Assembly of one discovered SQL model file from its pre-matched MODEL header."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from sqlbuild.compiler.discovery._helpers.sql.declarations import (
+    parse_model_constant_declarations,
+    parse_model_enum_declarations,
+)
+from sqlbuild.compiler.discovery._helpers.sql.model_files import (
+    matched_model_header_column_locations,
+    matched_model_output_column_locations,
+    parse_matched_model_sql,
+)
+from sqlbuild.compiler.discovery.models import DiscoveredSqlModelFile
+
+
+def discover_matched_model_file(
+    *,
+    file_path: Path,
+    relative_path: Path,
+    contents: str,
+    header_match: re.Match[str] | None,
+    extract_implicit_alias_columns: bool,
+    extract_output_column_locations: bool,
+) -> DiscoveredSqlModelFile:
+    """Build one model file from contents whose MODEL header was matched exactly once."""
+
+    header_values, query_sql = parse_matched_model_sql(
+        header_match=header_match, file_path=file_path
+    )
+    return DiscoveredSqlModelFile(
+        file_path=file_path,
+        relative_path=relative_path,
+        contents=contents,
+        header_values=header_values,
+        header_column_locations=matched_model_header_column_locations(
+            contents=contents, header_match=header_match, relative_path=relative_path
+        ),
+        output_column_locations=(
+            matched_model_output_column_locations(
+                contents=contents,
+                header_match=header_match,
+                relative_path=relative_path,
+                extract_implicit_alias_columns=extract_implicit_alias_columns,
+            )
+            if extract_output_column_locations
+            else {}
+        ),
+        query_sql=query_sql,
+        enum_declarations=parse_model_enum_declarations(
+            raw_value=header_values.get("enums"),
+            model_name=file_path.stem,
+            relative_path=relative_path,
+        ),
+        constant_declarations=parse_model_constant_declarations(
+            raw_value=header_values.get("constants"),
+            model_name=file_path.stem,
+            relative_path=relative_path,
+        ),
+        extract_implicit_alias_columns=extract_implicit_alias_columns,
+    )
diff --git a/src/sqlbuild/compiler/discovery/_helpers/filesystem/sql_test_files.py b/src/sqlbuild/compiler/discovery/_helpers/filesystem/sql_test_files.py
index 87d459a03f..c887759507 100644
--- a/src/sqlbuild/compiler/discovery/_helpers/filesystem/sql_test_files.py
+++ b/src/sqlbuild/compiler/discovery/_helpers/filesystem/sql_test_files.py
@@ -5,17 +5,25 @@
 from collections.abc import Callable
 from pathlib import Path
 
+from sqlbuild.compiler.discovery._helpers.filesystem.cached_files import (
+    decode_cached_sql_test_file,
+    encode_cached_sql_test_file,
+)
 from sqlbuild.compiler.discovery._helpers.sql.tests import (
     parse_sql_test_file,
     prepare_sql_test_file_headers,
 )
-from sqlbuild.compiler.discovery.constants import SQL_TESTS_OWNERSHIP_ROOT
+from sqlbuild.compiler.discovery.constants import (
+    DISCOVERY_SQL_TEST_FACT_KIND,
+    SQL_TESTS_OWNERSHIP_ROOT,
+)
 from sqlbuild.compiler.discovery.exceptions import SqlTestParseError
 from sqlbuild.compiler.discovery.models import (
     DiscoveredSqlTestBlock,
     DiscoveredSqlTestFile,
     DiscoveryFileFault,
 )
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 
 _TEST_FILE_BATCH_SIZE: int = 512
 
@@ -25,6 +33,7 @@ def discover_sql_test_files(
     project_dir: Path,
     file_paths: tuple[Path, ...],
     on_fault: Callable[[DiscoveryFileFault], None] | None,
+    fact_cache: FactCacheStore | None = None,
 ) -> tuple[DiscoveredSqlTestFile, ...]:
     """Read and parse ordered SQL test paths through bounded native header batches."""
 
@@ -35,12 +44,30 @@ def discover_sql_test_files(
         except (OSError, UnicodeError, ValueError, SyntaxError) as error:
             loaded.append((file_path, None, error))
 
+    cache_keys: dict[Path, str] = (
+        {
+            file_path: fact_cache.key(
+                DISCOVERY_SQL_TEST_FACT_KIND, str(project_dir), str(file_path), contents
+            )
+            for file_path, contents, error in loaded
+            if contents is not None and error is None
+        }
+        if fact_cache is not None and fact_cache.enabled
+        else {}
+    )
+    cached_facts: dict[str, object] = (
+        fact_cache.read_many(
+            tuple((_fact_slot(file_path), cache_key) for file_path, cache_key in cache_keys.items())
+        )
+        if fact_cache is not None
+        else {}
+    )
     discovered: list[DiscoveredSqlTestFile] = []
     for batch_start in range(0, len(loaded), _TEST_FILE_BATCH_SIZE):
         batch: list[tuple[Path, str | None, Exception | None]] = loaded[
             batch_start : batch_start + _TEST_FILE_BATCH_SIZE
         ]
-        _prepare_batch(batch)
+        _prepare_batch([item for item in batch if cache_keys.get(item[0]) not in cached_facts])
         for file_path, contents, read_error in batch:
             if read_error is not None:
                 if on_fault is None:
@@ -49,6 +76,17 @@ def discover_sql_test_files(
                 continue
             if contents is None:
                 raise SqlTestParseError("SQL test file read returned neither contents nor an error")
+            cache_key: str | None = cache_keys.get(file_path)
+            cached_test_file: DiscoveredSqlTestFile | None = (
+                decode_cached_sql_test_file(
+                    fact=cached_facts.get(cache_key), file_path=file_path, contents=contents
+                )
+                if cache_key is not None
+                else None
+            )
+            if cached_test_file is not None:
+                discovered.append(cached_test_file)
+                continue
             try:
                 blocks: tuple[DiscoveredSqlTestBlock, ...] = parse_sql_test_file(
                     contents=contents, file_path=file_path
@@ -58,18 +96,27 @@ def discover_sql_test_files(
                     raise
                 on_fault(_file_fault(project_dir=project_dir, path=file_path, error=error))
                 continue
-            discovered.append(
-                DiscoveredSqlTestFile(
-                    file_path=file_path,
-                    relative_path=file_path.relative_to(project_dir),
-                    contents=contents,
-                    blocks=blocks,
-                    ownership_root=Path(SQL_TESTS_OWNERSHIP_ROOT),
-                )
+            test_file: DiscoveredSqlTestFile = DiscoveredSqlTestFile(
+                file_path=file_path,
+                relative_path=file_path.relative_to(project_dir),
+                contents=contents,
+                blocks=blocks,
+                ownership_root=Path(SQL_TESTS_OWNERSHIP_ROOT),
             )
+            discovered.append(test_file)
+            if fact_cache is not None and cache_key is not None:
+                fact_cache.stage(
+                    key=cache_key,
+                    slot=_fact_slot(file_path),
+                    value=encode_cached_sql_test_file(test_file),
+                )
     return tuple(discovered)
 
 
+def _fact_slot(file_path: Path) -> str:
+    return f"{DISCOVERY_SQL_TEST_FACT_KIND}:{file_path}"
+
+
 def _prepare_batch(batch: list[tuple[Path, str | None, Exception | None]]) -> None:
     contents_batch: list[str] = []
     for _path, contents, error in batch:
diff --git a/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py b/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py
index 3d347d4242..4340aa967c 100644
--- a/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py
+++ b/src/sqlbuild/compiler/discovery/_helpers/sql/model_files.py
@@ -110,10 +110,23 @@ def put(self, *, header: str, tokenization: _ModelHeaderTokenization) -> None:
 _MODEL_HEADER_TOKEN_CACHE: _ModelHeaderTokenCache = _ModelHeaderTokenCache()
 
 
+def match_model_header(contents: str) -> re.Match[str] | None:
+    """Match the MODEL(...) header once so callers can reuse it for every header projection."""
+
+    return _MODEL_HEADER_PATTERN.match(contents)
+
+
 def parse_model_sql(*, contents: str, file_path: Path) -> tuple[dict[str, object], str]:
     """Parse a raw SQL model file into header values and SQL body."""
 
-    header_match: re.Match[str] | None = _MODEL_HEADER_PATTERN.match(contents)
+    return parse_matched_model_sql(header_match=match_model_header(contents), file_path=file_path)
+
+
+def parse_matched_model_sql(
+    *, header_match: re.Match[str] | None, file_path: Path
+) -> tuple[dict[str, object], str]:
+    """Parse header values and SQL body from one precomputed MODEL header match."""
+
     if header_match is None:
         raise ModelSqlParseError(
             f"SQL model '{file_path}' must start with a MODEL(...) header as the first "
@@ -136,7 +149,16 @@ def model_header_column_locations(
 ) -> dict[str, SourceLocation]:
     """Return authored locations for MODEL(columns) declarations."""
 
-    header_match: re.Match[str] | None = _MODEL_HEADER_PATTERN.match(contents)
+    return matched_model_header_column_locations(
+        contents=contents, header_match=match_model_header(contents), relative_path=relative_path
+    )
+
+
+def matched_model_header_column_locations(
+    *, contents: str, header_match: re.Match[str] | None, relative_path: Path
+) -> dict[str, SourceLocation]:
+    """Return authored MODEL(columns) locations from one precomputed header match."""
+
     if header_match is None:
         return {}
     return header_column_locations(
@@ -261,7 +283,23 @@ def model_output_column_locations(
 ) -> dict[str, SourceLocation]:
     """Return authored locations for simple SELECT output expressions."""
 
-    header_match: re.Match[str] | None = _MODEL_HEADER_PATTERN.match(contents)
+    return matched_model_output_column_locations(
+        contents=contents,
+        header_match=match_model_header(contents),
+        relative_path=relative_path,
+        extract_implicit_alias_columns=extract_implicit_alias_columns,
+    )
+
+
+def matched_model_output_column_locations(
+    *,
+    contents: str,
+    header_match: re.Match[str] | None,
+    relative_path: Path,
+    extract_implicit_alias_columns: bool,
+) -> dict[str, SourceLocation]:
+    """Return simple SELECT output locations from one precomputed header match."""
+
     if header_match is None:
         return {}
     sql_start: int = header_match.start("sql")
@@ -604,11 +642,17 @@ def _header_column_relative_locations(
 def prepare_model_file_headers(contents: list[str]) -> None:
     """Batch tokenization for syntactically recognizable MODEL file headers."""
 
+    prepare_matched_model_file_headers([match_model_header(value) for value in contents])
+
+
+def prepare_matched_model_file_headers(header_matches: list[re.Match[str] | None]) -> None:
+    """Batch tokenization for precomputed MODEL header matches."""
+
     prepare_model_header_tokens(
         [
             header_match.group("header")
-            for contents_value in contents
-            if (header_match := _MODEL_HEADER_PATTERN.match(contents_value)) is not None
+            for header_match in header_matches
+            if header_match is not None
         ]
     )
 
diff --git a/src/sqlbuild/compiler/discovery/constants.py b/src/sqlbuild/compiler/discovery/constants.py
index 14cc1a4330..2de8bc4062 100644
--- a/src/sqlbuild/compiler/discovery/constants.py
+++ b/src/sqlbuild/compiler/discovery/constants.py
@@ -76,3 +76,8 @@
     r"|[\"'](?:[^\\)]|\\.)*+"
     r"|\)(?!\s*;))*+)"
 )
+
+DISCOVERY_FACT_CACHE_NAMESPACE: str = "discovery"
+DISCOVERY_FACT_CACHE_ALGORITHM: str = "discovered-files-v1"
+DISCOVERY_SOURCE_FACT_KIND: str = "source"
+DISCOVERY_SQL_TEST_FACT_KIND: str = "sql_test"
diff --git a/src/sqlbuild/compiler/discovery/main/discover.py b/src/sqlbuild/compiler/discovery/main/discover.py
index 114dd9411e..0805f693a7 100644
--- a/src/sqlbuild/compiler/discovery/main/discover.py
+++ b/src/sqlbuild/compiler/discovery/main/discover.py
@@ -7,6 +7,7 @@
 from sqlbuild.compiler.discovery._helpers.filesystem.aggregation import (
     build_discovered_project_inputs,
 )
+from sqlbuild.compiler.discovery._helpers.filesystem.cache_root import discovery_cache_root
 from sqlbuild.compiler.discovery._helpers.filesystem.python_paths import (
     validate_project_python_paths,
 )
@@ -15,8 +16,13 @@
     load_local_config,
     load_project_config,
 )
-from sqlbuild.compiler.discovery.constants import SQL_ANALYSIS_SETTING_KEY
-from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.discovery.constants import (
+    DISCOVERY_FACT_CACHE_ALGORITHM,
+    DISCOVERY_FACT_CACHE_NAMESPACE,
+    SQL_ANALYSIS_SETTING_KEY,
+)
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveryCacheRequest
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
 from sqlbuild.spec.contracts.models import LocalConfig, ProjectConfig
 
@@ -26,6 +32,7 @@ def discover_project_inputs(
     project_dir: Path,
     sql_analysis_enabled_override: bool | None = None,
     extract_output_column_locations: bool = True,
+    cache_request: DiscoveryCacheRequest | None = None,
 ) -> DiscoveredProjectInputs:
     """Load all raw project inputs from disk before semantic resolution."""
 
@@ -34,6 +41,7 @@ def discover_project_inputs(
             project_dir=project_dir,
             sql_analysis_enabled_override=sql_analysis_enabled_override,
             extract_output_column_locations=extract_output_column_locations,
+            cache_request=cache_request,
         )
 
 
@@ -42,12 +50,14 @@ def _discover_project_inputs(
     project_dir: Path,
     sql_analysis_enabled_override: bool | None,
     extract_output_column_locations: bool,
+    cache_request: DiscoveryCacheRequest | None,
 ) -> DiscoveredProjectInputs:
     with OperationLifecycle(operation_kind="project", operation_name="discovery_project_assembly"):
         return _assemble_discovered_project_inputs(
             project_dir=project_dir,
             sql_analysis_enabled_override=sql_analysis_enabled_override,
             extract_output_column_locations=extract_output_column_locations,
+            cache_request=cache_request,
         )
 
 
@@ -56,6 +66,7 @@ def _assemble_discovered_project_inputs(
     project_dir: Path,
     sql_analysis_enabled_override: bool | None,
     extract_output_column_locations: bool,
+    cache_request: DiscoveryCacheRequest | None,
 ) -> DiscoveredProjectInputs:
     project_config: ProjectConfig = load_project_config(project_dir=project_dir)
     local_config: LocalConfig = load_local_config(project_dir=project_dir)
@@ -69,13 +80,24 @@ def _assemble_discovered_project_inputs(
             else project_config.settings.sql_analysis
         )
     )
-    discovered_inputs: DiscoveredProjectInputs = build_discovered_project_inputs(
-        project_dir=project_dir,
-        project_config=project_config,
-        local_config=local_config,
-        sql_analysis_enabled=sql_analysis_enabled,
-        extract_output_column_locations=extract_output_column_locations,
-    )
+    with FactCacheStore(
+        root=discovery_cache_root(
+            project_dir=project_dir,
+            project_config=project_config,
+            local_config=local_config,
+            cache_request=cache_request,
+        ),
+        namespace=DISCOVERY_FACT_CACHE_NAMESPACE,
+        algorithm=DISCOVERY_FACT_CACHE_ALGORITHM,
+    ) as fact_cache:
+        discovered_inputs: DiscoveredProjectInputs = build_discovered_project_inputs(
+            project_dir=project_dir,
+            project_config=project_config,
+            local_config=local_config,
+            sql_analysis_enabled=sql_analysis_enabled,
+            extract_output_column_locations=extract_output_column_locations,
+            fact_cache=fact_cache,
+        )
     validate_discovered_inputs(discovered_inputs)
     from sqlbuild.runtime.event_exporting.main.configure_discovered_event_exporters import (
         configure_discovered_event_exporters,
diff --git a/src/sqlbuild/compiler/discovery/models.py b/src/sqlbuild/compiler/discovery/models.py
index 7afe01b8b6..4f6f0a5ab7 100644
--- a/src/sqlbuild/compiler/discovery/models.py
+++ b/src/sqlbuild/compiler/discovery/models.py
@@ -592,6 +592,14 @@ class DiscoveredPythonNodeFunctions:
     audit_factories: tuple[DiscoveredAuditFactory, ...] = field(default_factory=tuple)
 
 
+@dataclass(frozen=True)
+class DiscoveryCacheRequest:
+    """Compile-cache controls that let discovery reuse exact per-file parse facts."""
+
+    selected_target: str | None = None
+    no_cache: bool = False
+
+
 @dataclass(frozen=True)
 class DiscoveredProjectInputs:
     """All raw project inputs discovered from disk before semantic resolution."""
diff --git a/src/sqlbuild/compiler/fact_cache/__init__.py b/src/sqlbuild/compiler/fact_cache/__init__.py
new file mode 100644
index 0000000000..996c10d790
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/__init__.py
@@ -0,0 +1 @@
+"""Content-addressed local cache for deterministic per-file compile facts."""
diff --git a/src/sqlbuild/compiler/fact_cache/_helpers/__init__.py b/src/sqlbuild/compiler/fact_cache/_helpers/__init__.py
new file mode 100644
index 0000000000..e3ff32475e
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/_helpers/__init__.py
@@ -0,0 +1 @@
+"""Fact cache helpers."""
diff --git a/src/sqlbuild/compiler/fact_cache/_helpers/code_identity.py b/src/sqlbuild/compiler/fact_cache/_helpers/code_identity.py
new file mode 100644
index 0000000000..9d5616bf4b
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/_helpers/code_identity.py
@@ -0,0 +1,88 @@
+"""Identity of the installed code that produces cached compile facts."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import platform
+from functools import cache
+from importlib.metadata import PackageNotFoundError, distribution
+from pathlib import Path
+from typing import Any
+
+_PACKAGE_NAME: str = "sqlbuild"
+_POLYGLOT_PACKAGE_NAME: str = "polyglot-sql-chio"
+_YAML_PACKAGE_NAME: str = "PyYAML"
+_DIRECT_URL_FILE: str = "direct_url.json"
+_SOURCE_SUFFIXES: frozenset[str] = frozenset({".py", ".so", ".pyd"})
+_BYTECODE_CACHE_DIRECTORY: str = "__pycache__"
+_PACKAGE_ROOT_PARENT_DEPTH: int = 3
+
+
+@cache
+def installed_code_identity() -> str:
+    """Return a digest that changes with every released or locally edited code version."""
+
+    digest: Any = hashlib.sha256()
+    for item in (
+        _package_version(_PACKAGE_NAME),
+        _package_version(_POLYGLOT_PACKAGE_NAME),
+        _package_version(_YAML_PACKAGE_NAME),
+        platform.python_implementation(),
+        ".".join(platform.python_version_tuple()[:2]),
+    ):
+        digest.update(item.encode())
+        digest.update(b"\0")
+    if _is_editable_install():
+        package_root: Path = Path(__file__).resolve().parents[_PACKAGE_ROOT_PARENT_DEPTH]
+        for relative_path, size, mtime_ns in _source_file_stats(root=package_root):
+            digest.update(f"{relative_path}\0{size}\0{mtime_ns}\0".encode())
+    return str(digest.hexdigest())
+
+
+def _source_file_stats(*, root: Path) -> list[tuple[str, int, int]]:
+    root_prefix_length: int = len(str(root)) + 1
+    stats: list[tuple[str, int, int]] = []
+    pending: list[str] = [str(root)]
+    while pending:
+        directory: str = pending.pop()
+        try:
+            with os.scandir(directory) as entries:
+                for entry in entries:
+                    if entry.is_dir(follow_symlinks=False):
+                        if entry.name != _BYTECODE_CACHE_DIRECTORY:
+                            pending.append(entry.path)
+                    elif os.path.splitext(entry.name)[1] in _SOURCE_SUFFIXES:
+                        stat: os.stat_result = entry.stat()
+                        stats.append(
+                            (entry.path[root_prefix_length:], stat.st_size, stat.st_mtime_ns)
+                        )
+        except OSError:
+            continue
+    stats.sort()
+    return stats
+
+
+def _package_version(package: str) -> str:
+    try:
+        return distribution(package).version
+    except PackageNotFoundError:
+        return "unknown"
+
+
+def _is_editable_install() -> bool:
+    try:
+        direct_url: str | None = distribution(_PACKAGE_NAME).read_text(_DIRECT_URL_FILE)
+    except PackageNotFoundError:
+        return True
+    if direct_url is None:
+        return False
+    try:
+        payload: object = json.loads(direct_url)
+    except ValueError:
+        return True
+    if not isinstance(payload, dict):
+        return True
+    dir_info: object = payload.get("dir_info")
+    return isinstance(dir_info, dict) and dir_info.get("editable") is True
diff --git a/src/sqlbuild/compiler/fact_cache/_helpers/publication.py b/src/sqlbuild/compiler/fact_cache/_helpers/publication.py
new file mode 100644
index 0000000000..53cabf2787
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/_helpers/publication.py
@@ -0,0 +1,102 @@
+"""Background publication of fact rows in multi-row statements so writes rarely take the GIL."""
+
+from __future__ import annotations
+
+import sqlite3
+import threading
+import zlib
+from contextlib import closing
+from pathlib import Path
+
+from sqlbuild.compiler.fact_cache.classes.fact_publication_registry import (
+    FactPublicationRegistry,
+)
+from sqlbuild.compiler.fact_cache.constants import (
+    FACT_CACHE_CREATE_TABLE_SQL,
+    FACT_CACHE_INSERT_CHUNK_ROWS,
+    FACT_CACHE_INSERT_ROW_SQL,
+    FACT_CACHE_INSERT_SQL,
+    FACT_CACHE_SQLITE_TIMEOUT_SECONDS,
+    FACT_CACHE_WRITE_PRAGMAS,
+    FACT_CACHE_WRITER_THREAD_NAME,
+)
+
+_PUBLICATIONS: FactPublicationRegistry = FactPublicationRegistry()
+
+
+def publish_fact_rows(
+    *, database_path: Path, rows: tuple[tuple[str, str, str, bytes], ...]
+) -> None:
+    """Publish (slot, key, digest, payload) rows after earlier publications to the same file."""
+
+    def write(previous: threading.Thread | None) -> None:
+        _write_rows(database_path=database_path, rows=rows, previous=previous)
+
+    _PUBLICATIONS.start(
+        database_path=database_path, write=write, thread_name=FACT_CACHE_WRITER_THREAD_NAME
+    )
+
+
+def await_fact_publication(*, database_path: Path | None = None) -> None:
+    """Wait until pending publications to one database, or to every database, are durable."""
+
+    _PUBLICATIONS.wait(database_path=database_path)
+
+
+def entry_digest(*, cache_key: str, payload: bytes) -> str:
+    """Return the CRC-32 checksum binding one payload to its exact cache key."""
+
+    return f"{zlib.crc32(payload, zlib.crc32(cache_key.encode() + b'\0')):08x}"
+
+
+def _write_rows(
+    *,
+    database_path: Path,
+    rows: tuple[tuple[str, str, str, bytes], ...],
+    previous: threading.Thread | None,
+) -> None:
+    if previous is not None:
+        previous.join()
+    try:
+        _insert_records(database_path=database_path, records=rows)
+    except sqlite3.DatabaseError as error:
+        if type(error) is sqlite3.DatabaseError:
+            _replace_unreadable_database(database_path=database_path, records=rows)
+    except (sqlite3.Error, OSError):
+        return
+
+
+def _replace_unreadable_database(
+    *, database_path: Path, records: tuple[tuple[str, str, str, bytes], ...]
+) -> None:
+    try:
+        database_path.unlink(missing_ok=True)
+        _insert_records(database_path=database_path, records=records)
+    except (OSError, sqlite3.Error):
+        return
+
+
+def _insert_records(
+    *, database_path: Path, records: tuple[tuple[str, str, str, bytes], ...]
+) -> None:
+    database_path.parent.mkdir(parents=True, exist_ok=True)
+    with (
+        closing(
+            sqlite3.connect(database_path, timeout=FACT_CACHE_SQLITE_TIMEOUT_SECONDS)
+        ) as connection,
+        connection,
+    ):
+        for pragma in FACT_CACHE_WRITE_PRAGMAS:
+            _ = connection.execute(pragma)
+        _ = connection.execute(FACT_CACHE_CREATE_TABLE_SQL)
+        for start in range(0, len(records), FACT_CACHE_INSERT_CHUNK_ROWS):
+            chunk: tuple[tuple[str, str, str, bytes], ...] = records[
+                start : start + FACT_CACHE_INSERT_CHUNK_ROWS
+            ]
+            parameters: list[str | bytes] = []
+            for record in chunk:
+                parameters.extend(record)
+            _ = connection.execute(
+                FACT_CACHE_INSERT_SQL + ",".join([FACT_CACHE_INSERT_ROW_SQL] * len(chunk)),
+                parameters,
+            )
diff --git a/src/sqlbuild/compiler/fact_cache/_helpers/restricted_pickle.py b/src/sqlbuild/compiler/fact_cache/_helpers/restricted_pickle.py
new file mode 100644
index 0000000000..05b7a1bf8d
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/_helpers/restricted_pickle.py
@@ -0,0 +1,50 @@
+"""Pickle codec restricted to SQLBuild value types for local fact cache payloads."""
+
+from __future__ import annotations
+
+import dataclasses
+import enum
+import io
+import pickle
+from typing import override
+
+from sqlbuild.compiler.fact_cache.constants import (
+    FACT_CACHE_ALLOWED_GLOBALS,
+    FACT_CACHE_ALLOWED_PACKAGE,
+    FACT_CACHE_PICKLE_PROTOCOL,
+)
+from sqlbuild.compiler.fact_cache.exceptions import FactCachePayloadError
+
+
+class _RestrictedUnpickler(pickle.Unpickler):
+    """Resolve only immutable-value classes, never functions or arbitrary callables."""
+
+    @override
+    def find_class(self, module: str, name: str) -> type:
+        if (module, name) in FACT_CACHE_ALLOWED_GLOBALS:
+            resolved: object = super().find_class(module, name)
+            if isinstance(resolved, type):
+                return resolved
+        if module == FACT_CACHE_ALLOWED_PACKAGE or module.startswith(
+            f"{FACT_CACHE_ALLOWED_PACKAGE}."
+        ):
+            resolved = super().find_class(module, name)
+            if isinstance(resolved, type) and (
+                dataclasses.is_dataclass(resolved)
+                or issubclass(resolved, enum.Enum)
+                or (issubclass(resolved, tuple) and hasattr(resolved, "_fields"))
+            ):
+                return resolved
+        raise FactCachePayloadError(f"fact cache payload references disallowed {module}.{name}")
+
+
+def dump_fact_payload(value: object) -> bytes:
+    """Serialize one cache value with the fixed fact-cache protocol."""
+
+    return pickle.dumps(value, protocol=FACT_CACHE_PICKLE_PROTOCOL)
+
+
+def load_fact_payload(payload: bytes) -> object:
+    """Deserialize one verified cache payload through the restricted class allowlist."""
+
+    return _RestrictedUnpickler(io.BytesIO(payload)).load()
diff --git a/src/sqlbuild/compiler/fact_cache/classes/__init__.py b/src/sqlbuild/compiler/fact_cache/classes/__init__.py
new file mode 100644
index 0000000000..595a1c6848
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/classes/__init__.py
@@ -0,0 +1 @@
+"""Fact cache classes."""
diff --git a/src/sqlbuild/compiler/fact_cache/classes/fact_cache_store.py b/src/sqlbuild/compiler/fact_cache/classes/fact_cache_store.py
new file mode 100644
index 0000000000..cd0809f5ca
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/classes/fact_cache_store.py
@@ -0,0 +1,189 @@
+"""SQLite-backed content-addressed store for deterministic per-file compile facts."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import pickle
+import sqlite3
+from collections.abc import Sequence
+from contextlib import closing
+from pathlib import Path
+from types import TracebackType
+from typing import Any
+
+from sqlbuild.compiler.fact_cache._helpers.code_identity import installed_code_identity
+from sqlbuild.compiler.fact_cache._helpers.publication import (
+    await_fact_publication,
+    entry_digest,
+    publish_fact_rows,
+)
+from sqlbuild.compiler.fact_cache._helpers.restricted_pickle import (
+    dump_fact_payload,
+    load_fact_payload,
+)
+from sqlbuild.compiler.fact_cache.constants import (
+    FACT_CACHE_DATABASE_SUFFIX,
+    FACT_CACHE_DIGEST_BYTES,
+    FACT_CACHE_DIRECTORY_PREFIX,
+    FACT_CACHE_MAX_ENTRY_BYTES,
+    FACT_CACHE_QUERY_CHUNK_SIZE,
+    FACT_CACHE_SQLITE_TIMEOUT_SECONDS,
+    FACT_CACHE_VERSION,
+)
+from sqlbuild.compiler.profiling.main._metric import record_compile_metric
+
+_READ_ERRORS: tuple[type[BaseException], ...] = (
+    OSError,
+    sqlite3.DatabaseError,
+    pickle.UnpicklingError,
+    AttributeError,
+    EOFError,
+    ImportError,
+    IndexError,
+    KeyError,
+    TypeError,
+    ValueError,
+    RecursionError,
+    MemoryError,
+)
+
+
+class FactCacheStore:
+    """Reuse exact facts keyed by every input and the producing code; faults are misses."""
+
+    def __init__(self, *, root: Path | None, namespace: str, algorithm: str) -> None:
+        self._database_path: Path | None = (
+            None
+            if root is None
+            else root
+            / f"{FACT_CACHE_DIRECTORY_PREFIX}{FACT_CACHE_VERSION}"
+            / f"{namespace}{FACT_CACHE_DATABASE_SUFFIX}"
+        )
+        self._key_prefix: bytes = (
+            f"{FACT_CACHE_VERSION}\0{namespace}\0{algorithm}\0".encode()
+            + (installed_code_identity().encode() if root is not None else b"")
+            + b"\0"
+        )
+        self._key_hasher: Any = hashlib.blake2b(
+            self._key_prefix, digest_size=FACT_CACHE_DIGEST_BYTES
+        )
+        self._pending: dict[str, tuple[str, str, bytes]] = {}
+        self._hits: int = 0
+        self._misses: int = 0
+
+    @property
+    def enabled(self) -> bool:
+        """Return whether this invocation reads and publishes cached facts."""
+
+        return self._database_path is not None
+
+    def __enter__(self) -> FactCacheStore:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        del exc_value, traceback
+        try:
+            if exc_type is None:
+                self._write_pending()
+        finally:
+            self._pending.clear()
+            if self._database_path is not None:
+                record_compile_metric(metric="fact_cache_hits", value=self._hits)
+                record_compile_metric(metric="fact_cache_misses", value=self._misses)
+
+    def key(self, *parts: str | bytes) -> str:
+        """Return the exact identity of one fact computed from the given ordered inputs."""
+
+        digest: Any = self._key_hasher.copy()
+        for part in parts:
+            encoded: bytes = (
+                part if isinstance(part, bytes) else part.encode("utf-8", "surrogatepass")
+            )
+            digest.update(len(encoded).to_bytes(8, "little"))
+            digest.update(encoded)
+        return str(digest.hexdigest())
+
+    def read_many(self, entries: Sequence[tuple[str, str]]) -> dict[str, object]:
+        """Return verified cached values for requested (slot, key) pairs; faults are misses."""
+
+        found: dict[str, object] = {}
+        if self._database_path is None or not entries:
+            return found
+        expected_keys: dict[str, str] = dict(entries)
+        await_fact_publication(database_path=self._database_path)
+        if self._database_path.is_file():
+            try:
+                with closing(
+                    sqlite3.connect(
+                        f"file:{self._database_path}?mode=ro",
+                        uri=True,
+                        timeout=FACT_CACHE_SQLITE_TIMEOUT_SECONDS,
+                    )
+                ) as connection:
+                    slots: tuple[str, ...] = tuple(expected_keys)
+                    for start in range(0, len(slots), FACT_CACHE_QUERY_CHUNK_SIZE):
+                        chunk: tuple[str, ...] = slots[start : start + FACT_CACHE_QUERY_CHUNK_SIZE]
+                        placeholders: str = ",".join("?" for _ in chunk)
+                        rows: list[tuple[object, object, object, object]] = connection.execute(
+                            f"SELECT slot, cache_key, digest, payload FROM fact "
+                            f"WHERE slot IN ({placeholders})",
+                            chunk,
+                        ).fetchall()
+                        for slot, cache_key, digest, payload in rows:
+                            if not isinstance(slot, str) or expected_keys.get(slot) != cache_key:
+                                continue
+                            value: object | None = _verified_value(
+                                cache_key=cache_key, digest=digest, payload=payload
+                            )
+                            if value is not None:
+                                found[expected_keys[slot]] = value
+            except _READ_ERRORS:
+                found = {}
+        self._hits += sum(1 for key in expected_keys.values() if key in found)
+        self._misses += sum(1 for key in expected_keys.values() if key not in found)
+        return found
+
+    def stage(self, *, key: str, slot: str, value: object) -> None:
+        """Queue one freshly computed fact for publication after a successful invocation."""
+
+        if self._database_path is None:
+            return
+        try:
+            payload: bytes = dump_fact_payload(value)
+        except (pickle.PicklingError, TypeError, AttributeError, RecursionError, ValueError):
+            return
+        if len(payload) > FACT_CACHE_MAX_ENTRY_BYTES:
+            return
+        self._pending[slot] = (key, entry_digest(cache_key=key, payload=payload), payload)
+
+    def _write_pending(self) -> None:
+        if self._database_path is None or not self._pending:
+            return
+        publish_fact_rows(
+            database_path=self._database_path,
+            rows=tuple(
+                (slot, key, digest, payload)
+                for slot, (key, digest, payload) in self._pending.items()
+            ),
+        )
+
+
+def _verified_value(*, cache_key: object, digest: object, payload: object) -> object | None:
+    if (
+        not isinstance(cache_key, str)
+        or not isinstance(digest, str)
+        or not isinstance(payload, bytes)
+        or len(payload) > FACT_CACHE_MAX_ENTRY_BYTES
+        or not hmac.compare_digest(digest, entry_digest(cache_key=cache_key, payload=payload))
+    ):
+        return None
+    try:
+        return load_fact_payload(payload)
+    except _READ_ERRORS:
+        return None
diff --git a/src/sqlbuild/compiler/fact_cache/classes/fact_publication_registry.py b/src/sqlbuild/compiler/fact_cache/classes/fact_publication_registry.py
new file mode 100644
index 0000000000..13824c2e4a
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/classes/fact_publication_registry.py
@@ -0,0 +1,44 @@
+"""Process-wide ordering of background fact publications per cache database."""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from pathlib import Path
+
+
+class FactPublicationRegistry:
+    """Serialize publications per database and let readers wait for pending writes."""
+
+    def __init__(self) -> None:
+        self._lock: threading.Lock = threading.Lock()
+        self._writers: dict[Path, threading.Thread] = {}
+
+    def start(
+        self,
+        *,
+        database_path: Path,
+        write: Callable[[threading.Thread | None], None],
+        thread_name: str,
+    ) -> None:
+        """Start one writer that first waits for the previous writer of the same database."""
+
+        with self._lock:
+            previous: threading.Thread | None = self._writers.get(database_path)
+            writer: threading.Thread = threading.Thread(
+                target=write, args=(previous,), name=thread_name
+            )
+            self._writers[database_path] = writer
+            writer.start()
+
+    def wait(self, *, database_path: Path | None) -> None:
+        """Wait for the latest writer of one database, or of every database when None."""
+
+        with self._lock:
+            writers: tuple[threading.Thread, ...] = tuple(
+                writer
+                for path, writer in self._writers.items()
+                if database_path is None or path == database_path
+            )
+        for writer in writers:
+            writer.join()
diff --git a/src/sqlbuild/compiler/fact_cache/constants.py b/src/sqlbuild/compiler/fact_cache/constants.py
new file mode 100644
index 0000000000..4bcf1840f9
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/constants.py
@@ -0,0 +1,61 @@
+"""Stable constants for the local compile fact cache."""
+
+from __future__ import annotations
+
+FACT_CACHE_VERSION: int = 1
+FACT_CACHE_DIRECTORY_PREFIX: str = "facts-v"
+FACT_CACHE_DATABASE_SUFFIX: str = ".sqlite3"
+FACT_CACHE_DIGEST_BYTES: int = 32
+FACT_CACHE_MAX_ENTRY_BYTES: int = 16 * 1024 * 1024
+FACT_CACHE_SQLITE_TIMEOUT_SECONDS: float = 0.1
+FACT_CACHE_QUERY_CHUNK_SIZE: int = 500
+FACT_CACHE_PICKLE_PROTOCOL: int = 5
+FACT_CACHE_CREATE_TABLE_SQL: str = """
+CREATE TABLE IF NOT EXISTS fact (
+    slot TEXT PRIMARY KEY,
+    cache_key TEXT NOT NULL,
+    digest TEXT NOT NULL,
+    payload BLOB NOT NULL
+)
+"""
+FACT_CACHE_INSERT_SQL: str = (
+    "INSERT OR REPLACE INTO fact (slot, cache_key, digest, payload) VALUES "
+)
+FACT_CACHE_INSERT_ROW_SQL: str = "(?, ?, ?, ?)"
+FACT_CACHE_PARAMETERS_PER_ROW: int = 4
+SQLITE_PRE_3_32_MAX_BOUND_PARAMETERS: int = 999
+FACT_CACHE_INSERT_CHUNK_ROWS: int = (
+    SQLITE_PRE_3_32_MAX_BOUND_PARAMETERS // FACT_CACHE_PARAMETERS_PER_ROW
+)
+FACT_CACHE_WRITE_PRAGMAS: tuple[str, ...] = (
+    "PRAGMA synchronous = OFF",
+    "PRAGMA journal_mode = MEMORY",
+)
+FACT_CACHE_WRITER_THREAD_NAME: str = "sqlbuild-fact-cache-writer"
+FACT_CACHE_ALLOWED_PACKAGE: str = "sqlbuild"
+FACT_CACHE_ALLOWED_GLOBALS: frozenset[tuple[str, str]] = frozenset(
+    {
+        ("pathlib", "Path"),
+        ("pathlib", "PosixPath"),
+        ("pathlib", "PurePath"),
+        ("pathlib", "PurePosixPath"),
+        ("pathlib", "PureWindowsPath"),
+        ("pathlib", "WindowsPath"),
+        ("pathlib._local", "Path"),
+        ("pathlib._local", "PosixPath"),
+        ("pathlib._local", "PurePath"),
+        ("pathlib._local", "PurePosixPath"),
+        ("pathlib._local", "PureWindowsPath"),
+        ("pathlib._local", "WindowsPath"),
+        ("decimal", "Decimal"),
+        ("datetime", "date"),
+        ("datetime", "datetime"),
+        ("datetime", "time"),
+        ("datetime", "timedelta"),
+        ("datetime", "timezone"),
+        ("builtins", "frozenset"),
+        ("builtins", "set"),
+        ("builtins", "complex"),
+        ("collections", "OrderedDict"),
+    }
+)
diff --git a/src/sqlbuild/compiler/fact_cache/exceptions.py b/src/sqlbuild/compiler/fact_cache/exceptions.py
new file mode 100644
index 0000000000..37bf4d202f
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/exceptions.py
@@ -0,0 +1,7 @@
+"""Fact cache exceptions."""
+
+from __future__ import annotations
+
+
+class FactCachePayloadError(ValueError):
+    """Raised when a stored fact payload is not a trusted value graph."""
diff --git a/src/sqlbuild/compiler/fact_cache/main/__init__.py b/src/sqlbuild/compiler/fact_cache/main/__init__.py
new file mode 100644
index 0000000000..56c61b674b
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/main/__init__.py
@@ -0,0 +1 @@
+"""Fact cache public entry modules."""
diff --git a/src/sqlbuild/compiler/fact_cache/main/_compile_cache_root.py b/src/sqlbuild/compiler/fact_cache/main/_compile_cache_root.py
new file mode 100644
index 0000000000..39067cf45e
--- /dev/null
+++ b/src/sqlbuild/compiler/fact_cache/main/_compile_cache_root.py
@@ -0,0 +1,27 @@
+"""Project-local compile-cache root resolution."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from sqlbuild.compiler.compile.constants import (
+    COMPILE_CACHE_DISABLE_ENV_VAR,
+    COMPILE_CACHE_DISABLE_VALUE,
+)
+from sqlbuild.spec.contracts.models import TargetConfig
+
+
+def compile_cache_root(
+    *, project_dir: Path | None, target_config: TargetConfig | None, no_cache: bool
+) -> Path | None:
+    """Return the shared compile-cache directory, or None when caching is disabled."""
+
+    if (
+        no_cache
+        or project_dir is None
+        or (target_config is not None and target_config.compile_cache is False)
+        or os.environ.get(COMPILE_CACHE_DISABLE_ENV_VAR) == COMPILE_CACHE_DISABLE_VALUE
+    ):
+        return None
+    return project_dir / "target" / "cache" / "compiler"
diff --git a/src/sqlbuild/compiler/profiling/constants.py b/src/sqlbuild/compiler/profiling/constants.py
index 2794341f76..6ca434213c 100644
--- a/src/sqlbuild/compiler/profiling/constants.py
+++ b/src/sqlbuild/compiler/profiling/constants.py
@@ -22,4 +22,6 @@
     "analysis_entry_cache_hits",
     "analysis_cache_misses",
     "analysis_cache_bypasses",
+    "fact_cache_hits",
+    "fact_cache_misses",
 )
diff --git a/src/sqlbuild/compiler/profiling/types.py b/src/sqlbuild/compiler/profiling/types.py
index 64d8b0dd73..abb9028ae4 100644
--- a/src/sqlbuild/compiler/profiling/types.py
+++ b/src/sqlbuild/compiler/profiling/types.py
@@ -22,4 +22,6 @@
     "analysis_entry_cache_hits",
     "analysis_cache_misses",
     "analysis_cache_bypasses",
+    "fact_cache_hits",
+    "fact_cache_misses",
 ]
diff --git a/src/sqlbuild/rule_engine/_helpers/run/rules.py b/src/sqlbuild/rule_engine/_helpers/run/rules.py
index da51274567..c42aa7120d 100644
--- a/src/sqlbuild/rule_engine/_helpers/run/rules.py
+++ b/src/sqlbuild/rule_engine/_helpers/run/rules.py
@@ -61,6 +61,12 @@
 _SQL_RULE_SUPPRESSION_CODE: str = "SQBRSQL000"
 
 
+@dataclass(frozen=True)
+class _SqlRuleInputs:
+    identities: dict[str, str]
+    project_files: dict[Path, str] | None
+
+
 @dataclass(frozen=True)
 class _SqlRulesEvaluation:
     findings: tuple[Finding, ...]
@@ -118,6 +124,13 @@ def evaluate_rules(
         select=tuple(rule.code for rule in selected),
         ignore=(),
     )
+    sql_rule_inputs: _SqlRuleInputs = _prepare_sql_rule_inputs(
+        rules=native_rules,
+        project=selected_project,
+        dialect=dialect,
+        project_dir=resolved_project_dir,
+        collect_files=prepared_sql is None and model_paths is None,
+    )
     with ThreadPoolExecutor(max_workers=1, thread_name_prefix="sqlbuild-rules") as executor:
         native_result: Future[RulesResult] = executor.submit(
             evaluate_native,
@@ -140,6 +153,7 @@ def evaluate_rules(
             cache_enabled=effective_config.cache.enabled,
             prepared_sql=prepared_sql,
             expansion_reuse=expansion_reuse,
+            prepared_inputs=sql_rule_inputs,
         )
         sql_ms: int = round((time.monotonic() - sql_started) * 1000)
         result: RulesResult = native_result.result()
@@ -174,6 +188,29 @@ def evaluate_rules(
     )
 
 
+def _prepare_sql_rule_inputs(
+    *,
+    rules: tuple[Rule, ...],
+    project: CompiledProject,
+    dialect: str,
+    project_dir: Path,
+    collect_files: bool,
+) -> _SqlRuleInputs:
+    """Hash SQL and read files first so GIL-releasing work never waits on the native thread."""
+
+    identities: dict[str, str] = _sql_model_rule_identities(
+        rules=rules, project=project, dialect=dialect
+    )
+    return _SqlRuleInputs(
+        identities=identities,
+        project_files=(
+            collect_project_files(project_dir=project_dir, selected_paths=None)
+            if identities and collect_files
+            else None
+        ),
+    )
+
+
 def _config_references_custom_rules(config: RulesConfig) -> bool:
     ignore_selectors: list[str] = []
     for entry in config.rule_ignores:
@@ -374,22 +411,20 @@ def _run_sql_rules(
     cache_enabled: bool,
     prepared_sql: PreparedSqlLint | None = None,
     expansion_reuse: SqlExpansionReuse | None = None,
+    prepared_inputs: _SqlRuleInputs | None = None,
 ) -> _SqlRulesEvaluation:
-    codes: tuple[str, ...] = tuple(rule.code for rule in rules if rule.code.startswith("SQBRSQL"))
+    codes: tuple[str, ...] = _sql_rule_codes(rules)
     if not codes:
         return _SqlRulesEvaluation(findings=(), cache_hits=0, cache_misses=0)
-    models_by_path: dict[str, CompiledModel] = {
-        model.relative_path.as_posix(): model
-        for model in project.models
-        if project.settings.sql_analysis and model.config.values.get("sql_analysis") is not False
-    }
+    models_by_path: dict[str, CompiledModel] = _sql_rule_models(project)
     bucket: dict[str, dict[str, object]] = (
         _read_sql_rule_cache(project_dir) if cache_enabled else {}
     )
-    identities: dict[str, str] = {
-        path: _sql_rule_identity(model=model, codes=codes, dialect=dialect)
-        for path, model in models_by_path.items()
-    }
+    identities: dict[str, str] = (
+        prepared_inputs.identities
+        if prepared_inputs is not None
+        else _sql_model_rule_identities(rules=rules, project=project, dialect=dialect)
+    )
     findings: list[Finding] = []
     misses: set[str] = set()
     for path, identity in identities.items():
@@ -411,7 +446,11 @@ def _run_sql_rules(
             prepared_sql=prepared_sql, project_dir=project_dir, codes=codes
         )
         if project_files is None:
-            project_files = collect_project_files(project_dir=project_dir, selected_paths=None)
+            project_files = (
+                prepared_inputs.project_files
+                if prepared_inputs is not None and prepared_inputs.project_files is not None
+                else collect_project_files(project_dir=project_dir, selected_paths=None)
+            )
         file_path: Path
         contents: str
         for file_path, contents in project_files.items():
@@ -533,6 +572,30 @@ def _lint_finding(*, violation: LintViolation, project_dir: Path) -> Finding:
     )
 
 
+def _sql_rule_codes(rules: tuple[Rule, ...]) -> tuple[str, ...]:
+    return tuple(rule.code for rule in rules if rule.code.startswith("SQBRSQL"))
+
+
+def _sql_rule_models(project: CompiledProject) -> dict[str, CompiledModel]:
+    return {
+        model.relative_path.as_posix(): model
+        for model in project.models
+        if project.settings.sql_analysis and model.config.values.get("sql_analysis") is not False
+    }
+
+
+def _sql_model_rule_identities(
+    *, rules: tuple[Rule, ...], project: CompiledProject, dialect: str
+) -> dict[str, str]:
+    codes: tuple[str, ...] = _sql_rule_codes(rules)
+    if not codes:
+        return {}
+    return {
+        path: _sql_rule_identity(model=model, codes=codes, dialect=dialect)
+        for path, model in _sql_rule_models(project).items()
+    }
+
+
 def _sql_rule_identity(*, model: CompiledModel, codes: tuple[str, ...], dialect: str) -> str:
     digest: Any = hashlib.sha256()
     digest.update(_SQL_RULE_CACHE_VERSION.encode())
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py
index 1e2d4724ac..736251b3b9 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py
@@ -86,6 +86,8 @@ def test_given_waffle_shop_when_running_compile_json_then_it_reports_offline_que
         "analysis_entry_cache_hits",
         "analysis_cache_misses",
         "analysis_cache_bypasses",
+        "fact_cache_hits",
+        "fact_cache_misses",
         "total_ms",
     }
     assert all(isinstance(value, int) and value >= 0 for value in timings.values())
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py
index 6f18ac6227..79487b7949 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_runtime_artifact_preservation.py
@@ -32,7 +32,7 @@
                 "target/run/functions/python/is_completed_order_py.sql",
             ),
             expected_exit_code=0,
-            expected_compiler_cache_count=3,
+            expected_compiler_cache_count=4,
             expected_local_history=False,
             expected_compiled_paths=(
                 "target/compiled/models/marts/fact_orders.sql",
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 dcf8d2f9ef..bd05de011b 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
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
 
 
 @dataclass(frozen=True)
@@ -345,3 +347,27 @@ class ExistingStatePlanGuardTestCase:
     expected_migrations: tuple[tuple[str, str, str], ...]
     expected_max_wall_seconds: float
     expected_max_rss_bytes: int
+
+
+@dataclass(frozen=True)
+class CompileCacheInvalidationTestCase:
+    """One authored input edit applied after a warm compile-cache run."""
+
+    description: str
+    edit: Callable[[Path], None]
+    edited_env: dict[str, str] = field(default_factory=dict)
+    expected_failure: bool = False
+    expected_output_change: bool = True
+    expected_rewarmed_fact_cache_misses: int = 0
+
+
+@dataclass(frozen=True)
+class CompileCacheDisabledTestCase:
+    """One supported control that disables compile-cache reads and writes."""
+
+    description: str
+    compile_args: tuple[str, ...] = ()
+    env: dict[str, str] = field(default_factory=dict)
+    edit: Callable[[Path], None] = lambda _root: None
+    expected_fact_cache_counts: tuple[int, int] = (0, 0)
+    expected_fact_databases: tuple[Path, ...] = ()
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py
index bc2754dd34..2679e022f6 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py
@@ -12,6 +12,7 @@
     BuiltBenchmark,
     InspectionCommandMeasurement,
     prepare_build_benchmark_project,
+    prepare_compile_cache_invalidation_project,
     prepare_inspection_benchmark_project,
     run_fresh_process_inspection_command,
 )
@@ -33,6 +34,13 @@ def semantic_playground(tmp_path_factory: pytest.TempPathFactory) -> Path:
     return project
 
 
+@pytest.fixture
+def cache_invalidation_project(tmp_path: Path) -> Path:
+    project_dir: Path = tmp_path / "orders_project"
+    prepare_compile_cache_invalidation_project(project_dir=project_dir)
+    return project_dir
+
+
 @pytest.fixture(scope="module")
 def inspection_benchmark_project(tmp_path_factory: pytest.TempPathFactory) -> Path:
     assert_required_cgroup_memory_limit()
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 8faed82407..6d30815bdf 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py
@@ -2334,3 +2334,156 @@ def write_relation_stub_project(
             ),
         },
     )
+
+
+COMPILE_CACHE_REGION_ENV_VAR: str = "SQB_CACHE_INVALIDATION_REGION"
+_COMPILE_CACHE_DIAGNOSTIC_PATTERN: re.Pattern[str] = re.compile(
+    r"^(?:error|warning)\[.*$", re.MULTILINE
+)
+_COMPILE_CACHE_EXTRA_PROJECT_FILES: dict[str, str] = {
+    "models/marts/_sqlbuild/_enums/order_channel.sql": (
+        'ENUM (\n  name order_channel,\n  members (WEB "web", PARTNER "partner"),\n);\n'
+    ),
+    "models/marts/_sqlbuild/_constants/min_quantity.sql": (
+        "CONSTANT (name min_quantity, value 1);\n"
+    ),
+    "models/marts/channel_orders.sql": (
+        "MODEL (\n"
+        "  materialized table,\n"
+        "  columns (\n"
+        "    order_id (nullable false, audits [not_null]),\n"
+        "  ),\n"
+        ");\n\n"
+        "SELECT\n"
+        "  o.order_id,\n"
+        '  @enum("order_channel").WEB AS order_channel,\n'
+        '  o.quantity >= @const("min_quantity") AS meets_minimum,\n'
+        "  CAST(@@quantity_multiplier AS INTEGER) * o.quantity AS scaled_quantity,\n"
+        "  '@@ENV:" + COMPILE_CACHE_REGION_ENV_VAR + "' AS region\n"
+        'FROM __ref("stg_orders") o\n'
+    ),
+    "tests/unit/test_channel_orders.sql": (
+        "TEST ();\n\n"
+        "WITH\n"
+        "__ref__stg_orders AS (\n"
+        "  SELECT 1 AS order_id, 2 AS quantity\n"
+        "),\n"
+        "__expected__channel_orders AS (\n"
+        "  SELECT 1 AS order_id, 'web' AS order_channel, TRUE AS meets_minimum\n"
+        ")\n"
+        "SELECT 1\n"
+    ),
+    "tests/unit/test_channel_blocks.sql": (
+        'TEST (name "channel_orders_first");\n\n'
+        "WITH\n"
+        "__ref__stg_orders AS (\n"
+        "  SELECT 1 AS order_id, 2 AS quantity\n"
+        "),\n"
+        "__expected__channel_orders AS (\n"
+        "  SELECT 1 AS order_id, 'web' AS order_channel\n"
+        ")\n"
+        "SELECT 1\n\n"
+        'TEST (name "channel_orders_second");\n\n'
+        "WITH\n"
+        "__ref__stg_orders AS (\n"
+        "  SELECT 2 AS order_id, 5 AS quantity\n"
+        "),\n"
+        "__expected__channel_orders AS (\n"
+        "  SELECT 2 AS order_id, 'web' AS order_channel\n"
+        ")\n"
+        "SELECT 1\n"
+    ),
+}
+
+
+class CompileCacheOutcome(NamedTuple):
+    """Comparable result of one fresh-process compile."""
+
+    returncode: int
+    diagnostics: tuple[str, ...]
+    fingerprint: str
+    fact_cache_hits: int
+    fact_cache_misses: int
+
+
+def run_installed_sqb(
+    *, project_dir: Path, args: tuple[str, ...], env: dict[str, str]
+) -> subprocess.CompletedProcess[str]:
+    """Run the installed sqb entrypoint in a fresh process with extra environment values."""
+
+    return subprocess.run(
+        [
+            str(Path(sys.executable).with_name("sqb")),
+            "--project-dir",
+            str(project_dir),
+            "--no-color",
+            *args,
+        ],
+        capture_output=True,
+        text=True,
+        env={**os.environ, **env},
+        check=False,
+    )
+
+
+def compile_cache_outcome(
+    *, project_dir: Path, env: dict[str, str], compile_args: tuple[str, ...] = ()
+) -> CompileCacheOutcome:
+    """Compile in a fresh process and return its semantic fingerprint and fact-cache counts."""
+
+    result: subprocess.CompletedProcess[str] = run_installed_sqb(
+        project_dir=project_dir, args=("compile", "--json", *compile_args), env=env
+    )
+    values: dict[str, object] = cast(dict[str, object], json.loads(result.stdout or "{}"))
+    timings: dict[str, int] = cast(dict[str, int], values.get("compile_timings", {}))
+    return CompileCacheOutcome(
+        returncode=result.returncode,
+        diagnostics=tuple(_COMPILE_CACHE_DIAGNOSTIC_PATTERN.findall(result.stderr)),
+        fingerprint=semantic_compile_fingerprint(
+            payload=values, compiled_dir=project_dir / "target" / "compiled"
+        ),
+        fact_cache_hits=timings.get("fact_cache_hits", 0),
+        fact_cache_misses=timings.get("fact_cache_misses", 0),
+    )
+
+
+def replace_project_text(project_dir: Path, relative_path: str, old: str, new: str) -> None:
+    """Replace the first occurrence of authored text, failing when it is absent."""
+
+    path: Path = project_dir / relative_path
+    contents: str = path.read_text(encoding="utf-8")
+    assert old in contents, f"{old!r} not found in {relative_path}"
+    path.write_text(contents.replace(old, new, 1), encoding="utf-8")
+
+
+def write_project_file(project_dir: Path, relative_path: str, contents: str) -> None:
+    """Write one authored project file, creating parent folders."""
+
+    path: Path = project_dir / relative_path
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(contents, encoding="utf-8")
+
+
+def move_project_file(project_dir: Path, source: str, destination: str) -> None:
+    """Move one authored project file into a possibly new folder."""
+
+    (project_dir / destination).parent.mkdir(parents=True, exist_ok=True)
+    (project_dir / source).rename(project_dir / destination)
+
+
+def prepare_compile_cache_invalidation_project(*, project_dir: Path) -> None:
+    """Create a playground project covering every cached compile input kind."""
+
+    result: subprocess.CompletedProcess[str] = run_installed_sqb(
+        project_dir=project_dir.parent, args=("playground", str(project_dir)), env={}
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    shutil.rmtree(project_dir / "target", ignore_errors=True)
+    for relative_path, contents in _COMPILE_CACHE_EXTRA_PROJECT_FILES.items():
+        write_project_file(project_dir, relative_path, contents)
+    replace_project_text(
+        project_dir,
+        "sqlbuild_project.toml",
+        "[settings]",
+        '[vars]\nquantity_multiplier = "2"\n\n[settings]',
+    )
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_cache_invalidation.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_cache_invalidation.py
new file mode 100644
index 0000000000..0328b31d8f
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_cache_invalidation.py
@@ -0,0 +1,268 @@
+"""Warm compile-cache reuse must match a cache-disabled compile after every input edit."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
+    CompileCacheDisabledTestCase,
+    CompileCacheInvalidationTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
+    COMPILE_CACHE_REGION_ENV_VAR,
+    CompileCacheOutcome,
+    compile_cache_outcome,
+    move_project_file,
+    replace_project_text,
+    run_installed_sqb,
+    write_project_file,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CompileCacheInvalidationTestCase(
+            description="model_sql_body_edit",
+            edit=lambda root: replace_project_text(
+                root, "models/staging/stg_orders.sql", "  quantity,", "  quantity + 0 AS quantity,"
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="model_header_metadata_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "models/staging/stg_orders.sql",
+                "customer_id (nullable false, audits [not_null]),",
+                'customer_id (nullable false, description "Buyer.", audits [not_null, unique]),',
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="model_header_syntax_error",
+            edit=lambda root: replace_project_text(
+                root, "models/staging/stg_orders.sql", "materialized view,", "materialized view,,(("
+            ),
+            expected_failure=True,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="model_added",
+            edit=lambda root: write_project_file(
+                root,
+                "models/marts/order_quantities.sql",
+                'MODEL ();\n\nSELECT order_id, quantity FROM __ref("stg_orders")\n',
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="model_deleted",
+            edit=lambda root: (root / "models/marts/channel_orders.sql").unlink(),
+            expected_failure=True,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="model_moved_between_folders",
+            edit=lambda root: move_project_file(
+                root, "models/staging/stg_payments.sql", "models/staging/payments/stg_payments.sql"
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="source_yaml_column_type_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "sources/raw.yml",
+                "      - name: quantity\n        type: INTEGER",
+                "      - name: quantity\n        type: BIGINT",
+            ),
+            expected_output_change=False,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="source_yaml_column_rename",
+            edit=lambda root: replace_project_text(
+                root,
+                "sources/raw.yml",
+                "      - name: quantity\n        type: INTEGER",
+                "      - name: item_count\n        type: INTEGER",
+            ),
+            expected_failure=True,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="source_yaml_syntax_error",
+            edit=lambda root: replace_project_text(
+                root, "sources/raw.yml", "sources:", "sources: [\n"
+            ),
+            expected_failure=True,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="unit_test_expected_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "tests/unit/test_channel_orders.sql",
+                "TRUE AS meets_minimum",
+                "FALSE AS meets_minimum",
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="second_test_block_in_multi_block_file_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "tests/unit/test_channel_blocks.sql",
+                "SELECT 2 AS order_id, 'web' AS order_channel",
+                "SELECT 2 AS order_id, 'partner' AS order_channel",
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="unit_test_structure_error",
+            edit=lambda root: replace_project_text(
+                root, "tests/unit/test_channel_orders.sql", "SELECT 1\n", ""
+            ),
+            expected_failure=True,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="project_macro_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "macros/currency.py",
+                'return f"{price_cents} * {quantity}"',
+                'return f"({price_cents}) * ({quantity})"',
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="scoped_macro_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "models/marts/_sqlbuild/_macros/currency.py",
+                'return f"ROUND(({column}) / 100.0, 2)"',
+                'return f"ROUND(({column}) / 100.0, 3)"',
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="enum_member_value_edit",
+            edit=lambda root: replace_project_text(
+                root, "models/marts/_sqlbuild/_enums/order_channel.sql", 'WEB "web"', 'WEB "online"'
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="constant_value_edit",
+            edit=lambda root: replace_project_text(
+                root, "models/marts/_sqlbuild/_constants/min_quantity.sql", "value 1", "value 3"
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="project_var_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "sqlbuild_project.toml",
+                'quantity_multiplier = "2"',
+                'quantity_multiplier = "5"',
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="target_schema_edit",
+            edit=lambda root: replace_project_text(
+                root, "sqlbuild_project.toml", 'schema = "dev"', 'schema = "dev_next"'
+            ),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="local_config_target_switch",
+            edit=lambda root: write_project_file(root, "sqlbuild_local.toml", 'target = "prod"\n'),
+        ),
+        CompileCacheInvalidationTestCase(
+            description="python_loader_edit",
+            edit=lambda root: replace_project_text(
+                root,
+                "python/loaders/waffle_sources.py",
+                "from sqlbuild.loaders import loader\n",
+                "from sqlbuild.loaders import loader\n\nLOADER_REVISION: int = 2\n",
+            ),
+            expected_output_change=False,
+        ),
+        CompileCacheInvalidationTestCase(
+            description="environment_variable_edit",
+            edit=lambda root: None,
+            edited_env={COMPILE_CACHE_REGION_ENV_VAR: "west"},
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_warm_compile_cache_when_input_changes_then_output_matches_cache_disabled_compile(
+    cache_invalidation_project: Path, test_case: CompileCacheInvalidationTestCase
+) -> None:
+    project_dir: Path = cache_invalidation_project
+    initial_env: dict[str, str] = {COMPILE_CACHE_REGION_ENV_VAR: "east"}
+    cold: CompileCacheOutcome = compile_cache_outcome(project_dir=project_dir, env=initial_env)
+    warm: CompileCacheOutcome = compile_cache_outcome(project_dir=project_dir, env=initial_env)
+    assert cold.returncode == 0
+    assert cold.fact_cache_misses > 0
+    assert warm == CompileCacheOutcome(
+        returncode=0,
+        diagnostics=cold.diagnostics,
+        fingerprint=cold.fingerprint,
+        fact_cache_hits=cold.fact_cache_misses,
+        fact_cache_misses=0,
+    )
+
+    test_case.edit(project_dir)
+    edited_env: dict[str, str] = {**initial_env, **test_case.edited_env}
+    edited: CompileCacheOutcome = compile_cache_outcome(project_dir=project_dir, env=edited_env)
+    reference: CompileCacheOutcome = compile_cache_outcome(
+        project_dir=project_dir, env=edited_env, compile_args=("--no-cache",)
+    )
+    rewarmed: CompileCacheOutcome = compile_cache_outcome(project_dir=project_dir, env=edited_env)
+
+    assert (edited.returncode != 0) is test_case.expected_failure
+    assert (
+        (edited.diagnostics, edited.fingerprint) != (cold.diagnostics, cold.fingerprint)
+    ) is test_case.expected_output_change
+    assert edited[:3] == reference[:3]
+    assert rewarmed[:3] == reference[:3]
+    assert rewarmed.fact_cache_misses == test_case.expected_rewarmed_fact_cache_misses
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        CompileCacheDisabledTestCase(description="no_cache_flag", compile_args=("--no-cache",)),
+        CompileCacheDisabledTestCase(
+            description="disable_environment_variable",
+            env={"SQLBUILD_DISABLE_COMPILE_CACHE": "1"},
+        ),
+        CompileCacheDisabledTestCase(
+            description="target_compile_cache_false",
+            edit=lambda root: replace_project_text(
+                root,
+                "sqlbuild_project.toml",
+                'schema = "dev"',
+                'schema = "dev"\ncompile_cache = false',
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_disabled_compile_cache_when_compiling_then_no_facts_are_read_or_written(
+    cache_invalidation_project: Path, test_case: CompileCacheDisabledTestCase
+) -> None:
+    project_dir: Path = cache_invalidation_project
+    test_case.edit(project_dir)
+    env: dict[str, str] = {COMPILE_CACHE_REGION_ENV_VAR: "east", **test_case.env}
+
+    first: subprocess.CompletedProcess[str] = run_installed_sqb(
+        project_dir=project_dir, args=("compile", "--json", *test_case.compile_args), env=env
+    )
+    second: subprocess.CompletedProcess[str] = run_installed_sqb(
+        project_dir=project_dir, args=("compile", "--json", *test_case.compile_args), env=env
+    )
+
+    assert (first.returncode, second.returncode) == (0, 0), first.stderr + second.stderr
+    first_timings: dict[str, int] = json.loads(first.stdout)["compile_timings"]
+    second_timings: dict[str, int] = json.loads(second.stdout)["compile_timings"]
+    assert (
+        first_timings["fact_cache_hits"],
+        first_timings["fact_cache_misses"],
+    ) == test_case.expected_fact_cache_counts
+    assert (
+        second_timings["fact_cache_hits"],
+        second_timings["fact_cache_misses"],
+    ) == test_case.expected_fact_cache_counts
+    assert tuple((project_dir / "target").rglob("facts-v*")) == test_case.expected_fact_databases
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
index 2f984de28d..82985c1d60 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py
@@ -19,7 +19,9 @@
 from sqlbuild.cli.commands.main.entrypoint.entry import main
 from sqlbuild.compiler.compile._helpers.assembly import source_bindings as source_bindings_module
 from sqlbuild.compiler.compile.models import PolyglotAnalysisResult
-from sqlbuild.compiler.discovery._helpers.filesystem import core as discovery_core_module
+from sqlbuild.compiler.discovery._helpers.filesystem import (
+    model_files as discovery_model_files_module,
+)
 from sqlbuild.spec.contracts.models import SourceLocation
 
 
@@ -449,12 +451,14 @@ def record_eager_output_column_scans(monkeypatch: pytest.MonkeyPatch) -> list[Pa
 
     scanned: list[Path] = []
     original: Callable[..., dict[str, SourceLocation]] = (
-        discovery_core_module.model_output_column_locations
+        discovery_model_files_module.matched_model_output_column_locations
     )
 
     def recording(**kwargs: Any) -> dict[str, SourceLocation]:
         scanned.append(cast(Path, kwargs["relative_path"]))
         return original(**kwargs)
 
-    monkeypatch.setattr(discovery_core_module, "model_output_column_locations", recording)
+    monkeypatch.setattr(
+        discovery_model_files_module, "matched_model_output_column_locations", recording
+    )
     return scanned
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py b/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py
index ebdc72c8ec..6a18174026 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py
@@ -732,10 +732,17 @@ def test_given_warm_scoped_macro_sql_test_when_compiling_then_rules_reuse_compil
     )
 
     def counting_build(
-        *, discovered_inputs: DiscoveredProjectInputs, loaded_macros: dict[str, LoadedMacro]
+        *,
+        discovered_inputs: DiscoveredProjectInputs,
+        loaded_macros: dict[str, LoadedMacro],
+        compile_cache_dir: Path | None = None,
     ) -> DeclarationScopeBuild:
         scope_builds.append(tmp_path)
-        return original_build(discovered_inputs=discovered_inputs, loaded_macros=loaded_macros)
+        return original_build(
+            discovered_inputs=discovered_inputs,
+            loaded_macros=loaded_macros,
+            compile_cache_dir=compile_cache_dir,
+        )
 
     monkeypatch.setattr(compile_inputs_module, "build_declaration_scope", counting_build)
     monkeypatch.setattr(expansion_context_module, "build_declaration_scope", counting_build)
@@ -785,10 +792,17 @@ def test_given_cold_scoped_macro_sql_test_when_compiling_then_early_lint_uses_pr
     )
 
     def counting_build(
-        *, discovered_inputs: DiscoveredProjectInputs, loaded_macros: dict[str, LoadedMacro]
+        *,
+        discovered_inputs: DiscoveredProjectInputs,
+        loaded_macros: dict[str, LoadedMacro],
+        compile_cache_dir: Path | None = None,
     ) -> DeclarationScopeBuild:
         scope_builds.append(tmp_path)
-        return original_build(discovered_inputs=discovered_inputs, loaded_macros=loaded_macros)
+        return original_build(
+            discovered_inputs=discovered_inputs,
+            loaded_macros=loaded_macros,
+            compile_cache_dir=compile_cache_dir,
+        )
 
     def recording_load(macro_files: tuple[DiscoveredMacroFile, ...]) -> dict[str, LoadedMacro]:
         loaded: dict[str, LoadedMacro] = original_load(macro_files)
diff --git a/tests/unit/src/sqlbuild/compiler/fact_cache/__init__.py b/tests/unit/src/sqlbuild/compiler/fact_cache/__init__.py
new file mode 100644
index 0000000000..89bbeb8c34
--- /dev/null
+++ b/tests/unit/src/sqlbuild/compiler/fact_cache/__init__.py
@@ -0,0 +1 @@
+"""Fact cache tests."""
diff --git a/tests/unit/src/sqlbuild/compiler/fact_cache/classes/__init__.py b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/__init__.py
new file mode 100644
index 0000000000..611c1f64d4
--- /dev/null
+++ b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/__init__.py
@@ -0,0 +1 @@
+"""Fact cache class tests."""
diff --git a/tests/unit/src/sqlbuild/compiler/fact_cache/classes/_test_types.py b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/_test_types.py
new file mode 100644
index 0000000000..78a760c91b
--- /dev/null
+++ b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/_test_types.py
@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class FactCacheReadTestCase:
+    """One read of a fact published for the ("orders", "SELECT 1") inputs."""
+
+    description: str
+    read_key_parts: tuple[str, ...]
+    read_algorithm: str
+    expected_hit: bool
+    expected_metrics: tuple[int, int]
+
+
+@dataclass(frozen=True)
+class FactCacheCodeIdentityTestCase:
+    """One change to the installed code that produced a published fact."""
+
+    description: str
+    next_code_identity: str
+    expected_found_count: int
+
+
+@dataclass(frozen=True)
+class FactCacheCorruptionTestCase:
+    """One way a persisted fact row can become untrustworthy."""
+
+    description: str
+    corrupt: Callable[[Path], None]
+    expected_found_count: int
+    expected_republished_hit: bool
+
+
+@dataclass(frozen=True)
+class FactCacheRetentionTestCase:
+    """A sequence of (key parts, slot) publications and the rows that must remain."""
+
+    description: str
+    publications: tuple[tuple[tuple[str, ...], str], ...]
+    stale_key_parts: tuple[str, ...]
+    expected_slots: list[str]
+    expected_stale_found_count: int
+
+
+@dataclass(frozen=True)
+class FactCacheNoPersistenceTestCase:
+    """One invocation shape that must leave no fact database behind."""
+
+    description: str
+    expected_enabled: bool
+    expected_database_files: list[Path]
+
+
+@dataclass(frozen=True)
+class FactCacheWriterErrorTestCase:
+    """One SQLite error raised while publishing after an earlier fact was stored."""
+
+    description: str
+    error: Exception
+    expected_earlier_found_count: int
+
+
+@dataclass(frozen=True)
+class FactCacheParameterLimitTestCase:
+    """One SQLite bound-parameter limit and a publication larger than one statement allows."""
+
+    description: str
+    max_bound_parameters: int
+    fact_count: int
+    expected_stored_count: int
diff --git a/tests/unit/src/sqlbuild/compiler/fact_cache/classes/helpers.py b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/helpers.py
new file mode 100644
index 0000000000..853b5dc9b3
--- /dev/null
+++ b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/helpers.py
@@ -0,0 +1,87 @@
+"""Helpers for fact cache store unit tests."""
+
+from __future__ import annotations
+
+import pickle
+import sqlite3
+from contextlib import closing
+from pathlib import Path
+
+from sqlbuild.compiler.fact_cache._helpers.publication import (
+    await_fact_publication,
+    entry_digest,
+)
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
+from sqlbuild.spec.contracts.models import SourceLocation
+
+FACT_NAMESPACE: str = "unit"
+FACT_ALGORITHM: str = "unit-facts-v1"
+FACT_VALUE: tuple[SourceLocation, ...] = (
+    SourceLocation(path=Path("models/orders.sql"), line=3, column=5, end_line=3, end_column=12),
+)
+
+
+def fact_database(root: Path) -> Path:
+    return next(root.rglob(f"{FACT_NAMESPACE}.sqlite3"))
+
+
+def publish_fact(
+    root: Path, *, key_parts: tuple[str, ...], slot: str, value: object = FACT_VALUE
+) -> str:
+    with FactCacheStore(root=root, namespace=FACT_NAMESPACE, algorithm=FACT_ALGORITHM) as store:
+        key: str = store.key(*key_parts)
+        store.stage(key=key, slot=slot, value=value)
+    await_fact_publication()
+    return key
+
+
+def read_fact(
+    root: Path | None,
+    *,
+    key_parts: tuple[str, ...],
+    slot: str = "orders",
+    algorithm: str = FACT_ALGORITHM,
+) -> dict[str, object]:
+    with FactCacheStore(root=root, namespace=FACT_NAMESPACE, algorithm=algorithm) as store:
+        return store.read_many(((slot, store.key(*key_parts)),))
+
+
+def stored_fact_slots(root: Path) -> list[str]:
+    with closing(sqlite3.connect(fact_database(root))) as connection:
+        rows: list[tuple[str]] = connection.execute(
+            "SELECT slot FROM fact ORDER BY slot"
+        ).fetchall()
+    return [slot for (slot,) in rows]
+
+
+def _overwrite_verified_payload(database: Path, payload: bytes) -> None:
+    with closing(sqlite3.connect(database)) as connection, connection:
+        cache_key: str = connection.execute("SELECT cache_key FROM fact").fetchone()[0]
+        digest: str = entry_digest(cache_key=cache_key, payload=payload)
+        _ = connection.execute("UPDATE fact SET payload = ?, digest = ?", (payload, digest))
+
+
+def flip_payload_byte(root: Path) -> None:
+    database: Path = fact_database(root)
+    with closing(sqlite3.connect(database)) as connection, connection:
+        payload: bytes = connection.execute("SELECT payload FROM fact").fetchone()[0]
+        _ = connection.execute("UPDATE fact SET payload = ?", (payload[:-1] + b"\x00",))
+
+
+def write_untrusted_global(root: Path) -> None:
+    _overwrite_verified_payload(fact_database(root), pickle.dumps(Path.home, protocol=5))
+
+
+def truncate_verified_payload(root: Path) -> None:
+    _overwrite_verified_payload(fact_database(root), b"\x80\x05")
+
+
+def replace_database_with_garbage(root: Path) -> None:
+    fact_database(root).write_bytes(b"not a sqlite database")
+
+
+def publish_facts(root: Path, *, slots: tuple[str, ...]) -> None:
+    with FactCacheStore(root=root, namespace=FACT_NAMESPACE, algorithm=FACT_ALGORITHM) as store:
+        for slot in slots:
+            store.stage(key=store.key(slot), slot=slot, value=FACT_VALUE)
+    await_fact_publication()
diff --git a/tests/unit/src/sqlbuild/compiler/fact_cache/classes/test_fact_cache_store.py b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/test_fact_cache_store.py
new file mode 100644
index 0000000000..69d738f151
--- /dev/null
+++ b/tests/unit/src/sqlbuild/compiler/fact_cache/classes/test_fact_cache_store.py
@@ -0,0 +1,314 @@
+from __future__ import annotations
+
+import sqlite3
+import threading
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from sqlbuild.compiler.fact_cache._helpers import publication
+from sqlbuild.compiler.fact_cache.classes import fact_cache_store
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
+from sqlbuild.compiler.profiling.main.collect import collect_compile_timings
+from tests.unit.src.sqlbuild.compiler.fact_cache.classes._test_types import (
+    FactCacheCodeIdentityTestCase,
+    FactCacheCorruptionTestCase,
+    FactCacheNoPersistenceTestCase,
+    FactCacheParameterLimitTestCase,
+    FactCacheReadTestCase,
+    FactCacheRetentionTestCase,
+    FactCacheWriterErrorTestCase,
+)
+from tests.unit.src.sqlbuild.compiler.fact_cache.classes.helpers import (
+    FACT_ALGORITHM,
+    FACT_NAMESPACE,
+    FACT_VALUE,
+    flip_payload_byte,
+    publish_fact,
+    publish_facts,
+    read_fact,
+    replace_database_with_garbage,
+    stored_fact_slots,
+    truncate_verified_payload,
+    write_untrusted_global,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheReadTestCase(
+            description="same_inputs_hit",
+            read_key_parts=("orders", "SELECT 1"),
+            read_algorithm=FACT_ALGORITHM,
+            expected_hit=True,
+            expected_metrics=(1, 0),
+        ),
+        FactCacheReadTestCase(
+            description="changed_input_misses",
+            read_key_parts=("orders", "SELECT 2"),
+            read_algorithm=FACT_ALGORITHM,
+            expected_hit=False,
+            expected_metrics=(0, 1),
+        ),
+        FactCacheReadTestCase(
+            description="extra_empty_input_misses",
+            read_key_parts=("orders", "SELECT 1", ""),
+            read_algorithm=FACT_ALGORITHM,
+            expected_hit=False,
+            expected_metrics=(0, 1),
+        ),
+        FactCacheReadTestCase(
+            description="concatenated_inputs_miss",
+            read_key_parts=("ordersSELECT 1",),
+            read_algorithm=FACT_ALGORITHM,
+            expected_hit=False,
+            expected_metrics=(0, 1),
+        ),
+        FactCacheReadTestCase(
+            description="changed_algorithm_misses",
+            read_key_parts=("orders", "SELECT 1"),
+            read_algorithm="unit-facts-v2",
+            expected_hit=False,
+            expected_metrics=(0, 1),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_published_fact_when_reading_then_hits_only_for_identical_inputs(
+    tmp_path: Path, test_case: FactCacheReadTestCase
+) -> None:
+    key: str = publish_fact(tmp_path, key_parts=("orders", "SELECT 1"), slot="orders")
+
+    with collect_compile_timings() as timings:
+        found: dict[str, object] = read_fact(
+            tmp_path, key_parts=test_case.read_key_parts, algorithm=test_case.read_algorithm
+        )
+
+    assert (found == {key: FACT_VALUE}) is test_case.expected_hit
+    assert (
+        timings.metrics["fact_cache_hits"],
+        timings.metrics["fact_cache_misses"],
+    ) == test_case.expected_metrics
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheCodeIdentityTestCase(
+            description="next_release", next_code_identity="next-release", expected_found_count=0
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_published_fact_when_producing_code_changes_then_misses(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, test_case: FactCacheCodeIdentityTestCase
+) -> None:
+    _ = publish_fact(tmp_path, key_parts=("orders",), slot="orders")
+    monkeypatch.setattr(
+        fact_cache_store, "installed_code_identity", lambda: test_case.next_code_identity
+    )
+
+    assert len(read_fact(tmp_path, key_parts=("orders",))) == test_case.expected_found_count
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheCorruptionTestCase(
+            description="payload_bytes_changed",
+            corrupt=flip_payload_byte,
+            expected_found_count=0,
+            expected_republished_hit=True,
+        ),
+        FactCacheCorruptionTestCase(
+            description="verified_payload_references_function",
+            corrupt=write_untrusted_global,
+            expected_found_count=0,
+            expected_republished_hit=True,
+        ),
+        FactCacheCorruptionTestCase(
+            description="verified_payload_truncated",
+            corrupt=truncate_verified_payload,
+            expected_found_count=0,
+            expected_republished_hit=True,
+        ),
+        FactCacheCorruptionTestCase(
+            description="database_file_replaced",
+            corrupt=replace_database_with_garbage,
+            expected_found_count=0,
+            expected_republished_hit=True,
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_corrupted_fact_when_reading_then_misses_and_next_publication_repairs_it(
+    tmp_path: Path, test_case: FactCacheCorruptionTestCase
+) -> None:
+    _ = publish_fact(tmp_path, key_parts=("orders",), slot="orders")
+    test_case.corrupt(tmp_path)
+
+    assert len(read_fact(tmp_path, key_parts=("orders",))) == test_case.expected_found_count
+
+    key: str = publish_fact(tmp_path, key_parts=("orders",), slot="orders")
+    assert (
+        read_fact(tmp_path, key_parts=("orders",)) == {key: FACT_VALUE}
+    ) is test_case.expected_republished_hit
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheRetentionTestCase(
+            description="same_slot_replaced",
+            publications=(
+                (("orders", "v1"), "orders"),
+                (("orders", "v2"), "orders"),
+                (("customers", "v1"), "customers"),
+            ),
+            stale_key_parts=("orders", "v1"),
+            expected_slots=["customers", "orders"],
+            expected_stale_found_count=0,
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_new_fact_for_same_slot_when_publishing_then_previous_fact_is_pruned(
+    tmp_path: Path, test_case: FactCacheRetentionTestCase
+) -> None:
+    for key_parts, slot in test_case.publications:
+        _ = publish_fact(tmp_path, key_parts=key_parts, slot=slot)
+
+    assert stored_fact_slots(tmp_path) == test_case.expected_slots
+    assert (
+        len(read_fact(tmp_path, key_parts=test_case.stale_key_parts))
+        == test_case.expected_stale_found_count
+    )
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheNoPersistenceTestCase(
+            description="disabled_root", expected_enabled=False, expected_database_files=[]
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_disabled_root_when_staging_then_nothing_is_persisted(
+    tmp_path: Path, test_case: FactCacheNoPersistenceTestCase
+) -> None:
+    with FactCacheStore(root=None, namespace=FACT_NAMESPACE, algorithm=FACT_ALGORITHM) as store:
+        store.stage(key=store.key("orders"), slot="orders", value=FACT_VALUE)
+        assert store.enabled is test_case.expected_enabled
+        assert store.read_many((("orders", store.key("orders")),)) == {}
+
+    assert list(tmp_path.iterdir()) == test_case.expected_database_files
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheNoPersistenceTestCase(
+            description="failed_invocation", expected_enabled=True, expected_database_files=[]
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_failed_invocation_when_exiting_then_staged_facts_are_discarded(
+    tmp_path: Path, test_case: FactCacheNoPersistenceTestCase
+) -> None:
+    with (
+        pytest.raises(RuntimeError),
+        FactCacheStore(root=tmp_path, namespace=FACT_NAMESPACE, algorithm=FACT_ALGORITHM) as store,
+    ):
+        assert store.enabled is test_case.expected_enabled
+        store.stage(key=store.key("orders"), slot="orders", value=FACT_VALUE)
+        raise RuntimeError("compile failed")
+
+    assert list(tmp_path.rglob("*.sqlite3")) == test_case.expected_database_files
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheWriterErrorTestCase(
+            description="busy_database_keeps_rows",
+            error=sqlite3.OperationalError("database is locked"),
+            expected_earlier_found_count=1,
+        ),
+        FactCacheWriterErrorTestCase(
+            description="integrity_error_keeps_rows",
+            error=sqlite3.IntegrityError("constraint failed"),
+            expected_earlier_found_count=1,
+        ),
+        FactCacheWriterErrorTestCase(
+            description="interface_error_keeps_rows",
+            error=sqlite3.InterfaceError("bad parameter"),
+            expected_earlier_found_count=1,
+        ),
+        FactCacheWriterErrorTestCase(
+            description="unreadable_database_is_replaced",
+            error=sqlite3.DatabaseError("file is not a database"),
+            expected_earlier_found_count=0,
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_writer_error_when_publishing_then_only_corruption_discards_rows_and_nothing_escapes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, test_case: FactCacheWriterErrorTestCase
+) -> None:
+    _ = publish_fact(tmp_path, key_parts=("orders",), slot="orders")
+    escaped: list[threading.ExceptHookArgs] = []
+    monkeypatch.setattr(threading, "excepthook", escaped.append)
+
+    def failing_insert(**_: object) -> None:
+        raise test_case.error
+
+    monkeypatch.setattr(publication, "_insert_records", failing_insert)
+    _ = publish_fact(tmp_path, key_parts=("customers",), slot="customers")
+    monkeypatch.undo()
+
+    assert escaped == []
+    assert len(read_fact(tmp_path, key_parts=("orders",))) == (
+        test_case.expected_earlier_found_count
+    )
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        FactCacheParameterLimitTestCase(
+            description="sqlite_before_3_32",
+            max_bound_parameters=999,
+            fact_count=600,
+            expected_stored_count=600,
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_sqlite_parameter_limit_when_publishing_many_facts_then_all_are_stored(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, test_case: FactCacheParameterLimitTestCase
+) -> None:
+    connect: Callable[..., sqlite3.Connection] = sqlite3.connect
+
+    def limited_connect(database: Path, timeout: float = 5.0) -> sqlite3.Connection:
+        connection: sqlite3.Connection = connect(database, timeout=timeout)
+        _ = connection.setlimit(
+            sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, test_case.max_bound_parameters
+        )
+        return connection
+
+    monkeypatch.setattr(sqlite3, "connect", limited_connect)
+    slots: tuple[str, ...] = tuple(f"order_{index:04d}" for index in range(test_case.fact_count))
+
+    publish_facts(tmp_path, slots=slots)
+
+    stored: list[str] = stored_fact_slots(tmp_path)
+    assert len(stored) == test_case.expected_stored_count
+    assert stored == list(slots)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])