From 4626857b4fe92d4e117b45aec956d0df46726c1e Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Sun, 4 Oct 2026 09:43:57 +0100
Subject: [PATCH 1/3] perf(rules): evaluate built-in model rules incrementally

Declare an explicit scope on every built-in rule and reject findings a rule
emits outside its declared scope, so a new rule must state whether its
findings are model-local or cross-model.

Send each model to the native engine as its own encoded payload with a
digest of those exact bytes. The digest keys that model's cached findings,
the whole-response memo is keyed by the model-free request plus the ordered
digests, and models are decoded in parallel. Model findings no longer depend
on a project-wide fingerprint or on suppression lists, cross-model rules
always re-run, and cache entries for deleted model files are pruned.

Hash rule identities with BLAKE2b instead of SHA-256.
---
 .../src/bindings/_helpers/functions.rs        |  18 +-
 .../src/engine/_helpers/evaluation.rs         | 244 +++++++-----
 .../src/engine/main/evaluate_parts.rs         |   7 +
 .../src/engine/main/mod.rs                    |   1 +
 .../sqlbuild-rules-native/src/engine/tests.rs |   2 +
 .../src/engine/tests/helpers.rs               | 104 ++++-
 .../engine/tests/test_incremental_rules.rs    | 375 ++++++++++++++++++
 .../src/engine/tests/test_sql_test_rules.rs   |  35 +-
 .../src/engine/tests/test_types.rs            |  38 ++
 crates/sqlbuild-rules-native/src/models.rs    |  25 +-
 .../src/rules/_helpers/catalogue.rs           |  46 ++-
 src/sqlbuild/_native.pyi                      |   3 +
 .../rule_engine/_helpers/engine/native.py     |  97 +++--
 .../rule_engine/_helpers/run/native_memo.py   |  17 +-
 .../rule_engine/_helpers/run/rules.py         |   4 +-
 src/sqlbuild/rule_engine/constants.py         |   2 +-
 .../cli/commands/main/rules/_test_types.py    |  10 +
 .../cli/commands/main/rules/helpers.py        |  30 ++
 .../rules/test_rules_incremental_sequence.py  | 179 +++++++++
 .../_helpers/engine/_test_types.py            |  19 +
 .../rule_engine/_helpers/engine/helpers.py    | 106 ++++-
 .../engine/test_incremental_native_rules.py   | 162 ++++++++
 .../_helpers/engine/test_native_memo.py       |   2 +-
 .../_helpers/engine/test_native_request.py    |   8 +-
 24 files changed, 1378 insertions(+), 156 deletions(-)
 create mode 100644 crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
 create mode 100644 crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
 create mode 100644 tests/e2e/src/sqlbuild/cli/commands/main/rules/test_rules_incremental_sequence.py
 create mode 100644 tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_incremental_native_rules.py

diff --git a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
index 260f595971..a2ac476c74 100644
--- a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
+++ b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
@@ -3,12 +3,12 @@
 use pyo3::prelude::{
     Bound, IntoPyObject, Py, PyAny, PyErr, PyModule, PyModuleMethods, PyResult, Python,
 };
-use pyo3::types::{PyDict, PyDictMethods, PyList, PyTuple};
+use pyo3::types::{PyBytes, PyBytesMethods, PyDict, PyDictMethods, PyList, PyTuple};
 use pyo3::{FromPyObject, pyfunction, wrap_pyfunction};
 
 use crate::configuration::main::load;
 use crate::constants::{API_VERSION, NATIVE_BUILD_IDENTITY};
-use crate::engine::main::evaluate;
+use crate::engine::main::{evaluate, evaluate_parts};
 use crate::models::CatalogueResponse;
 use crate::rules::main::{catalogue, selected_codes};
 
@@ -73,6 +73,19 @@ fn evaluate_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
         .map_err(value_error)
 }
 
+#[pyfunction]
+fn evaluate_rules_parts(
+    py: Python<'_>,
+    request_json: Bound<'_, PyBytes>,
+    model_jsons: Vec<Bound<'_, PyBytes>>,
+    model_digests: Vec<String>,
+) -> PyResult<String> {
+    let request: &[u8] = request_json.as_bytes();
+    let models: Vec<&[u8]> = model_jsons.iter().map(|model| model.as_bytes()).collect();
+    py.compiler_detach(|| evaluate_parts::evaluate_parts(request, &models, &model_digests))
+        .map_err(value_error)
+}
+
 #[pyfunction]
 fn run_custom_host_json(py: Python<'_>, spec_json: &str) -> PyResult<String> {
     py.compiler_detach(|| crate::engine::main::custom_host::run_custom_host_json(spec_json))
@@ -465,6 +478,7 @@ fn skill_freshness(content: Option<&str>, input_fingerprint: &str) -> String {
 
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(evaluate_json, module)?)?;
+    module.add_function(wrap_pyfunction!(evaluate_rules_parts, module)?)?;
     module.add_function(wrap_pyfunction!(finalize_rule_findings_json, module)?)?;
     module.add_function(wrap_pyfunction!(run_custom_host_json, module)?)?;
     module.add_function(wrap_pyfunction!(lint_sql_json, module)?)?;
diff --git a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
index 746de6cda6..e36c78b35d 100644
--- a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
+++ b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
@@ -2,7 +2,8 @@ use crate::configuration::main::validate as config;
 use crate::constants::{API_VERSION, NATIVE_BUILD_IDENTITY};
 use crate::engine::_helpers::cache::{Cache, RuleCacheEntry};
 use crate::models::{
-    EvaluateRequest, EvaluateResponse, Fault, Model, RuleMetadata, RulesCodeGrammar,
+    EvaluateRequest, EvaluateResponse, Fault, Model, RuleMetadata, RuleScope, RulesCodeGrammar,
+    RulesConfig,
 };
 use crate::rules::main::{
     assemble_catalogue, evaluate as rules, evaluate_project, fingerprint,
@@ -11,13 +12,11 @@ use crate::rules::main::{
 use crate::rules::models::{
     ModelEvaluationRequest, ProjectEvaluationRequest, ResolvedThresholdOverride,
 };
-use fensu_policy::lifecycle::constants::ANALYSIS_BATCH_SCHEMA_VERSION;
+use fensu_policy::apply_suppressions;
 use fensu_policy::lifecycle::errors::LifecycleError;
 use fensu_policy::lifecycle::models::{
-    AnalysisBatchRequest, AnalysisInput, ApplySuppressionsRequest, ExactSuppression, Finding,
-    FindingSeverity, RuntimeIdentity, ScopedIgnore,
+    ApplySuppressionsRequest, ExactSuppression, Finding, FindingSeverity, ScopedIgnore,
 };
-use fensu_policy::{apply_suppressions, evaluate_batch};
 use rayon::iter::{
     IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
 };
@@ -28,7 +27,7 @@ use std::time::Instant;
 
 const NATIVE_RULE_WORKERS: usize = 4;
 const NATIVE_RULE_STACK_BYTES: usize = 16 * 1024 * 1024;
-const COMPILER_FACTS_FINGERPRINT_SEED: &str = "compiler-owned-rules-facts-v2";
+const MODEL_RULES_IDENTITY_SEED: &[u8] = b"sqlbuild-rules-model-v2";
 
 struct PendingModelRule<'a> {
     index: usize,
@@ -51,8 +50,7 @@ struct ModelRuleBatchContext<'a> {
 struct NativeModelRuleRequest<'a> {
     context: ModelRuleBatchContext<'a>,
     cache: Option<&'a Cache>,
-    ruleset_fingerprint: &'a str,
-    project_fingerprint: Option<&'a str>,
+    model_rules_fingerprint: &'a str,
 }
 
 #[derive(Default)]
@@ -68,8 +66,7 @@ fn evaluate_native_models(
     let NativeModelRuleRequest {
         context,
         cache,
-        ruleset_fingerprint,
-        project_fingerprint,
+        model_rules_fingerprint,
     } = input;
     if !context
         .selected
@@ -90,7 +87,7 @@ fn evaluate_native_models(
     });
     let model_limit = if all_models_required { models.len() } else { 1 };
     let mut bucket = cache
-        .map(|store| store.native_rules_bucket(ruleset_fingerprint))
+        .map(|store| store.native_rules_bucket(model_rules_fingerprint))
         .transpose()?
         .unwrap_or_default();
     let mut bucket_changed = false;
@@ -98,14 +95,10 @@ fn evaluate_native_models(
     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()
+        model_cache_identities(&models[..model_limit], model_rules_fingerprint)?
+            .into_iter()
+            .map(Some)
+            .collect()
     } else {
         vec![None; model_limit]
     };
@@ -128,6 +121,7 @@ fn evaluate_native_models(
             identity,
         });
     }
+    let project_dir = Path::new(&context.request.project_dir);
     let completed = evaluate_pending_models(pending_models, context)?;
     for completed_model in completed {
         let faults = completed_model.faults?;
@@ -145,7 +139,14 @@ fn evaluate_native_models(
     }
     result.faults = model_results.into_iter().flatten().flatten().collect();
     if bucket_changed && let Some(cache) = cache {
-        cache.put_native_rules_bucket(ruleset_fingerprint, &bucket)?;
+        let current: BTreeSet<&str> = models[..model_limit]
+            .iter()
+            .map(|model| model.relative_path.as_str())
+            .collect();
+        bucket
+            .entries
+            .retain(|path, _| current.contains(path.as_str()) || project_dir.join(path).is_file());
+        cache.put_native_rules_bucket(model_rules_fingerprint, &bucket)?;
     }
     Ok(result)
 }
@@ -163,17 +164,13 @@ fn evaluate_pending_models(
             selected: context.selected,
             request: context.request,
             threshold_overrides: context.threshold_overrides,
-        }),
+        })
+        .and_then(|faults| scoped_faults(context.selected, faults, RuleScope::evaluates_models)),
     };
     if pending_models.len() <= 1 {
         return Ok(pending_models.into_iter().map(evaluate_pending).collect());
     }
-    let pool = rayon::ThreadPoolBuilder::new()
-        .num_threads(NATIVE_RULE_WORKERS.min(pending_models.len()))
-        .stack_size(NATIVE_RULE_STACK_BYTES)
-        .build()
-        .map_err(|error| error.to_string())?;
-    Ok(pool.install(|| {
+    Ok(native_rule_pool(pending_models.len())?.install(|| {
         pending_models
             .into_par_iter()
             .map(evaluate_pending)
@@ -184,6 +181,95 @@ fn evaluate_pending_models(
 pub(crate) fn evaluate_json(request_json: &str) -> Result<String, String> {
     let request: EvaluateRequest = serde_json::from_str(request_json)
         .map_err(|error| format!("invalid rules request: {error}"))?;
+    evaluate_request(request)
+}
+
+/// Evaluate a request whose models arrive as separate payloads, each with its exact digest.
+pub(crate) fn evaluate_parts(
+    request_json: &[u8],
+    model_jsons: &[&[u8]],
+    model_digests: &[String],
+) -> Result<String, String> {
+    let mut request: EvaluateRequest = serde_json::from_slice(request_json)
+        .map_err(|error| format!("invalid rules request: {error}"))?;
+    if !request.models.is_empty() {
+        return Err("invalid rules request: models must be sent as separate payloads".to_owned());
+    }
+    if model_jsons.len() != model_digests.len() {
+        return Err("invalid rules request: every model payload needs one digest".to_owned());
+    }
+    request.models = parse_model_payloads(model_jsons, model_digests)?;
+    evaluate_request(request)
+}
+
+fn parse_model_payloads(
+    model_jsons: &[&[u8]],
+    model_digests: &[String],
+) -> Result<Vec<Model>, String> {
+    let parse = |(payload, digest): (&&[u8], &String)| -> Result<Model, String> {
+        let mut model: Model = serde_json::from_slice(payload)
+            .map_err(|error| format!("invalid rules request: {error}"))?;
+        model.payload_digest = Some(digest.clone());
+        Ok(model)
+    };
+    let results: Vec<Result<Model, String>> = if model_jsons.len() <= 1 {
+        model_jsons.iter().zip(model_digests).map(parse).collect()
+    } else {
+        native_rule_pool(model_jsons.len())?.install(|| {
+            model_jsons
+                .par_iter()
+                .zip(model_digests.par_iter())
+                .map(parse)
+                .collect()
+        })
+    };
+    results.into_iter().collect()
+}
+
+fn native_rule_pool(items: usize) -> Result<rayon::ThreadPool, String> {
+    rayon::ThreadPoolBuilder::new()
+        .num_threads(NATIVE_RULE_WORKERS.min(items))
+        .stack_size(NATIVE_RULE_STACK_BYTES)
+        .build()
+        .map_err(|error| error.to_string())
+}
+
+/// Reject findings from a phase that the emitting rule's declared scope does not cover.
+fn scoped_faults(
+    selected: &BTreeMap<String, &RuleMetadata>,
+    faults: Vec<Fault>,
+    covers: fn(RuleScope) -> bool,
+) -> Result<Vec<Fault>, String> {
+    if let Some(fault) = faults.iter().find(|fault| {
+        selected
+            .get(&fault.code)
+            .is_some_and(|rule| !covers(rule.scope))
+    }) {
+        return Err(format!(
+            "rule {} emitted a finding outside its declared scope",
+            fault.code
+        ));
+    }
+    Ok(faults)
+}
+
+/// Fingerprint model-rule inputs; suppressions and graph-edge exceptions never reach model rules.
+fn model_rules_fingerprint(
+    selected: &[&RuleMetadata],
+    config: &RulesConfig,
+    dialect: &str,
+) -> Result<String, String> {
+    let model_config = RulesConfig {
+        rule_exceptions: Vec::new(),
+        rule_ignores: Vec::new(),
+        graph_edge_exceptions: Vec::new(),
+        cache: Default::default(),
+        ..config.clone()
+    };
+    fingerprint::fingerprint(selected, &model_config, dialect)
+}
+
+fn evaluate_request(request: EvaluateRequest) -> Result<String, String> {
     if request.version != API_VERSION {
         return Err(format!(
             "unsupported rules native API version {}; expected {API_VERSION}",
@@ -229,29 +315,22 @@ pub(crate) fn evaluate_json(request_json: &str) -> Result<String, String> {
     }
 
     let cache_enabled = request.config.cache.enabled;
-    let project_aware = selected.iter().any(|rule| rule.project_wide || rule.custom);
-    let project_fingerprint = project_aware
-        .then(|| {
-            fingerprint_request_facts(
-                request
-                    .project_fingerprint
-                    .as_deref()
-                    .unwrap_or(COMPILER_FACTS_FINGERPRINT_SEED)
-                    .as_bytes(),
-                &request,
-            )
-        })
-        .transpose()?;
+    let model_rules_fingerprint = cache_enabled
+        .then(|| model_rules_fingerprint(&selected, &request.config, &request.dialect))
+        .transpose()?
+        .unwrap_or_default();
     let cache = cache_enabled
         .then(|| Cache::open(Path::new(&request.project_dir)))
         .transpose()?;
     let built_in_started = Instant::now();
     let mut raw_faults = request.initial_findings.clone();
-    raw_faults.extend(evaluate_project::evaluate_project(
-        ProjectEvaluationRequest {
+    raw_faults.extend(scoped_faults(
+        &selected_by_code,
+        evaluate_project::evaluate_project(ProjectEvaluationRequest {
             selected: &selected_by_code,
             request: &request,
-        },
+        })?,
+        RuleScope::evaluates_project,
     )?);
     let native = evaluate_native_models(NativeModelRuleRequest {
         context: ModelRuleBatchContext {
@@ -260,8 +339,7 @@ pub(crate) fn evaluate_json(request_json: &str) -> Result<String, String> {
             threshold_overrides: &threshold_overrides,
         },
         cache: cache.as_ref(),
-        ruleset_fingerprint: &ruleset_fingerprint,
-        project_fingerprint: project_fingerprint.as_deref(),
+        model_rules_fingerprint: &model_rules_fingerprint,
     })?;
     let cache_hits = native.hits;
     let cache_misses = native.misses;
@@ -288,66 +366,38 @@ pub(crate) fn evaluate_json(request_json: &str) -> Result<String, String> {
 
 fn model_cache_identities(
     models: &[&crate::models::Model],
-    ruleset: &str,
-    project: Option<&str>,
+    model_rules: &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 identity = |model: &&crate::models::Model| model_cache_identity(model, model_rules);
     let results: Vec<Result<String, String>> = if models.len() <= 1 {
-        models.iter().enumerate().map(identity).collect()
+        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().enumerate().map(identity).collect())
+        native_rule_pool(models.len())?.install(|| models.par_iter().map(identity).collect())
     };
     results.into_iter().collect()
 }
 
-fn model_cache_identity(
-    model: &crate::models::Model,
-    ruleset: &str,
-    project: Option<&str>,
-) -> Result<String, String> {
-    let request = AnalysisBatchRequest {
-        schema_version: ANALYSIS_BATCH_SCHEMA_VERSION,
-        required_capabilities: Vec::new(),
-        identity: RuntimeIdentity {
-            producer: "sqlbuild-rules".to_owned(),
-            fact_schema: "sqlbuild-rules-model-v1".to_owned(),
-            runtime: NATIVE_BUILD_IDENTITY.to_owned(),
-            rule_pack: ruleset.to_owned(),
-            configuration: project.unwrap_or("model-local").to_owned(),
-        },
-        inputs: vec![AnalysisInput {
-            path: model.relative_path.clone(),
-            fingerprint: "compiled-model-v1".to_owned(),
-            facts: model,
-        }],
-    };
-    evaluate_batch(&request, &[], |_| Ok(Vec::new()))
-        .map(|response| response.cache_identity)
-        .map_err(lifecycle_error)
-}
-
-fn fingerprint_request_facts(seed: &[u8], request: &EvaluateRequest) -> Result<String, String> {
+/// Key one model's findings by payload, model-rules fingerprint, and derived test counts.
+fn model_cache_identity(model: &crate::models::Model, model_rules: &str) -> Result<String, String> {
     let mut digest = Sha256::new();
-    digest.update(seed);
-    digest.update(
-        serde_json::to_vec(&(
-            &request.models,
-            &request.public_enums,
-            &request.public_constants,
-            &request.sql_tests,
-            &request.sql_scenarios,
-            &request.scope_index,
-            &request.custom_rules,
-        ))
-        .map_err(|error| error.to_string())?,
-    );
+    digest.update(MODEL_RULES_IDENTITY_SEED);
+    digest.update([0]);
+    digest.update(NATIVE_BUILD_IDENTITY.as_bytes());
+    digest.update([0]);
+    digest.update(model_rules.as_bytes());
+    digest.update([0]);
+    match &model.payload_digest {
+        Some(payload) => {
+            digest.update(b"payload:");
+            digest.update(payload.as_bytes());
+        }
+        None => {
+            digest.update(b"model:");
+            digest.update(serde_json::to_vec(model).map_err(|error| error.to_string())?);
+        }
+    }
+    digest.update([0]);
+    digest.update(model.empty_input_only_test_count.to_le_bytes());
     Ok(format!("{:x}", digest.finalize()))
 }
 
diff --git a/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs b/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
new file mode 100644
index 0000000000..aae707f28b
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
@@ -0,0 +1,7 @@
+pub(crate) fn evaluate_parts(
+    request_json: &[u8],
+    model_jsons: &[&[u8]],
+    model_digests: &[String],
+) -> Result<String, String> {
+    crate::engine::_helpers::evaluation::evaluate_parts(request_json, model_jsons, model_digests)
+}
diff --git a/crates/sqlbuild-rules-native/src/engine/main/mod.rs b/crates/sqlbuild-rules-native/src/engine/main/mod.rs
index ae4f2a1b3e..e35b0f83c0 100644
--- a/crates/sqlbuild-rules-native/src/engine/main/mod.rs
+++ b/crates/sqlbuild-rules-native/src/engine/main/mod.rs
@@ -1,3 +1,4 @@
 pub(crate) mod custom_host;
 pub(crate) mod evaluate;
+pub(crate) mod evaluate_parts;
 pub(crate) mod finalize;
diff --git a/crates/sqlbuild-rules-native/src/engine/tests.rs b/crates/sqlbuild-rules-native/src/engine/tests.rs
index 08d7e5b848..1dd70e7f7c 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests.rs
@@ -6,6 +6,8 @@ mod empty_input_tests;
 mod evaluation;
 #[path = "tests/helpers.rs"]
 mod helpers;
+#[path = "tests/test_incremental_rules.rs"]
+mod incremental_rules;
 #[path = "tests/test_scope_facts.rs"]
 mod scope_facts;
 #[path = "tests/test_sql_scanning.rs"]
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs b/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
index ee51aec85c..d7fc43079e 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
@@ -1,7 +1,9 @@
 use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
 use tempfile::TempDir;
 
 use crate::engine::main::evaluate::evaluate_json;
+use crate::engine::main::evaluate_parts::evaluate_parts;
 
 pub(crate) fn request(project_dir: &TempDir, config: &Value) -> String {
     json!({
@@ -212,7 +214,6 @@ pub(crate) fn request_with_scope(scope: Value) -> Value {
         "public_enums": [],
         "public_constants": [],
         "custom_rules": [],
-        "project_fingerprint": null,
         "scope_index": scope
     })
 }
@@ -270,7 +271,6 @@ pub(crate) fn cached_sql_test_rules_request(project_dir: &TempDir, name: Option<
     json!({
         "version": 1,
         "project_dir": project_dir.path(),
-        "project_fingerprint": "compiler-project-v1",
         "config": {"select": ["SQBRTEST104"], "cache": {"enabled": true}},
         "models": [{
             "name": "orders", "relative_path": "models/orders.sql",
@@ -377,3 +377,103 @@ pub(crate) fn with_field(mut fact: Value, field: &str, value: Value) -> Value {
     fact[field] = value;
     fact
 }
+
+pub(crate) fn orders_model(path: &str, query: &str) -> Value {
+    let name = path
+        .rsplit('/')
+        .next()
+        .and_then(|file| file.strip_suffix(".sql"))
+        .unwrap_or("commerce__mart__orders");
+    json!({
+        "name": name,
+        "relative_path": path,
+        "query_sql": query,
+        "authored_sql": query,
+        "config": {"materialized": "table"},
+        "references": [{"ref_kind": "ref", "ref_name": "commerce__int_clean__orders"}],
+        "targeting_test_count": 0
+    })
+}
+
+pub(crate) fn orders_models() -> Vec<Value> {
+    vec![
+        orders_model(
+            "models/commerce/mart/commerce__mart__orders.sql",
+            "SELECT * FROM warehouse.raw.orders",
+        ),
+        orders_model(
+            "models/commerce/mart/commerce__mart__customers.sql",
+            "WITH customers AS (SELECT 1 AS customer_id) SELECT customer_id FROM customers",
+        ),
+        orders_model(
+            "models/commerce/staging/commerce__stg__products.sql",
+            "SELECT product_id, 1 AS quantity FROM products WHERE status = 3",
+        ),
+    ]
+}
+
+pub(crate) fn split_request(project_dir: &TempDir, config: &Value) -> Value {
+    json!({
+        "version": 1,
+        "project_dir": project_dir.path(),
+        "dialect": "duckdb",
+        "config": config,
+        "models": [],
+        "defer_suppressions": true
+    })
+}
+
+pub(crate) fn evaluate_split(rest: &Value, models: &[Value]) -> Result<Value, String> {
+    let encoded: Vec<Vec<u8>> = models
+        .iter()
+        .map(|model| serde_json::to_vec(model).map_err(|error| error.to_string()))
+        .collect::<Result<_, _>>()?;
+    let payloads: Vec<&[u8]> = encoded.iter().map(Vec::as_slice).collect();
+    let digests: Vec<String> = payloads
+        .iter()
+        .map(|payload| format!("{:x}", Sha256::digest(payload)))
+        .collect();
+    let rest_json = serde_json::to_vec(rest).map_err(|error| error.to_string())?;
+    serde_json::from_str(&evaluate_parts(&rest_json, &payloads, &digests)?)
+        .map_err(|error| error.to_string())
+}
+
+pub(crate) fn evaluate_whole(rest: &Value, models: &[Value]) -> Result<Value, String> {
+    let mut request = rest.clone();
+    request["models"] = Value::Array(models.to_vec());
+    serde_json::from_str(&evaluate_json(&request.to_string())?).map_err(|error| error.to_string())
+}
+
+pub(crate) fn uncached(rest: &Value) -> Value {
+    let mut request = rest.clone();
+    request["config"]["cache"] = json!({"enabled": false});
+    request
+}
+
+pub(crate) fn write_model_files(project_dir: &TempDir, models: &[Value]) -> Result<(), String> {
+    for model in models {
+        let path = project_dir
+            .path()
+            .join(model["relative_path"].as_str().unwrap_or_default());
+        let parent = path
+            .parent()
+            .ok_or_else(|| "model path has no parent".to_owned())?;
+        std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
+        std::fs::write(&path, "SELECT 1").map_err(|error| error.to_string())?;
+    }
+    Ok(())
+}
+
+pub(crate) fn cached_paths(project_dir: &TempDir) -> Result<Vec<String>, String> {
+    let bytes = std::fs::read(
+        project_dir
+            .path()
+            .join("target/rules-cache/bulk/native.json"),
+    )
+    .map_err(|error| error.to_string())?;
+    let bucket: Value = serde_json::from_slice(&bytes).map_err(|error| error.to_string())?;
+    Ok(bucket["entries"]
+        .as_object()
+        .map(|entries| entries.keys().cloned().collect())
+        .unwrap_or_default())
+}
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
new file mode 100644
index 0000000000..6e58c7bfd6
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
@@ -0,0 +1,375 @@
+use std::collections::BTreeMap;
+
+use serde_json::{Value, json};
+use tempfile::TempDir;
+
+use crate::engine::main::evaluate_parts::evaluate_parts;
+use crate::engine::tests::helpers::{
+    cached_paths, evaluate_split, evaluate_whole, orders_models, split_request, uncached,
+    write_model_files,
+};
+use crate::engine::tests::test_types;
+use crate::models::RuleScope;
+use crate::rules::main::catalogue::catalogue;
+
+const REVIEWED_SCOPES: &[(&str, RuleScope)] = &[
+    ("SQBRCONTRACT101", RuleScope::Model),
+    ("SQBRCONTRACT102", RuleScope::Model),
+    ("SQBRCONTRACT103", RuleScope::Model),
+    ("SQBRCONTRACT104", RuleScope::Model),
+    ("SQBRCONTRACT105", RuleScope::Model),
+    ("SQBRCONTRACT106", RuleScope::Model),
+    ("SQBRDECLARATION101", RuleScope::Model),
+    ("SQBRDECLARATION102", RuleScope::Model),
+    ("SQBRDECLARATION201", RuleScope::Project),
+    ("SQBRDECLARATION301", RuleScope::Project),
+    ("SQBRDECLARATION302", RuleScope::Project),
+    ("SQBRDECLARATION303", RuleScope::Project),
+    ("SQBRDECLARATION304", RuleScope::Project),
+    ("SQBRDECLARATION305", RuleScope::Project),
+    ("SQBRDECLARATION306", RuleScope::Project),
+    ("SQBRGRAPH101", RuleScope::Project),
+    ("SQBRGRAPH102", RuleScope::Model),
+    ("SQBRMODEL101", RuleScope::Model),
+    ("SQBRMODEL102", RuleScope::Model),
+    ("SQBRMODEL103", RuleScope::Model),
+    ("SQBRMODEL104", RuleScope::Model),
+    ("SQBRPROJECT101", RuleScope::Model),
+    ("SQBRPROJECT102", RuleScope::Model),
+    ("SQBRPROJECT103", RuleScope::Model),
+    ("SQBRPROJECT104", RuleScope::Model),
+    ("SQBRPROJECT105", RuleScope::Model),
+    ("SQBRPROJECT106", RuleScope::Model),
+    ("SQBRPROJECT201", RuleScope::Project),
+    ("SQBRPROJECT202", RuleScope::Project),
+    ("SQBRPROJECT203", RuleScope::Project),
+    ("SQBRPROJECT204", RuleScope::Project),
+    ("SQBRTEST101", RuleScope::Project),
+    ("SQBRTEST102", RuleScope::Project),
+    ("SQBRTEST103", RuleScope::Project),
+    ("SQBRTEST104", RuleScope::Project),
+    ("SQBRTEST201", RuleScope::ModelAndProject),
+    ("SQBRTEST202", RuleScope::ModelAndProject),
+    ("SQBRTEST203", RuleScope::Project),
+    ("SQBRTEST301", RuleScope::Project),
+];
+
+#[test]
+fn given_built_in_catalogue_when_reading_rule_scopes_then_every_rule_matches_reviewed_scope() {
+    let test_cases = [test_types::RuleScopeTestCase {
+        description: "every built-in rule declares its reviewed scope",
+        expected_scopes: REVIEWED_SCOPES,
+        expected_sql_scope: RuleScope::SqlLint,
+    }];
+    for test_case in &test_cases {
+        let declared: BTreeMap<String, RuleScope> = catalogue()
+            .into_iter()
+            .filter(|rule| !rule.code.starts_with("SQBRSQL"))
+            .map(|rule| (rule.code, rule.scope))
+            .collect();
+        let declared: Vec<(&str, RuleScope)> = declared
+            .iter()
+            .map(|(code, scope)| (code.as_str(), *scope))
+            .collect();
+        let sql_scopes_match = catalogue()
+            .iter()
+            .filter(|rule| rule.code.starts_with("SQBRSQL"))
+            .all(|rule| rule.scope == test_case.expected_sql_scope);
+
+        assert_eq!(
+            declared, test_case.expected_scopes,
+            "{}",
+            test_case.description
+        );
+        assert!(sql_scopes_match, "{}", test_case.description);
+    }
+}
+
+#[test]
+fn given_every_built_in_rule_when_evaluating_then_findings_stay_inside_declared_scopes()
+-> Result<(), String> {
+    let test_cases = [test_types::ScopedEvaluationTestCase {
+        description: "model and project findings come from scoped phases",
+        expected_codes: &["SQBRCONTRACT101", "SQBRGRAPH102", "SQBRTEST202"],
+    }];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let rest = split_request(
+            &project_dir,
+            &json!({"select": ["SQBR"], "domains": ["commerce"], "cache": {"enabled": false}}),
+        );
+
+        let result = evaluate_whole(&rest, &orders_models())?;
+        let codes: Vec<&str> = result["faults"]
+            .as_array()
+            .into_iter()
+            .flatten()
+            .filter_map(|fault| fault["code"].as_str())
+            .collect();
+
+        let found: Vec<&str> = test_case
+            .expected_codes
+            .iter()
+            .copied()
+            .filter(|code| codes.contains(code))
+            .collect();
+
+        assert_eq!(found, test_case.expected_codes, "{}", test_case.description);
+    }
+    Ok(())
+}
+
+#[test]
+fn given_split_model_payloads_when_evaluating_then_response_matches_single_request()
+-> Result<(), String> {
+    let test_cases = [
+        test_types::IncrementalRulesTestCase {
+            description: "uncached split request",
+            cache_enabled: false,
+            compared_fields: &["faults", "selected_codes", "evaluated_models"],
+            expected_mismatched_fields: &[],
+        },
+        test_types::IncrementalRulesTestCase {
+            description: "cached split request",
+            cache_enabled: true,
+            compared_fields: &["faults", "selected_codes", "evaluated_models"],
+            expected_mismatched_fields: &[],
+        },
+    ];
+    for test_case in &test_cases {
+        let whole_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let split_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let config = json!({"select": ["SQBR"], "cache": {"enabled": test_case.cache_enabled}});
+
+        let whole = evaluate_whole(&split_request(&whole_dir, &config), &orders_models())?;
+        let split = evaluate_split(&split_request(&split_dir, &config), &orders_models())?;
+
+        let mismatched: Vec<&str> = test_case
+            .compared_fields
+            .iter()
+            .copied()
+            .filter(|field| whole[field] != split[field])
+            .collect();
+
+        assert_eq!(
+            mismatched, test_case.expected_mismatched_fields,
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
+
+#[test]
+fn given_malformed_split_request_when_evaluating_then_request_is_rejected() -> Result<(), String> {
+    let test_cases = [
+        test_types::MalformedSplitRequestTestCase {
+            description: "models embedded in the request",
+            inline_model_count: 3,
+            payload_count: 0,
+            expected_error: "separate payloads",
+        },
+        test_types::MalformedSplitRequestTestCase {
+            description: "payload without a digest",
+            inline_model_count: 0,
+            payload_count: 1,
+            expected_error: "one digest",
+        },
+    ];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let mut request = split_request(&project_dir, &json!({"select": ["SQBR"]}));
+        request["models"] = Value::Array(
+            orders_models()
+                .into_iter()
+                .take(test_case.inline_model_count)
+                .collect(),
+        );
+        let rest = serde_json::to_vec(&request).map_err(|error| error.to_string())?;
+        let model = serde_json::to_vec(&orders_models()[0]).map_err(|error| error.to_string())?;
+        let payloads: Vec<&[u8]> = vec![model.as_slice(); test_case.payload_count];
+
+        let error = evaluate_parts(&rest, &payloads, &[]);
+
+        assert!(
+            error.is_err_and(|message| message.contains(test_case.expected_error)),
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
+
+#[test]
+fn given_cached_models_when_editing_one_model_then_only_it_reevaluates_with_uncached_findings()
+-> Result<(), String> {
+    let test_cases = [test_types::ModelCacheReuseTestCase {
+        description: "one edited model misses while the others hit",
+        expected_hits: 2,
+        expected_misses: 1,
+    }];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let rest = split_request(
+            &project_dir,
+            &json!({"select": ["SQBR"], "cache": {"enabled": true}}),
+        );
+        let mut models = orders_models();
+        write_model_files(&project_dir, &models)?;
+        let _ = evaluate_split(&rest, &models)?;
+        models[1]["query_sql"] = json!("SELECT * FROM warehouse.raw.customers");
+        models[1]["authored_sql"] = models[1]["query_sql"].clone();
+
+        let edited = evaluate_split(&rest, &models)?;
+        let expected = evaluate_split(&uncached(&rest), &models)?;
+
+        assert_eq!(
+            edited["cache_hits"], test_case.expected_hits,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            edited["cache_misses"], test_case.expected_misses,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            edited["faults"], expected["faults"],
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
+
+#[test]
+fn given_suppression_change_when_evaluating_then_cached_models_are_reused_with_uncached_findings()
+-> Result<(), String> {
+    let test_cases = [test_types::ModelCacheReuseTestCase {
+        description: "a scoped ignore does not invalidate model findings",
+        expected_hits: 3,
+        expected_misses: 0,
+    }];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let models = orders_models();
+        write_model_files(&project_dir, &models)?;
+        let mut rest = split_request(
+            &project_dir,
+            &json!({"select": ["SQBR"], "cache": {"enabled": true}}),
+        );
+        rest["defer_suppressions"] = json!(false);
+        let _ = evaluate_split(&rest, &models)?;
+        rest["config"]["rule_ignores"] = json!([{
+            "rules": ["SQBRGRAPH102"],
+            "paths": ["models/commerce/mart/commerce__mart__orders.sql"],
+            "reason": "legacy warehouse table"
+        }]);
+
+        let ignored = evaluate_split(&rest, &models)?;
+        let expected = evaluate_split(&uncached(&rest), &models)?;
+
+        assert_eq!(
+            ignored["cache_hits"], test_case.expected_hits,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            ignored["cache_misses"], test_case.expected_misses,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            ignored["faults"], expected["faults"],
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
+
+#[test]
+fn given_removed_model_file_when_cache_is_rewritten_then_only_its_entry_is_dropped()
+-> Result<(), String> {
+    let test_cases = [test_types::CachePruningTestCase {
+        description: "deleted model is pruned while an unselected model is kept",
+        expected_paths: &[
+            "models/commerce/mart/commerce__mart__customers.sql",
+            "models/commerce/staging/commerce__stg__products.sql",
+        ],
+    }];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let rest = split_request(
+            &project_dir,
+            &json!({"select": ["SQBR"], "cache": {"enabled": true}}),
+        );
+        let mut models = orders_models();
+        write_model_files(&project_dir, &models)?;
+        let _ = evaluate_split(&rest, &models)?;
+        let removed = models.remove(0);
+        std::fs::remove_file(
+            project_dir
+                .path()
+                .join(removed["relative_path"].as_str().unwrap_or_default()),
+        )
+        .map_err(|error| error.to_string())?;
+        let _unselected = models.remove(0);
+        models[0]["query_sql"] = json!("SELECT 2 AS product_id");
+        models[0]["authored_sql"] = models[0]["query_sql"].clone();
+
+        let _ = evaluate_split(&rest, &models)?;
+
+        assert_eq!(
+            cached_paths(&project_dir)?,
+            test_case.expected_paths,
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
+
+#[test]
+fn given_corrupt_model_cache_when_evaluating_then_every_model_reevaluates() -> Result<(), String> {
+    let test_cases = [test_types::ModelCacheReuseTestCase {
+        description: "a truncated cache file is a miss for every model",
+        expected_hits: 0,
+        expected_misses: 3,
+    }];
+    for test_case in &test_cases {
+        let project_dir = TempDir::new().map_err(|error| error.to_string())?;
+        let rest = split_request(
+            &project_dir,
+            &json!({"select": ["SQBR"], "cache": {"enabled": true}}),
+        );
+        let models = orders_models();
+        let expected = evaluate_split(&rest, &models)?;
+        std::fs::write(
+            project_dir
+                .path()
+                .join("target/rules-cache/bulk/native.json"),
+            b"{\"fingerprint\": \"truncated",
+        )
+        .map_err(|error| error.to_string())?;
+
+        let recovered = evaluate_split(&rest, &models)?;
+
+        assert_eq!(
+            recovered["cache_hits"], test_case.expected_hits,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            recovered["cache_misses"], test_case.expected_misses,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            recovered["faults"], expected["faults"],
+            "{}",
+            test_case.description
+        );
+    }
+    Ok(())
+}
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_sql_test_rules.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_sql_test_rules.rs
index a503f7d2e2..6a6d2d26bb 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests/test_sql_test_rules.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests/test_sql_test_rules.rs
@@ -243,14 +243,16 @@ fn given_sql_test_facts_when_evaluating_project_rules_then_returns_expected_faul
 }
 
 #[test]
-fn given_test_fact_change_when_evaluating_cached_project_rule_then_model_cache_is_invalidated()
+fn given_test_fact_change_when_evaluating_cached_project_rule_then_project_rule_reruns_over_cached_models()
 -> Result<(), String> {
     let project_dir = TempDir::new().map_err(|error| error.to_string())?;
     let test_cases = [test_types::SqlTestRulesCacheTestCase {
-        description: "test fact change invalidates model cache identity",
+        description: "test fact change re-evaluates the project rule and reuses the model",
         expected_first_misses: 1,
-        expected_second_hits: 0,
-        expected_second_misses: 1,
+        expected_second_hits: 1,
+        expected_second_misses: 0,
+        expected_first_codes: &["SQBRTEST104"],
+        expected_second_codes: &["SQBRTEST104"],
     }];
     for test_case in &test_cases {
         let first: Value = serde_json::from_str(&evaluate_json(
@@ -263,6 +265,14 @@ fn given_test_fact_change_when_evaluating_cached_project_rule_then_model_cache_i
                 Some("orders: keeps paid orders"),
             ))?)
             .map_err(|error| error.to_string())?;
+        let codes = |response: &Value| -> Vec<String> {
+            response["faults"]
+                .as_array()
+                .into_iter()
+                .flatten()
+                .filter_map(|fault| fault["code"].as_str().map(str::to_owned))
+                .collect()
+        };
         assert_eq!(
             first["cache_misses"], test_case.expected_first_misses,
             "{}",
@@ -278,6 +288,23 @@ fn given_test_fact_change_when_evaluating_cached_project_rule_then_model_cache_i
             "{}",
             test_case.description
         );
+        assert_eq!(
+            codes(&first),
+            test_case.expected_first_codes,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            codes(&second),
+            test_case.expected_second_codes,
+            "{}",
+            test_case.description
+        );
+        assert_ne!(
+            first["faults"], second["faults"],
+            "{}",
+            test_case.description
+        );
     }
     Ok(())
 }
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs
index f2410357e2..0d348f2d96 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs
@@ -157,6 +157,8 @@ pub(crate) struct SqlTestRulesCacheTestCase {
     pub(crate) expected_first_misses: u64,
     pub(crate) expected_second_hits: u64,
     pub(crate) expected_second_misses: u64,
+    pub(crate) expected_first_codes: &'static [&'static str],
+    pub(crate) expected_second_codes: &'static [&'static str],
 }
 
 pub(crate) struct EmptyInputTestRuleTestCase {
@@ -202,3 +204,39 @@ pub(crate) struct SqlScannerTestCase {
     pub(crate) expected_lint_site: Option<&'static str>,
     pub(crate) expected_reference_fast_path: bool,
 }
+
+pub(crate) struct IncrementalRulesTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) cache_enabled: bool,
+    pub(crate) compared_fields: &'static [&'static str],
+    pub(crate) expected_mismatched_fields: &'static [&'static str],
+}
+
+pub(crate) struct RuleScopeTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) expected_scopes: &'static [(&'static str, crate::models::RuleScope)],
+    pub(crate) expected_sql_scope: crate::models::RuleScope,
+}
+
+pub(crate) struct ScopedEvaluationTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) expected_codes: &'static [&'static str],
+}
+
+pub(crate) struct MalformedSplitRequestTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) inline_model_count: usize,
+    pub(crate) payload_count: usize,
+    pub(crate) expected_error: &'static str,
+}
+
+pub(crate) struct ModelCacheReuseTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) expected_hits: u64,
+    pub(crate) expected_misses: u64,
+}
+
+pub(crate) struct CachePruningTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) expected_paths: &'static [&'static str],
+}
diff --git a/crates/sqlbuild-rules-native/src/models.rs b/crates/sqlbuild-rules-native/src/models.rs
index e45f0528f6..74d9eccebe 100644
--- a/crates/sqlbuild-rules-native/src/models.rs
+++ b/crates/sqlbuild-rules-native/src/models.rs
@@ -567,6 +567,8 @@ pub(crate) struct Model {
     pub targeting_test_count: u32,
     #[serde(skip_deserializing)]
     pub empty_input_only_test_count: u32,
+    #[serde(skip)]
+    pub payload_digest: Option<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -727,7 +729,6 @@ pub(crate) struct EvaluateRequest {
     pub initial_findings: Vec<Fault>,
     pub defer_suppressions: bool,
     pub custom_rules: Vec<CustomRule>,
-    pub project_fingerprint: Option<String>,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -764,7 +765,6 @@ impl Default for EvaluateRequest {
             initial_findings: vec![],
             defer_suppressions: false,
             custom_rules: vec![],
-            project_fingerprint: None,
         }
     }
 }
@@ -807,6 +807,26 @@ impl RuleGuidance {
     }
 }
 
+/// Inputs that can change a rule's findings; only `Model` findings are cached per model.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum RuleScope {
+    Model,
+    Project,
+    ModelAndProject,
+    SqlLint,
+}
+
+impl RuleScope {
+    pub(crate) fn evaluates_models(self) -> bool {
+        matches!(self, Self::Model | Self::ModelAndProject)
+    }
+
+    pub(crate) fn evaluates_project(self) -> bool {
+        matches!(self, Self::Project | Self::ModelAndProject)
+    }
+}
+
 #[derive(Clone, Debug, Serialize)]
 pub(crate) struct RuleMetadata {
     pub code: String,
@@ -819,6 +839,7 @@ pub(crate) struct RuleMetadata {
     pub enabled_by_default: bool,
     pub project_wide: bool,
     pub custom: bool,
+    pub scope: RuleScope,
 }
 
 impl fensu_policy::policy::types::PolicyRule for RuleMetadata {
diff --git a/crates/sqlbuild-rules-native/src/rules/_helpers/catalogue.rs b/crates/sqlbuild-rules-native/src/rules/_helpers/catalogue.rs
index 5d5741fb04..2173f679a0 100644
--- a/crates/sqlbuild-rules-native/src/rules/_helpers/catalogue.rs
+++ b/crates/sqlbuild-rules-native/src/rules/_helpers/catalogue.rs
@@ -1,6 +1,6 @@
 use crate::constants::{API_VERSION, NATIVE_BUILD_IDENTITY};
 use crate::models::RulesCodeGrammar;
-use crate::models::{CustomRule, ResolveRulesRequest, RuleGuidance, RuleMetadata};
+use crate::models::{CustomRule, ResolveRulesRequest, RuleGuidance, RuleMetadata, RuleScope};
 use fensu_policy::policy::errors::PolicyError;
 use fensu_policy::policy::main::resolve_policy::resolve_policy;
 use fensu_policy::policy::models::PolicySelectors;
@@ -10,7 +10,7 @@ use sha2::{Digest, Sha256};
 const CUSTOM_RULE_COVERAGE_CODE: &str = "SQBRTEST301";
 
 macro_rules! rule {
-    ($code:expr, $family:expr, $slug:expr, $message:expr, $remediation:expr $(,)?) => {
+    ($code:expr, $scope:expr, $family:expr, $slug:expr, $message:expr, $remediation:expr $(,)?) => {
         RuleMetadata {
             code: $code.into(),
             family: $family.into(),
@@ -42,6 +42,7 @@ macro_rules! rule {
                     | "SQBRTEST301"
             ),
             custom: false,
+            scope: $scope,
         }
     };
 }
@@ -50,6 +51,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
     let mut rules = vec![
         rule!(
             "SQBRMODEL101",
+            RuleScope::Model,
             "structure",
             "dependency-import-ctes",
             "dependencies must be isolated in import CTEs",
@@ -57,6 +59,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRMODEL102",
+            RuleScope::Model,
             "structure",
             "select-star-discipline",
             "SELECT * is allowed only inside dependency import CTEs",
@@ -64,6 +67,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRMODEL103",
+            RuleScope::Model,
             "structure",
             "view-marker",
             "view materialization and model v marker must agree",
@@ -71,6 +75,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRMODEL104",
+            RuleScope::Model,
             "structure",
             "central-model-schema",
             "model schemas must come from central path defaults",
@@ -78,6 +83,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRGRAPH101",
+            RuleScope::Project,
             "graph",
             "forward-only-references",
             "model dependencies must flow forward through the layer order",
@@ -85,6 +91,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRGRAPH102",
+            RuleScope::Model,
             "graph",
             "declared-table-references",
             "table dependencies must use __ref or __source",
@@ -92,6 +99,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT101",
+            RuleScope::Model,
             "repository",
             "model-name-grammar",
             "model names must use the closed rule layer grammar",
@@ -99,6 +107,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT102",
+            RuleScope::Model,
             "repository",
             "folder-layer",
             "model layer segments must be contiguous in their folders",
@@ -106,6 +115,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT103",
+            RuleScope::Model,
             "repository",
             "source-token-policy",
             "model source suffixes must use approved, current tokens",
@@ -113,6 +123,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT104",
+            RuleScope::Model,
             "repository",
             "reference-name-policy",
             "referenced model identifiers must follow rule naming grammar",
@@ -120,6 +131,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT105",
+            RuleScope::Model,
             "repository",
             "name-folder-alignment",
             "model name layer must match its canonical folder",
@@ -127,6 +139,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT106",
+            RuleScope::Model,
             "repository",
             "name-schema-alignment",
             "model name layer must match its effective logical schema",
@@ -134,6 +147,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT101",
+            RuleScope::Model,
             "contracts",
             "contract-enforced-required",
             "models must declare an enforced output contract",
@@ -141,6 +155,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT201",
+            RuleScope::Project,
             "repository",
             "domain-level-layout",
             "models must resolve to one configured domain root and level",
@@ -148,6 +163,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT202",
+            RuleScope::Project,
             "repository",
             "owner-leaf-or-branch",
             "model owners must be either leaves or branches",
@@ -155,6 +171,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT203",
+            RuleScope::Project,
             "repository",
             "maximum-subdomain-depth",
             "model ownership must stay within the configured subdomain depth",
@@ -162,6 +179,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRPROJECT204",
+            RuleScope::Project,
             "repository",
             "shared-owner-prefix",
             "sibling owner names must not hide an implicit hierarchy",
@@ -169,6 +187,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT102",
+            RuleScope::Model,
             "contracts",
             "boolean-column-name",
             "boolean column names must have BOOLEAN types",
@@ -176,6 +195,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT103",
+            RuleScope::Model,
             "contracts",
             "timestamp-column-name",
             "timestamp column names must have timestamp types",
@@ -183,6 +203,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT104",
+            RuleScope::Model,
             "contracts",
             "date-column-name",
             "date column names must have DATE types",
@@ -190,6 +211,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT105",
+            RuleScope::Model,
             "contracts",
             "explicit-output-types",
             "contract outputs must establish their declared types explicitly",
@@ -197,6 +219,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRCONTRACT106",
+            RuleScope::Model,
             "contracts",
             "typed-contract-columns",
             "enforced contract columns must declare types",
@@ -204,6 +227,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION101",
+            RuleScope::Model,
             "declarations",
             "named-enum-decisions",
             "enum comparisons must use declared members and normalized operands",
@@ -211,6 +235,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION102",
+            RuleScope::Model,
             "declarations",
             "named-numeric-decisions",
             "non-canonical numeric comparisons must use constants",
@@ -218,6 +243,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION201",
+            RuleScope::Project,
             "declarations",
             "duplicate-enums",
             "identical enum domains must be consolidated",
@@ -225,6 +251,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION301",
+            RuleScope::Project,
             "declarations",
             "declaration-domain-placement",
             "public enum and constant files must live under a configured domain folder",
@@ -232,6 +259,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION302",
+            RuleScope::Project,
             "declarations",
             "declaration-container-shape",
             "declaration role containers must be flat or grouped",
@@ -239,6 +267,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION303",
+            RuleScope::Project,
             "declarations",
             "declaration-container-depth",
             "declaration role buckets must stay within the configured depth",
@@ -246,6 +275,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION304",
+            RuleScope::Project,
             "declarations",
             "declaration-container-capacity",
             "declaration role containers and buckets must remain bounded",
@@ -253,6 +283,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION305",
+            RuleScope::Project,
             "declarations",
             "declaration-bucket-name",
             "declaration role buckets must name a specific concern",
@@ -260,6 +291,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRDECLARATION306",
+            RuleScope::Project,
             "declarations",
             "declaration-container-prefix",
             "declaration filenames must not hide an obvious navigation bucket",
@@ -267,6 +299,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST101",
+            RuleScope::Project,
             "tests",
             "canonical-test-roots",
             "SQL unit tests and scenarios must use their compiler-owned canonical roots",
@@ -274,6 +307,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST102",
+            RuleScope::Project,
             "tests",
             "test-filename-grammar",
             "SQL test and scenario filenames must identify their subject and behavior",
@@ -281,6 +315,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST103",
+            RuleScope::Project,
             "tests",
             "semantic-test-mirroring",
             "SQL unit tests must mirror compiler-resolved resource ownership",
@@ -288,6 +323,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST104",
+            RuleScope::Project,
             "tests",
             "structured-test-name",
             "every SQL unit-test block must have a target-aware subject__expected_behavior name",
@@ -295,6 +331,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST201",
+            RuleScope::ModelAndProject,
             "tests",
             "minimum-audits",
             "non-passthrough models must declare the configured minimum audits",
@@ -303,6 +340,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         minimum_tests_rule(),
         rule!(
             "SQBRTEST203",
+            RuleScope::Project,
             "tests",
             "empty-input-only-test",
             "SQL unit tests must not only prove that empty inputs produce no rows",
@@ -310,6 +348,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
         ),
         rule!(
             "SQBRTEST301",
+            RuleScope::Project,
             "tests",
             "custom-rule-test-coverage",
             "selected custom rules must have public-harness test cases",
@@ -330,6 +369,7 @@ pub(crate) fn catalogue() -> Vec<RuleMetadata> {
                 enabled_by_default: true,
                 project_wide: false,
                 custom: false,
+                scope: RuleScope::SqlLint,
             }),
     );
     rules.sort_by(|a, b| a.code.cmp(&b.code));
@@ -358,6 +398,7 @@ fn minimum_tests_rule() -> RuleMetadata {
         enabled_by_default: true,
         project_wide: false,
         custom: false,
+        scope: RuleScope::ModelAndProject,
     }
 }
 
@@ -379,6 +420,7 @@ pub(crate) fn with_custom(custom: &[CustomRule]) -> Result<Vec<RuleMetadata>, St
             enabled_by_default: item.enabled_by_default,
             project_wide: item.project_wide,
             custom: true,
+            scope: RuleScope::Project,
         });
     }
     result.sort_by(|a, b| a.code.cmp(&b.code));
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index e132ef6d4d..74416002de 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -13,6 +13,9 @@ class LintPreparationRequest(TypedDict):
 def analyze_column_references_json(request_json: str) -> str: ...
 def catalogue_json() -> str: ...
 def evaluate_json(request_json: str) -> str: ...
+def evaluate_rules_parts(
+    request_json: bytes, model_jsons: list[bytes], model_digests: list[str]
+) -> str: ...
 def finalize_rule_findings_json(request_json: str) -> str: ...
 def run_custom_host_json(spec_json: str) -> str: ...
 def format_sql_json(request_json: str) -> str: ...
diff --git a/src/sqlbuild/rule_engine/_helpers/engine/native.py b/src/sqlbuild/rule_engine/_helpers/engine/native.py
index 9d31604b6c..1022b910a5 100644
--- a/src/sqlbuild/rule_engine/_helpers/engine/native.py
+++ b/src/sqlbuild/rule_engine/_helpers/engine/native.py
@@ -4,6 +4,7 @@
 
 import inspect
 import json
+from collections.abc import Iterator
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
 from dataclasses import asdict, dataclass, replace
 from decimal import Decimal
@@ -34,6 +35,7 @@
 )
 from sqlbuild.rule_engine._helpers.engine.custom_rules import evaluate_custom_rules_cached
 from sqlbuild.rule_engine._helpers.run.native_memo import (
+    native_payload_digests,
     native_request_identity,
     read_native_response,
     write_native_response,
@@ -54,11 +56,18 @@
 from sqlbuild.sql_values.types import SqlValueKind
 
 
+@dataclass(frozen=True)
+class _EncodedNativeRequest:
+    request_json: bytes
+    model_jsons: list[bytes]
+    model_digests: list[str]
+
+
 @dataclass(frozen=True)
 class _PreparedNativeRequest:
     identity: str | None
     reused: str | None
-    request_text: str | None
+    encoded: _EncodedNativeRequest | None
 
 
 def evaluate_native(
@@ -82,15 +91,13 @@ def evaluate_native(
         else custom_payloads
     )
     prepared: _PreparedNativeRequest = _prepare_native_request(
-        request_json=_serialise_native_request(
-            _native_request(
-                project=project,
-                config=config,
-                project_dir=project_dir,
-                dialect=dialect,
-                initial_findings=initial_findings,
-                custom_payloads=payloads,
-            )
+        encoded=_encode_native_request(
+            project=project,
+            config=config,
+            project_dir=project_dir,
+            dialect=dialect,
+            initial_findings=initial_findings,
+            custom_payloads=payloads,
         ),
         project_dir=project_dir,
         cache_enabled=config.cache.enabled,
@@ -154,6 +161,42 @@ def evaluate_native(
     )
 
 
+def _encode_native_request(
+    *,
+    project: CompiledProject,
+    config: RulesConfig,
+    project_dir: Path,
+    dialect: str,
+    initial_findings: tuple[Finding, ...],
+    custom_payloads: list[dict[str, object]],
+) -> _EncodedNativeRequest:
+    """Encode each model separately so its digest keys cached findings without a second pass."""
+
+    include_type_proof: bool = any(
+        _rule_selected(config=config, code=code) for code in TYPE_PROOF_RULE_CODES
+    )
+    model_jsons: list[bytes] = [
+        _serialise_native_request(payload)
+        for payload in _model_payloads(
+            project=project, dialect=dialect, include_type_proof=include_type_proof
+        )
+    ]
+    return _EncodedNativeRequest(
+        request_json=_serialise_native_request(
+            _native_request(
+                project=project,
+                config=config,
+                project_dir=project_dir,
+                dialect=dialect,
+                initial_findings=initial_findings,
+                custom_payloads=custom_payloads,
+            )
+        ),
+        model_jsons=model_jsons,
+        model_digests=native_payload_digests(model_jsons),
+    )
+
+
 def _native_request(
     *,
     project: CompiledProject,
@@ -168,13 +211,6 @@ def _native_request(
         "project_dir": str(project_dir.resolve()),
         "dialect": dialect,
         "config": _config_payload(config),
-        "models": _model_payloads(
-            project=project,
-            dialect=dialect,
-            include_type_proof=any(
-                _rule_selected(config=config, code=code) for code in TYPE_PROOF_RULE_CODES
-            ),
-        ),
         "sql_tests": _sql_test_payloads(project),
         "sql_scenarios": _sql_scenario_payloads(project),
         "public_enums": [
@@ -238,7 +274,7 @@ def _selected_custom_rules(
 
 
 def _serialise_native_request(request: dict[str, object]) -> bytes:
-    """Encode the native request, reporting values the encoder rejects as rules errors."""
+    """Encode one native request part, reporting values the encoder rejects as rules errors."""
 
     try:
         return orjson.dumps(request, option=orjson.OPT_SORT_KEYS, default=str)
@@ -247,11 +283,17 @@ def _serialise_native_request(request: dict[str, object]) -> bytes:
 
 
 def _prepare_native_request(
-    *, request_json: bytes, project_dir: Path, cache_enabled: bool
+    *, encoded: _EncodedNativeRequest, project_dir: Path, cache_enabled: bool
 ) -> _PreparedNativeRequest:
-    """Resolve the memo and decode once; the caller's bytes die when this returns."""
+    """Resolve the memo; a reused response releases the encoded request immediately."""
 
-    identity: str | None = native_request_identity(request_json) if cache_enabled else None
+    identity: str | None = (
+        native_request_identity(
+            request_json=encoded.request_json, model_digests=encoded.model_digests
+        )
+        if cache_enabled
+        else None
+    )
     reused: str | None = (
         read_native_response(project_dir=project_dir, identity=identity)
         if identity is not None
@@ -260,14 +302,17 @@ def _prepare_native_request(
     return _PreparedNativeRequest(
         identity=identity,
         reused=reused,
-        request_text=request_json.decode() if reused is None else None,
+        encoded=encoded if reused is None else None,
     )
 
 
 def _evaluate_request(*, prepared: _PreparedNativeRequest, project_dir: Path) -> str:
     if prepared.reused is not None:
         return prepared.reused
-    response: str = _native.evaluate_json(cast(str, prepared.request_text))
+    encoded: _EncodedNativeRequest = cast(_EncodedNativeRequest, prepared.encoded)
+    response: str = _native.evaluate_rules_parts(
+        encoded.request_json, encoded.model_jsons, encoded.model_digests
+    )
     if prepared.identity is not None:
         write_native_response(
             project_dir=project_dir, identity=prepared.identity, response=response
@@ -414,7 +459,7 @@ def _rule_selected(*, config: RulesConfig, code: str) -> bool:
 
 def _model_payloads(
     *, project: CompiledProject, dialect: str, include_type_proof: bool
-) -> list[dict[str, object]]:
+) -> Iterator[dict[str, object]]:
     audit_counts: dict[str, int] = {}
     for audit in project.audits:
         if audit.attached_target_name is not None:
@@ -427,7 +472,7 @@ def _model_payloads(
             continue
         for name in test.target_model_names:
             test_counts[name] = test_counts.get(name, 0) + 1
-    return [
+    return (
         dict(
             _model_payload(
                 model=model,
@@ -440,7 +485,7 @@ def _model_payloads(
             or model.config.values.get("sql_analysis") is False,
         )
         for model in project.models
-    ]
+    )
 
 
 def _sql_test_payloads(project: CompiledProject) -> list[dict[str, object]]:
diff --git a/src/sqlbuild/rule_engine/_helpers/run/native_memo.py b/src/sqlbuild/rule_engine/_helpers/run/native_memo.py
index e11cf9dc2b..4620c174fe 100644
--- a/src/sqlbuild/rule_engine/_helpers/run/native_memo.py
+++ b/src/sqlbuild/rule_engine/_helpers/run/native_memo.py
@@ -1,4 +1,4 @@
-"""Reuse the complete built-in rules response for a byte-identical native request."""
+"""Reuse the complete built-in rules response for an identical native request."""
 
 from __future__ import annotations
 
@@ -14,15 +14,24 @@
 from sqlbuild.rule_engine.constants import NATIVE_RULES_MEMO_FILE, NATIVE_RULES_MEMO_VERSION
 
 
-def native_request_identity(request_json: bytes) -> str:
-    """Digest one serialized native request together with the installed code identity."""
+def native_payload_digests(payloads: list[bytes]) -> list[str]:
+    """Digest each exact encoded model payload; the digest keys that model's cached findings."""
 
-    digest: Any = hashlib.sha256()
+    return [hashlib.blake2b(payload, digest_size=32).hexdigest() for payload in payloads]
+
+
+def native_request_identity(*, request_json: bytes, model_digests: list[str]) -> str:
+    """Digest the model-free request plus ordered digests of each exact model payload."""
+
+    digest: Any = hashlib.blake2b(digest_size=32)
     digest.update(NATIVE_RULES_MEMO_VERSION.encode())
     digest.update(b"\0")
     digest.update(compiled_code_identity().encode())
     digest.update(b"\0")
     digest.update(request_json)
+    digest.update(b"\0")
+    digest.update(len(model_digests).to_bytes(8, "little"))
+    digest.update("\0".join(model_digests).encode())
     return digest.hexdigest()
 
 
diff --git a/src/sqlbuild/rule_engine/_helpers/run/rules.py b/src/sqlbuild/rule_engine/_helpers/run/rules.py
index e93d3b5bba..b2962166c9 100644
--- a/src/sqlbuild/rule_engine/_helpers/run/rules.py
+++ b/src/sqlbuild/rule_engine/_helpers/run/rules.py
@@ -740,7 +740,7 @@ def _sql_model_rule_identities(
 
 
 def _sql_rule_identity(*, model: CompiledModel, codes: tuple[str, ...], dialect: str) -> str:
-    digest: Any = hashlib.sha256()
+    digest: Any = hashlib.blake2b(digest_size=32)
     digest.update(_SQL_RULE_CACHE_VERSION.encode())
     digest.update(_RULES_BUILD_IDENTITY.encode())
     digest.update(dialect.encode())
@@ -776,7 +776,7 @@ def _sql_file_rule_identity(
         or file_path in sql_expansions
     ):
         return None
-    digest: Any = hashlib.sha256()
+    digest: Any = hashlib.blake2b(digest_size=32)
     for value in (
         _SQL_RULE_CACHE_VERSION,
         "file",
diff --git a/src/sqlbuild/rule_engine/constants.py b/src/sqlbuild/rule_engine/constants.py
index bc80715cb5..306024b046 100644
--- a/src/sqlbuild/rule_engine/constants.py
+++ b/src/sqlbuild/rule_engine/constants.py
@@ -13,7 +13,7 @@
 CUSTOM_RULE_PROJECT_SUBJECT: str = ""
 CUSTOM_RULES_CACHE_VERSION: str = "custom-rules-reads-v1"
 CUSTOM_RULES_CACHE_FILE: str = "target/rules-cache/bulk/custom-rules.json"
-NATIVE_RULES_MEMO_VERSION: str = "native-rules-response-v1"
+NATIVE_RULES_MEMO_VERSION: str = "native-rules-response-v2"
 NATIVE_RULES_MEMO_FILE: str = "target/rules-cache/bulk/native-response.json"
 CUSTOM_HOST_INPUT_TUPLE_SIZE: int = 2
 CUSTOM_HOST_MODULE: str = "sqlbuild.rule_engine._helpers.host.custom_host"
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py
index a6ec47889b..0c2866e28f 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py
@@ -212,3 +212,13 @@ class SplitHostStateCase:
     selected_rules: tuple[str, ...]
     files: tuple[tuple[str, str], ...]
     expected_findings: tuple[tuple[str, str], ...]
+
+
+@dataclass(frozen=True)
+class RulesEditChainCase:
+    """A chain of authored edits, each compiled warm and compared with a cache-free compile."""
+
+    description: str
+    edits: tuple[RulesCacheEditCase, ...]
+    expected_cold_codes: tuple[str, ...]
+    expected_exit_code: int = 1
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py
index f961cc6a6c..c6b8081c7a 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py
@@ -381,6 +381,36 @@ def flagged_values(*, model: Model, ctx: RuleContext) -> list[Finding]:
 }
 
 
+BUILT_IN_RULES_PROJECT: dict[str, str] = {
+    "sqlbuild_project.toml": (
+        'name = "orders"\nadapter = "duckdb"\n\n'
+        '[connection]\ndatabase = "warehouse.duckdb"\n\n'
+        '[rules]\nselect = ["SQBR"]\ndomains = ["commerce"]\n\n'
+        "[rules.thresholds]\nmin_tests_per_model = 1\n"
+    ),
+    "models/commerce/staging/commerce__stg__orders.sql": (
+        'MODEL (description "Staged orders");\n\n'
+        "SELECT CAST(1 AS INTEGER) AS order_id, CAST(10.5 AS DOUBLE) AS amount, "
+        "'paid' AS status\n"
+    ),
+    "models/commerce/mart/commerce__mart__order_totals.sql": (
+        'MODEL (description "Order totals");\n\n'
+        'WITH orders AS (\n  SELECT * FROM __ref("commerce__stg__orders")\n)\n'
+        "SELECT order_id, amount * 2 AS doubled_amount FROM orders "
+        "WHERE status = 'paid' AND amount > 7\n"
+    ),
+    "models/commerce/mart/commerce__mart__customers.sql": (
+        'MODEL (description "Customers");\n\nSELECT customer_id FROM warehouse.raw.customers\n'
+    ),
+    "tests/unit/test_commerce__mart__order_totals.sql": (
+        "TEST ();\n\nWITH\n__ref__commerce__stg__orders AS (\n"
+        "  SELECT 1 AS order_id, 10.0 AS amount, 'paid' AS status\n),\n"
+        "__expected__commerce__mart__order_totals AS (\n"
+        "  SELECT 1 AS order_id, 20.0 AS doubled_amount\n)\nSELECT 1\n"
+    ),
+}
+
+
 def write_project_files(*, project_dir: Path, files: dict[str, str]) -> None:
     """Write authored project files beneath one project directory."""
 
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_rules_incremental_sequence.py b/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_rules_incremental_sequence.py
new file mode 100644
index 0000000000..b9ec54381c
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/rules/test_rules_incremental_sequence.py
@@ -0,0 +1,179 @@
+"""Incremental built-in Rules must equal a cache-free compile across a chain of edits."""
+
+import shutil
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
+    move_project_file,
+    replace_project_text,
+    write_project_file,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.rules._test_types import (
+    RulesCacheEditCase,
+    RulesEditChainCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.rules.helpers import (
+    BUILT_IN_RULES_PROJECT,
+    rule_cache_counts,
+    rules_compile_outcome,
+    write_project_files,
+)
+
+_CONFIG: str = "sqlbuild_project.toml"
+_ORDER_TOTALS: str = "models/commerce/mart/commerce__mart__order_totals.sql"
+_PRODUCTS: str = "models/commerce/mart/commerce__mart__products.sql"
+
+EDIT_SEQUENCE: tuple[RulesCacheEditCase, ...] = (
+    RulesCacheEditCase(
+        "model_sql_edited",
+        lambda root: replace_project_text(root, _ORDER_TOTALS, "amount > 7", "amount > 9"),
+    ),
+    RulesCacheEditCase(
+        "model_added",
+        lambda root: write_project_file(
+            root,
+            _PRODUCTS,
+            'MODEL (description "Products");\n\nSELECT * FROM __ref("commerce__stg__orders")\n',
+        ),
+    ),
+    RulesCacheEditCase(
+        "model_removed",
+        lambda root: (root / "models/commerce/mart/commerce__mart__customers.sql").unlink(),
+    ),
+    RulesCacheEditCase(
+        "model_renamed",
+        lambda root: move_project_file(
+            root, _PRODUCTS, "models/commerce/mart/commerce__mart__catalog_products.sql"
+        ),
+    ),
+    RulesCacheEditCase(
+        "threshold_changed",
+        lambda root: replace_project_text(
+            root, _CONFIG, "min_tests_per_model = 1", "min_tests_per_model = 0"
+        ),
+    ),
+    RulesCacheEditCase(
+        "scoped_ignore_added",
+        lambda root: replace_project_text(
+            root,
+            _CONFIG,
+            "[rules.thresholds]",
+            '[[rules.rule_ignores]]\nrules = ["SQBRTEST201"]\n'
+            'paths = ["models/commerce/staging/commerce__stg__orders.sql"]\n'
+            'reason = "Audits arrive with the source contract"\n\n[rules.thresholds]',
+        ),
+    ),
+    RulesCacheEditCase(
+        "exception_added",
+        lambda root: replace_project_text(
+            root,
+            _CONFIG,
+            "[rules.thresholds]",
+            f'[[rules.rule_exceptions]]\nrule = "SQBRCONTRACT101"\npath = "{_ORDER_TOTALS}"\n'
+            'reason = "Contract arrives with the next release"\n\n[rules.thresholds]',
+        ),
+    ),
+    RulesCacheEditCase(
+        "rules_toggled_off",
+        lambda root: replace_project_text(
+            root,
+            _CONFIG,
+            'domains = ["commerce"]',
+            'domains = ["commerce"]\nignore = ["SQBRDECLARATION", "SQBRGRAPH"]',
+        ),
+    ),
+    RulesCacheEditCase(
+        "dialect_changed",
+        lambda root: replace_project_text(
+            root,
+            _CONFIG,
+            'adapter = "duckdb"\n\n[connection]\ndatabase = "warehouse.duckdb"',
+            'adapter = "snowflake"\n\n[defaults]\ndatabase = "warehouse"\nschema = "analytics"',
+        ),
+    ),
+    RulesCacheEditCase(
+        "rules_toggled_on",
+        lambda root: replace_project_text(
+            root, _CONFIG, '\nignore = ["SQBRDECLARATION", "SQBRGRAPH"]', ""
+        ),
+    ),
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        RulesEditChainCase(
+            description="model_config_and_dialect_edits",
+            edits=EDIT_SEQUENCE,
+            expected_cold_codes=("SQBRGRAPH102", "SQBRTEST104"),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_warm_rules_cache_when_applying_edit_chain_then_every_compile_equals_uncached(
+    tmp_path: Path, test_case: RulesEditChainCase
+) -> None:
+    project_dir: Path = tmp_path / "orders"
+    write_project_files(project_dir=project_dir, files=BUILT_IN_RULES_PROJECT)
+    cold: tuple[int, str, str] = rules_compile_outcome(project_dir)
+    assert cold[0] == test_case.expected_exit_code
+    for code in test_case.expected_cold_codes:
+        assert f'"{code}"' in cold[1], code
+    oracle: tuple[int, str, str] = cold
+
+    for step, edit_case in enumerate(test_case.edits):
+        edit_case.edit(project_dir)
+        edited: tuple[int, str, str] = rules_compile_outcome(project_dir)
+        oracle_dir: Path = tmp_path / f"oracle_{step}"
+        shutil.copytree(project_dir, oracle_dir, ignore=shutil.ignore_patterns("target"))
+        oracle = rules_compile_outcome(oracle_dir, "--no-cache")
+
+        assert edited == oracle, edit_case.description
+        assert edited[0] == edit_case.expected_exit_code, edit_case.description
+
+    rewarmed: tuple[int, str, str] = rules_compile_outcome(project_dir)
+    assert rewarmed == oracle
+    assert rule_cache_counts(project_dir)[1] == 0
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        RulesCacheEditCase(
+            "corrupt_model_findings_cache",
+            lambda root: (root / "target/rules-cache/bulk/native.json").write_text(
+                '{"fingerprint": "trunc', encoding="utf-8"
+            ),
+        ),
+        RulesCacheEditCase(
+            "corrupt_response_memo",
+            lambda root: (root / "target/rules-cache/bulk/native-response.json").write_text(
+                '{"identity": ', encoding="utf-8"
+            ),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_corrupt_built_in_rules_cache_when_compiling_then_rules_fall_back_to_evaluation(
+    tmp_path: Path, test_case: RulesCacheEditCase
+) -> None:
+    project_dir: Path = tmp_path / "orders"
+    write_project_files(project_dir=project_dir, files=BUILT_IN_RULES_PROJECT)
+    cold: tuple[int, str, str] = rules_compile_outcome(project_dir)
+    replace_project_text(project_dir, _ORDER_TOTALS, "amount > 7", "amount > 9")
+    test_case.edit(project_dir)
+
+    recovered: tuple[int, str, str] = rules_compile_outcome(project_dir)
+    oracle_dir: Path = tmp_path / "oracle"
+    shutil.copytree(project_dir, oracle_dir, ignore=shutil.ignore_patterns("target"))
+    oracle: tuple[int, str, str] = rules_compile_outcome(oracle_dir, "--no-cache")
+
+    assert cold[0] == recovered[0] == test_case.expected_exit_code
+    assert recovered == oracle
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"]))
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/_test_types.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/_test_types.py
index d68bf921ab..75209e1cf2 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/_test_types.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/_test_types.py
@@ -228,3 +228,22 @@ class NativeRequestEncodeErrorTestCase:
     description: str
     rejected_value: object
     expected_message: str
+
+
+@dataclass(frozen=True)
+class IncrementalRulesModeTestCase:
+    """Evaluation options under which incremental results must equal full evaluation."""
+
+    description: str
+    defer_suppressions: bool
+    expected_observed_codes: tuple[str, ...]
+    initial_codes: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class IncrementalCacheCountsTestCase:
+    """One warm evaluation after editing a single model."""
+
+    description: str
+    expected_hits: int
+    expected_misses: int
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
index 600de06dda..175dc54088 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
@@ -2,7 +2,7 @@
 
 import json
 from collections.abc import Callable
-from dataclasses import replace
+from dataclasses import dataclass, field, replace
 from operator import attrgetter
 from pathlib import Path
 from typing import Any, cast
@@ -28,7 +28,10 @@
 from sqlbuild.rule_engine.main._evaluate import evaluate
 from sqlbuild.rule_engine.models import (
     CustomRulesOutcome,
+    Finding,
     Rule,
+    RuleExemption,
+    RuleIgnore,
     RulesCacheConfig,
     RulesConfig,
     RulesResult,
@@ -58,8 +61,11 @@ def captured_native_request(
 
     captured: dict[str, Any] = {}
 
-    def evaluate_json(request_json: str) -> str:
+    def evaluate_rules_parts(
+        request_json: bytes, model_jsons: list[bytes], model_digests: list[str]
+    ) -> str:
         captured.update(cast(dict[str, Any], json.loads(request_json)))
+        captured["models"] = [json.loads(model_json) for model_json in model_jsons]
         return json.dumps(
             {
                 "version": 1,
@@ -71,7 +77,7 @@ def evaluate_json(request_json: str) -> str:
             }
         )
 
-    monkeypatch.setattr(native._native, "evaluate_json", evaluate_json)
+    monkeypatch.setattr(native._native, "evaluate_rules_parts", evaluate_rules_parts)
     native.evaluate_native(
         project=project,
         config=RulesConfig(cache=RulesCacheConfig(enabled=False)),
@@ -303,15 +309,97 @@ def evaluate_contract_rule(
     return evaluate(project=project, config=config, project_dir=project_dir)
 
 
-def record_native_evaluations(*, monkeypatch: pytest.MonkeyPatch) -> list[str]:
+def record_native_evaluations(*, monkeypatch: pytest.MonkeyPatch) -> list[bytes]:
     """Count every request that reaches the native built-in rules engine."""
 
-    requests: list[str] = []
-    evaluate_json: Callable[[str], str] = native._native.evaluate_json
+    requests: list[bytes] = []
+    evaluate_rules_parts: Callable[[bytes, list[bytes], list[str]], str] = (
+        native._native.evaluate_rules_parts
+    )
 
-    def recording(request_json: str) -> str:
+    def recording(request_json: bytes, model_jsons: list[bytes], model_digests: list[str]) -> str:
         requests.append(request_json)
-        return evaluate_json(request_json)
+        return evaluate_rules_parts(request_json, model_jsons, model_digests)
 
-    monkeypatch.setattr(native._native, "evaluate_json", recording)
+    monkeypatch.setattr(native._native, "evaluate_rules_parts", recording)
     return requests
+
+
+@dataclass(frozen=True)
+class IncrementalModelSpec:
+    """One synthetic compiled model in an incremental built-in rules step."""
+
+    name: str
+    relative_path: str
+    sql: str
+    config_values: dict[str, object] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class IncrementalRulesStep:
+    """One project and configuration state evaluated after the previous step."""
+
+    description: str
+    models: tuple[IncrementalModelSpec, ...]
+    select: tuple[str, ...] = ("SQBR",)
+    ignore: tuple[str, ...] = ()
+    thresholds: dict[str, int] = field(default_factory=dict)
+    rule_ignores: tuple[RuleIgnore, ...] = ()
+    rule_exceptions: tuple[RuleExemption, ...] = ()
+    dialect: str = "duckdb"
+
+
+def evaluate_incremental_step(
+    *,
+    step: IncrementalRulesStep,
+    project_dir: Path,
+    cache_enabled: bool,
+    defer_suppressions: bool,
+    initial_codes: tuple[str, ...] = (),
+) -> RulesResult:
+    """Write the step's model files and evaluate built-in rules natively over its models."""
+
+    for model in step.models:
+        path: Path = project_dir / model.relative_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(model.sql, encoding="utf-8")
+    config: RulesConfig = RulesConfig(
+        select=step.select,
+        ignore=step.ignore,
+        thresholds=step.thresholds,
+        rule_ignores=step.rule_ignores,
+        rule_exceptions=step.rule_exceptions,
+        cache=RulesCacheConfig(enabled=cache_enabled),
+    )
+    return native.evaluate_native(
+        project=_incremental_project(step.models),
+        config=config,
+        project_dir=project_dir,
+        catalogue=build_catalogue(config=config, project_dir=project_dir, include_custom=False),
+        dialect=step.dialect,
+        initial_findings=tuple(
+            Finding(
+                code=code,
+                path=Path(step.models[0].relative_path),
+                line=1,
+                column=1,
+                message=f"{code} reported before built-in rules",
+                remediation="Review the authored SQL.",
+            )
+            for code in initial_codes
+        ),
+        defer_suppressions=defer_suppressions,
+    )
+
+
+def _incremental_project(models: tuple[IncrementalModelSpec, ...]) -> CompiledProject:
+    projects: list[CompiledProject] = [
+        build_project(
+            name=model.name,
+            relative_path=model.relative_path,
+            sql=model.sql,
+            config_values=model.config_values,
+        )
+        for model in models
+    ]
+    return replace(projects[0], models=tuple(project.models[0] for project in projects))
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_incremental_native_rules.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_incremental_native_rules.py
new file mode 100644
index 0000000000..85aa97e09d
--- /dev/null
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_incremental_native_rules.py
@@ -0,0 +1,162 @@
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from sqlbuild.rule_engine.models import RuleExemption, RuleIgnore, RulesResult
+from tests.unit.src.sqlbuild.rule_engine._helpers.engine._test_types import (
+    IncrementalCacheCountsTestCase,
+    IncrementalRulesModeTestCase,
+)
+from tests.unit.src.sqlbuild.rule_engine._helpers.engine.helpers import (
+    IncrementalModelSpec,
+    IncrementalRulesStep,
+    evaluate_incremental_step,
+)
+
+_ORDERS: IncrementalModelSpec = IncrementalModelSpec(
+    name="commerce__mart__orders",
+    relative_path="models/commerce/mart/commerce__mart__orders.sql",
+    sql="WITH orders AS (SELECT * FROM source_orders) SELECT order_id FROM orders WHERE amount > 7",
+    config_values={"materialized": "table"},
+)
+_CUSTOMERS: IncrementalModelSpec = IncrementalModelSpec(
+    name="commerce__mart__customers",
+    relative_path="models/commerce/mart/commerce__mart__customers.sql",
+    sql="SELECT customer_id FROM warehouse.raw.customers",
+)
+_PRODUCTS: IncrementalModelSpec = IncrementalModelSpec(
+    name="commerce__stg__products",
+    relative_path="models/commerce/staging/commerce__stg__products.sql",
+    sql="SELECT * FROM warehouse.raw.products",
+)
+_EDITED_ORDERS: IncrementalModelSpec = replace(_ORDERS, sql=_ORDERS.sql.replace("> 7", "> 9"))
+_RENAMED_PRODUCTS: IncrementalModelSpec = replace(
+    _PRODUCTS,
+    name="commerce__stg__catalog_products",
+    relative_path="models/commerce/staging/commerce__stg__catalog_products.sql",
+)
+
+EDIT_SEQUENCE: tuple[IncrementalRulesStep, ...] = (
+    IncrementalRulesStep("cold", (_ORDERS, _CUSTOMERS)),
+    IncrementalRulesStep("warm", (_ORDERS, _CUSTOMERS)),
+    IncrementalRulesStep("model_edited", (_EDITED_ORDERS, _CUSTOMERS)),
+    IncrementalRulesStep("model_added", (_EDITED_ORDERS, _CUSTOMERS, _PRODUCTS)),
+    IncrementalRulesStep("model_removed", (_EDITED_ORDERS, _PRODUCTS)),
+    IncrementalRulesStep("model_renamed", (_EDITED_ORDERS, _RENAMED_PRODUCTS)),
+    IncrementalRulesStep(
+        "threshold_changed",
+        (_EDITED_ORDERS, _RENAMED_PRODUCTS),
+        thresholds={"min_audits_per_model": 2},
+    ),
+    IncrementalRulesStep(
+        "scoped_ignore_added",
+        (_EDITED_ORDERS, _RENAMED_PRODUCTS),
+        rule_ignores=(
+            RuleIgnore(
+                rules=("SQBRGRAPH102",),
+                paths=(_RENAMED_PRODUCTS.relative_path,),
+                reason="Raw products load is being retired",
+            ),
+        ),
+    ),
+    IncrementalRulesStep(
+        "exception_added",
+        (_EDITED_ORDERS, _RENAMED_PRODUCTS),
+        rule_exceptions=(
+            RuleExemption(
+                rule="SQBRCONTRACT101",
+                path=_EDITED_ORDERS.relative_path,
+                reason="Contract arrives with the next release",
+            ),
+        ),
+    ),
+    IncrementalRulesStep(
+        "rules_toggled_off", (_EDITED_ORDERS, _RENAMED_PRODUCTS), ignore=("SQBRMODEL",)
+    ),
+    IncrementalRulesStep(
+        "dialect_changed", (_EDITED_ORDERS, _RENAMED_PRODUCTS), dialect="snowflake"
+    ),
+    IncrementalRulesStep("rules_toggled_on", (_EDITED_ORDERS, _RENAMED_PRODUCTS)),
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        IncrementalRulesModeTestCase(
+            description="suppressions applied",
+            defer_suppressions=False,
+            expected_observed_codes=("SQBRGRAPH102", "SQBRCONTRACT101", "SQBRDECLARATION102"),
+        ),
+        IncrementalRulesModeTestCase(
+            description="suppressions deferred with initial findings",
+            defer_suppressions=True,
+            expected_observed_codes=("SQBRSQL034", "SQBRGRAPH102", "SQBRCONTRACT101"),
+            initial_codes=("SQBRSQL034",),
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_edit_sequence_when_evaluating_incrementally_then_each_result_equals_full_evaluation(
+    tmp_path: Path, test_case: IncrementalRulesModeTestCase
+) -> None:
+    cached_dir: Path = tmp_path / "cached"
+    cached_dir.mkdir()
+    observed: set[str] = set()
+
+    for index, step in enumerate(EDIT_SEQUENCE):
+        full_dir: Path = tmp_path / f"full_{index}"
+        full_dir.mkdir()
+        incremental: RulesResult = evaluate_incremental_step(
+            step=step,
+            project_dir=cached_dir,
+            cache_enabled=True,
+            defer_suppressions=test_case.defer_suppressions,
+            initial_codes=test_case.initial_codes,
+        )
+        full: RulesResult = evaluate_incremental_step(
+            step=step,
+            project_dir=full_dir,
+            cache_enabled=False,
+            defer_suppressions=test_case.defer_suppressions,
+            initial_codes=test_case.initial_codes,
+        )
+
+        assert incremental.findings == full.findings, step.description
+        assert incremental.evaluated_models == full.evaluated_models, step.description
+        observed.update(finding.code for finding in incremental.findings)
+
+    assert set(test_case.expected_observed_codes) <= observed
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    (
+        IncrementalCacheCountsTestCase(
+            description="one edited model of three", expected_hits=2, expected_misses=1
+        ),
+    ),
+    ids=lambda case: case.description,
+)
+def test_given_one_edited_model_when_evaluating_incrementally_then_only_that_model_misses(
+    tmp_path: Path, test_case: IncrementalCacheCountsTestCase
+) -> None:
+    cold: IncrementalRulesStep = IncrementalRulesStep("cold", (_ORDERS, _CUSTOMERS, _PRODUCTS))
+    edited: IncrementalRulesStep = replace(cold, models=(_EDITED_ORDERS, _CUSTOMERS, _PRODUCTS))
+    _ = evaluate_incremental_step(
+        step=cold, project_dir=tmp_path, cache_enabled=True, defer_suppressions=False
+    )
+
+    result: RulesResult = evaluate_incremental_step(
+        step=edited, project_dir=tmp_path, cache_enabled=True, defer_suppressions=False
+    )
+
+    assert (result.cache_hits, result.cache_misses) == (
+        test_case.expected_hits,
+        test_case.expected_misses,
+    )
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"]))
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
index 50b39076a0..33222f9143 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
@@ -75,7 +75,7 @@ def test_given_memoized_response_from_another_native_build_when_evaluating_then_
     cold: RulesResult = evaluate_contract_rule(
         config_values=config_values, project_dir=tmp_path, cache_enabled=True
     )
-    native_calls: list[str] = record_native_evaluations(monkeypatch=monkeypatch)
+    native_calls: list[bytes] = record_native_evaluations(monkeypatch=monkeypatch)
     monkeypatch.setattr(native_module, "BUILD_IDENTITY", f"{native_module.BUILD_IDENTITY}-rebuilt")
     rebuilt: RulesResult = evaluate_contract_rule(
         config_values=config_values, project_dir=tmp_path, cache_enabled=True
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
index 589e4707e2..36423ceba1 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
@@ -41,13 +41,13 @@ def failing_model_payloads(**_: object) -> NoReturn:
     def record_custom_start(**_: object) -> None:
         started.append("custom")
 
-    def record_native_call(_: str) -> str:
+    def record_native_call(*_: object) -> str:
         started.append("native")
         return "{}"
 
     monkeypatch.setattr(native, "_model_payloads", failing_model_payloads)
     monkeypatch.setattr(native, "start_custom_rules", record_custom_start)
-    monkeypatch.setattr(native._native, "evaluate_json", record_native_call)
+    monkeypatch.setattr(native._native, "evaluate_rules_parts", record_native_call)
 
     with pytest.raises(type(test_case.expected_error)) as raised:
         native.evaluate_native(
@@ -86,13 +86,13 @@ def oversized_model_payloads(**_: object) -> list[dict[str, object]]:
     def record_custom_start(**_: object) -> None:
         started.append("custom")
 
-    def record_native_call(_: str) -> str:
+    def record_native_call(*_: object) -> str:
         started.append("native")
         return "{}"
 
     monkeypatch.setattr(native, "_model_payloads", oversized_model_payloads)
     monkeypatch.setattr(native, "start_custom_rules", record_custom_start)
-    monkeypatch.setattr(native._native, "evaluate_json", record_native_call)
+    monkeypatch.setattr(native._native, "evaluate_rules_parts", record_native_call)
 
     with pytest.raises(native.RulesError, match=test_case.expected_message):
         native.evaluate_native(

From 750379896dc0b011bc76971adb267a5d6fc64955 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Sun, 4 Oct 2026 10:31:21 +0100
Subject: [PATCH 2/3] perf(rules): release encoded rules payloads before native
 evaluation

Decode the split native rules request into a native handle while the memo is
resolved, so the encoded model payloads are freed before built-in rules run
and the decoded request is dropped as soon as evaluation finishes.
---
 .../src/bindings/_helpers/functions.rs        | 34 +++++++++++---
 .../src/engine/_helpers/evaluation.rs         | 10 ++---
 .../src/engine/main/evaluate_parsed.rs        |  5 +++
 .../src/engine/main/evaluate_parts.rs         |  7 ---
 .../src/engine/main/mod.rs                    |  3 +-
 .../src/engine/main/parse_parts.rs            |  9 ++++
 .../src/engine/tests/helpers.rs               |  9 ++--
 .../engine/tests/test_incremental_rules.rs    |  4 +-
 crates/sqlbuild-rules-native/src/models.rs    |  9 ++++
 src/sqlbuild/_native.pyi                      |  8 +++-
 .../rule_engine/_helpers/engine/native.py     | 20 ++++++---
 .../rule_engine/_helpers/engine/helpers.py    | 26 ++++++-----
 .../_helpers/engine/test_native_memo.py       |  2 +-
 .../_helpers/engine/test_native_request.py    | 45 ++++++++++++++++++-
 14 files changed, 144 insertions(+), 47 deletions(-)
 create mode 100644 crates/sqlbuild-rules-native/src/engine/main/evaluate_parsed.rs
 delete mode 100644 crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
 create mode 100644 crates/sqlbuild-rules-native/src/engine/main/parse_parts.rs

diff --git a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
index a2ac476c74..1ff8a22929 100644
--- a/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
+++ b/crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs
@@ -8,9 +8,11 @@ use pyo3::{FromPyObject, pyfunction, wrap_pyfunction};
 
 use crate::configuration::main::load;
 use crate::constants::{API_VERSION, NATIVE_BUILD_IDENTITY};
-use crate::engine::main::{evaluate, evaluate_parts};
+use crate::engine::main::{evaluate, evaluate_parsed, parse_parts};
 use crate::models::CatalogueResponse;
+use crate::models::ParsedRulesRequest;
 use crate::rules::main::{catalogue, selected_codes};
+use std::sync::Mutex;
 
 const SKILL_OWNER: &str = "sqlbuild";
 use crate::bindings::_helpers::panics::compiler_guard;
@@ -74,15 +76,35 @@ fn evaluate_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
 }
 
 #[pyfunction]
-fn evaluate_rules_parts(
+fn parse_rules_parts(
     py: Python<'_>,
     request_json: Bound<'_, PyBytes>,
     model_jsons: Vec<Bound<'_, PyBytes>>,
     model_digests: Vec<String>,
-) -> PyResult<String> {
+) -> PyResult<ParsedRulesRequest> {
     let request: &[u8] = request_json.as_bytes();
     let models: Vec<&[u8]> = model_jsons.iter().map(|model| model.as_bytes()).collect();
-    py.compiler_detach(|| evaluate_parts::evaluate_parts(request, &models, &model_digests))
+    let parsed = py
+        .compiler_detach(|| parse_parts::parse_parts(request, &models, &model_digests))
+        .map_err(value_error)?;
+    Ok(ParsedRulesRequest {
+        request: Mutex::new(Some(parsed)),
+    })
+}
+
+#[pyfunction]
+fn evaluate_parsed_rules(
+    py: Python<'_>,
+    parsed: &Bound<'_, ParsedRulesRequest>,
+) -> PyResult<String> {
+    let request = parsed
+        .borrow()
+        .request
+        .lock()
+        .map_err(|_| value_error("rules request lock is poisoned"))?
+        .take()
+        .ok_or_else(|| value_error("rules request was already evaluated"))?;
+    py.compiler_detach(|| evaluate_parsed::evaluate_parsed(request))
         .map_err(value_error)
 }
 
@@ -478,7 +500,9 @@ fn skill_freshness(content: Option<&str>, input_fingerprint: &str) -> String {
 
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(evaluate_json, module)?)?;
-    module.add_function(wrap_pyfunction!(evaluate_rules_parts, module)?)?;
+    module.add_function(wrap_pyfunction!(parse_rules_parts, module)?)?;
+    module.add_function(wrap_pyfunction!(evaluate_parsed_rules, module)?)?;
+    module.add_class::<ParsedRulesRequest>()?;
     module.add_function(wrap_pyfunction!(finalize_rule_findings_json, module)?)?;
     module.add_function(wrap_pyfunction!(run_custom_host_json, module)?)?;
     module.add_function(wrap_pyfunction!(lint_sql_json, module)?)?;
diff --git a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
index e36c78b35d..6652778977 100644
--- a/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
+++ b/crates/sqlbuild-rules-native/src/engine/_helpers/evaluation.rs
@@ -184,12 +184,12 @@ pub(crate) fn evaluate_json(request_json: &str) -> Result<String, String> {
     evaluate_request(request)
 }
 
-/// Evaluate a request whose models arrive as separate payloads, each with its exact digest.
-pub(crate) fn evaluate_parts(
+/// Decode a request whose models arrive as separate payloads, each with its exact digest.
+pub(crate) fn parse_parts(
     request_json: &[u8],
     model_jsons: &[&[u8]],
     model_digests: &[String],
-) -> Result<String, String> {
+) -> Result<EvaluateRequest, String> {
     let mut request: EvaluateRequest = serde_json::from_slice(request_json)
         .map_err(|error| format!("invalid rules request: {error}"))?;
     if !request.models.is_empty() {
@@ -199,7 +199,7 @@ pub(crate) fn evaluate_parts(
         return Err("invalid rules request: every model payload needs one digest".to_owned());
     }
     request.models = parse_model_payloads(model_jsons, model_digests)?;
-    evaluate_request(request)
+    Ok(request)
 }
 
 fn parse_model_payloads(
@@ -269,7 +269,7 @@ fn model_rules_fingerprint(
     fingerprint::fingerprint(selected, &model_config, dialect)
 }
 
-fn evaluate_request(request: EvaluateRequest) -> Result<String, String> {
+pub(crate) fn evaluate_request(request: EvaluateRequest) -> Result<String, String> {
     if request.version != API_VERSION {
         return Err(format!(
             "unsupported rules native API version {}; expected {API_VERSION}",
diff --git a/crates/sqlbuild-rules-native/src/engine/main/evaluate_parsed.rs b/crates/sqlbuild-rules-native/src/engine/main/evaluate_parsed.rs
new file mode 100644
index 0000000000..3331c2a9be
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/engine/main/evaluate_parsed.rs
@@ -0,0 +1,5 @@
+use crate::models::EvaluateRequest;
+
+pub(crate) fn evaluate_parsed(request: EvaluateRequest) -> Result<String, String> {
+    crate::engine::_helpers::evaluation::evaluate_request(request)
+}
diff --git a/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs b/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
deleted file mode 100644
index aae707f28b..0000000000
--- a/crates/sqlbuild-rules-native/src/engine/main/evaluate_parts.rs
+++ /dev/null
@@ -1,7 +0,0 @@
-pub(crate) fn evaluate_parts(
-    request_json: &[u8],
-    model_jsons: &[&[u8]],
-    model_digests: &[String],
-) -> Result<String, String> {
-    crate::engine::_helpers::evaluation::evaluate_parts(request_json, model_jsons, model_digests)
-}
diff --git a/crates/sqlbuild-rules-native/src/engine/main/mod.rs b/crates/sqlbuild-rules-native/src/engine/main/mod.rs
index e35b0f83c0..097f97d0fc 100644
--- a/crates/sqlbuild-rules-native/src/engine/main/mod.rs
+++ b/crates/sqlbuild-rules-native/src/engine/main/mod.rs
@@ -1,4 +1,5 @@
 pub(crate) mod custom_host;
 pub(crate) mod evaluate;
-pub(crate) mod evaluate_parts;
+pub(crate) mod evaluate_parsed;
 pub(crate) mod finalize;
+pub(crate) mod parse_parts;
diff --git a/crates/sqlbuild-rules-native/src/engine/main/parse_parts.rs b/crates/sqlbuild-rules-native/src/engine/main/parse_parts.rs
new file mode 100644
index 0000000000..017234272a
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/engine/main/parse_parts.rs
@@ -0,0 +1,9 @@
+use crate::models::EvaluateRequest;
+
+pub(crate) fn parse_parts(
+    request_json: &[u8],
+    model_jsons: &[&[u8]],
+    model_digests: &[String],
+) -> Result<EvaluateRequest, String> {
+    crate::engine::_helpers::evaluation::parse_parts(request_json, model_jsons, model_digests)
+}
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs b/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
index d7fc43079e..0203fa36cc 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests/helpers.rs
@@ -3,7 +3,8 @@ use sha2::{Digest, Sha256};
 use tempfile::TempDir;
 
 use crate::engine::main::evaluate::evaluate_json;
-use crate::engine::main::evaluate_parts::evaluate_parts;
+use crate::engine::main::evaluate_parsed::evaluate_parsed;
+use crate::engine::main::parse_parts::parse_parts;
 
 pub(crate) fn request(project_dir: &TempDir, config: &Value) -> String {
     json!({
@@ -434,8 +435,10 @@ pub(crate) fn evaluate_split(rest: &Value, models: &[Value]) -> Result<Value, St
         .map(|payload| format!("{:x}", Sha256::digest(payload)))
         .collect();
     let rest_json = serde_json::to_vec(rest).map_err(|error| error.to_string())?;
-    serde_json::from_str(&evaluate_parts(&rest_json, &payloads, &digests)?)
-        .map_err(|error| error.to_string())
+    serde_json::from_str(&evaluate_parsed(parse_parts(
+        &rest_json, &payloads, &digests,
+    )?)?)
+    .map_err(|error| error.to_string())
 }
 
 pub(crate) fn evaluate_whole(rest: &Value, models: &[Value]) -> Result<Value, String> {
diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
index 6e58c7bfd6..6ed210ee3a 100644
--- a/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
+++ b/crates/sqlbuild-rules-native/src/engine/tests/test_incremental_rules.rs
@@ -3,7 +3,7 @@ use std::collections::BTreeMap;
 use serde_json::{Value, json};
 use tempfile::TempDir;
 
-use crate::engine::main::evaluate_parts::evaluate_parts;
+use crate::engine::main::parse_parts::parse_parts;
 use crate::engine::tests::helpers::{
     cached_paths, evaluate_split, evaluate_whole, orders_models, split_request, uncached,
     write_model_files,
@@ -189,7 +189,7 @@ fn given_malformed_split_request_when_evaluating_then_request_is_rejected() -> R
         let model = serde_json::to_vec(&orders_models()[0]).map_err(|error| error.to_string())?;
         let payloads: Vec<&[u8]> = vec![model.as_slice(); test_case.payload_count];
 
-        let error = evaluate_parts(&rest, &payloads, &[]);
+        let error = parse_parts(&rest, &payloads, &[]);
 
         assert!(
             error.is_err_and(|message| message.contains(test_case.expected_error)),
diff --git a/crates/sqlbuild-rules-native/src/models.rs b/crates/sqlbuild-rules-native/src/models.rs
index 74d9eccebe..ac7197deac 100644
--- a/crates/sqlbuild-rules-native/src/models.rs
+++ b/crates/sqlbuild-rules-native/src/models.rs
@@ -1,6 +1,8 @@
+use pyo3::pyclass;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::collections::BTreeMap;
+use std::sync::Mutex;
 
 use crate::constants::{
     API_VERSION, BUILT_IN_RULE_NAMESPACE, CUSTOM_RULE_NAMESPACE, RULE_CODE_NUMBER_LENGTH,
@@ -749,6 +751,13 @@ pub(crate) struct ResolveRulesRequest {
     pub custom_rules: Vec<CustomRule>,
 }
 
+/// A decoded rules request held natively so the caller can release its encoded payloads.
+#[pyclass(module = "sqlbuild._native")]
+#[derive(Debug)]
+pub(crate) struct ParsedRulesRequest {
+    pub(crate) request: Mutex<Option<EvaluateRequest>>,
+}
+
 impl Default for EvaluateRequest {
     fn default() -> Self {
         Self {
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index 74416002de..3eb370a343 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -10,12 +10,16 @@ class LintPreparationRequest(TypedDict):
     prior_sites: list[int]
     dialect: str
 
+class ParsedRulesRequest:
+    """Decoded built-in rules request held natively until it is evaluated once."""
+
 def analyze_column_references_json(request_json: str) -> str: ...
 def catalogue_json() -> str: ...
 def evaluate_json(request_json: str) -> str: ...
-def evaluate_rules_parts(
+def parse_rules_parts(
     request_json: bytes, model_jsons: list[bytes], model_digests: list[str]
-) -> str: ...
+) -> ParsedRulesRequest: ...
+def evaluate_parsed_rules(parsed: ParsedRulesRequest) -> str: ...
 def finalize_rule_findings_json(request_json: str) -> str: ...
 def run_custom_host_json(spec_json: str) -> str: ...
 def format_sql_json(request_json: str) -> str: ...
diff --git a/src/sqlbuild/rule_engine/_helpers/engine/native.py b/src/sqlbuild/rule_engine/_helpers/engine/native.py
index 1022b910a5..872498fa0f 100644
--- a/src/sqlbuild/rule_engine/_helpers/engine/native.py
+++ b/src/sqlbuild/rule_engine/_helpers/engine/native.py
@@ -67,7 +67,7 @@ class _EncodedNativeRequest:
 class _PreparedNativeRequest:
     identity: str | None
     reused: str | None
-    encoded: _EncodedNativeRequest | None
+    parsed: _native.ParsedRulesRequest | None
 
 
 def evaluate_native(
@@ -285,7 +285,7 @@ def _serialise_native_request(request: dict[str, object]) -> bytes:
 def _prepare_native_request(
     *, encoded: _EncodedNativeRequest, project_dir: Path, cache_enabled: bool
 ) -> _PreparedNativeRequest:
-    """Resolve the memo; a reused response releases the encoded request immediately."""
+    """Resolve the memo, else decode natively so the encoded payloads die when this returns."""
 
     identity: str | None = (
         native_request_identity(
@@ -302,17 +302,23 @@ def _prepare_native_request(
     return _PreparedNativeRequest(
         identity=identity,
         reused=reused,
-        encoded=encoded if reused is None else None,
+        parsed=_parse_native_request(encoded) if reused is None else None,
     )
 
 
+def _parse_native_request(encoded: _EncodedNativeRequest) -> _native.ParsedRulesRequest:
+    try:
+        return _native.parse_rules_parts(
+            encoded.request_json, encoded.model_jsons, encoded.model_digests
+        )
+    except (ValueError, TypeError) as error:
+        raise RulesError(str(error)) from error
+
+
 def _evaluate_request(*, prepared: _PreparedNativeRequest, project_dir: Path) -> str:
     if prepared.reused is not None:
         return prepared.reused
-    encoded: _EncodedNativeRequest = cast(_EncodedNativeRequest, prepared.encoded)
-    response: str = _native.evaluate_rules_parts(
-        encoded.request_json, encoded.model_jsons, encoded.model_digests
-    )
+    response: str = _native.evaluate_parsed_rules(cast(_native.ParsedRulesRequest, prepared.parsed))
     if prepared.identity is not None:
         write_native_response(
             project_dir=project_dir, identity=prepared.identity, response=response
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
index 175dc54088..1bc91d3624 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/helpers.py
@@ -61,11 +61,14 @@ def captured_native_request(
 
     captured: dict[str, Any] = {}
 
-    def evaluate_rules_parts(
+    def parse_rules_parts(
         request_json: bytes, model_jsons: list[bytes], model_digests: list[str]
-    ) -> str:
+    ) -> object:
         captured.update(cast(dict[str, Any], json.loads(request_json)))
         captured["models"] = [json.loads(model_json) for model_json in model_jsons]
+        return object()
+
+    def evaluate_parsed_rules(parsed: object) -> str:
         return json.dumps(
             {
                 "version": 1,
@@ -77,7 +80,8 @@ def evaluate_rules_parts(
             }
         )
 
-    monkeypatch.setattr(native._native, "evaluate_rules_parts", evaluate_rules_parts)
+    monkeypatch.setattr(native._native, "parse_rules_parts", parse_rules_parts)
+    monkeypatch.setattr(native._native, "evaluate_parsed_rules", evaluate_parsed_rules)
     native.evaluate_native(
         project=project,
         config=RulesConfig(cache=RulesCacheConfig(enabled=False)),
@@ -309,19 +313,17 @@ def evaluate_contract_rule(
     return evaluate(project=project, config=config, project_dir=project_dir)
 
 
-def record_native_evaluations(*, monkeypatch: pytest.MonkeyPatch) -> list[bytes]:
+def record_native_evaluations(*, monkeypatch: pytest.MonkeyPatch) -> list[object]:
     """Count every request that reaches the native built-in rules engine."""
 
-    requests: list[bytes] = []
-    evaluate_rules_parts: Callable[[bytes, list[bytes], list[str]], str] = (
-        native._native.evaluate_rules_parts
-    )
+    requests: list[object] = []
+    evaluate_parsed_rules: Callable[[Any], str] = native._native.evaluate_parsed_rules
 
-    def recording(request_json: bytes, model_jsons: list[bytes], model_digests: list[str]) -> str:
-        requests.append(request_json)
-        return evaluate_rules_parts(request_json, model_jsons, model_digests)
+    def recording(parsed: Any) -> str:
+        requests.append(parsed)
+        return evaluate_parsed_rules(parsed)
 
-    monkeypatch.setattr(native._native, "evaluate_rules_parts", recording)
+    monkeypatch.setattr(native._native, "evaluate_parsed_rules", recording)
     return requests
 
 
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
index 33222f9143..3273ff01c4 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_memo.py
@@ -75,7 +75,7 @@ def test_given_memoized_response_from_another_native_build_when_evaluating_then_
     cold: RulesResult = evaluate_contract_rule(
         config_values=config_values, project_dir=tmp_path, cache_enabled=True
     )
-    native_calls: list[bytes] = record_native_evaluations(monkeypatch=monkeypatch)
+    native_calls: list[object] = record_native_evaluations(monkeypatch=monkeypatch)
     monkeypatch.setattr(native_module, "BUILD_IDENTITY", f"{native_module.BUILD_IDENTITY}-rebuilt")
     rebuilt: RulesResult = evaluate_contract_rule(
         config_values=config_values, project_dir=tmp_path, cache_enabled=True
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
index 36423ceba1..0715875638 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_native_request.py
@@ -47,7 +47,7 @@ def record_native_call(*_: object) -> str:
 
     monkeypatch.setattr(native, "_model_payloads", failing_model_payloads)
     monkeypatch.setattr(native, "start_custom_rules", record_custom_start)
-    monkeypatch.setattr(native._native, "evaluate_rules_parts", record_native_call)
+    monkeypatch.setattr(native._native, "parse_rules_parts", record_native_call)
 
     with pytest.raises(type(test_case.expected_error)) as raised:
         native.evaluate_native(
@@ -92,7 +92,48 @@ def record_native_call(*_: object) -> str:
 
     monkeypatch.setattr(native, "_model_payloads", oversized_model_payloads)
     monkeypatch.setattr(native, "start_custom_rules", record_custom_start)
-    monkeypatch.setattr(native._native, "evaluate_rules_parts", record_native_call)
+    monkeypatch.setattr(native._native, "parse_rules_parts", record_native_call)
+
+    with pytest.raises(native.RulesError, match=test_case.expected_message):
+        native.evaluate_native(
+            project=project,
+            config=RulesConfig(cache=RulesCacheConfig(enabled=False)),
+            project_dir=tmp_path,
+            catalogue=(),
+        )
+
+    assert started == []
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        NativeRequestEncodeErrorTestCase(
+            description="model field the native engine rejects",
+            rejected_value=1,
+            expected_message="unknown field",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_payload_the_native_decoder_rejects_when_evaluating_native_then_rules_error_is_raised(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    test_case: NativeRequestEncodeErrorTestCase,
+) -> None:
+    project: CompiledProject = build_project(
+        name="orders", relative_path="models/orders.sql", sql="SELECT 1", config_values={}
+    )
+    started: list[str] = []
+
+    def unknown_field_payloads(**_: object) -> list[dict[str, object]]:
+        return [{"name": "orders", "priority": test_case.rejected_value}]
+
+    def record_custom_start(**_: object) -> None:
+        started.append("custom")
+
+    monkeypatch.setattr(native, "_model_payloads", unknown_field_payloads)
+    monkeypatch.setattr(native, "start_custom_rules", record_custom_start)
 
     with pytest.raises(native.RulesError, match=test_case.expected_message):
         native.evaluate_native(

From 9bb85e35b57987fe438467c9e2639ffa89b96b01 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Sun, 4 Oct 2026 12:09:51 +0100
Subject: [PATCH 3/3] test(rules): pin two rule cache misses for a dense
 one-model edit

The first model's built-in rule cache key used to include a project-wide
fingerprint, so any edit re-evaluated it with identical findings. Model keys
are now model-local, leaving one built-in miss and one SQL-lint miss for the
edited model.
---
 .../commands/main/compile/test_dense_compile_performance.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py
index 214f137ba9..2d801ccd1d 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py
@@ -144,7 +144,7 @@ def test_given_dense_project_when_compiling_cold_then_preserves_rules_semantics_
             3 * _GIB // 2,
             "3d26ef1a4d30ea96dd97d60b2e2f6c93473e354099138a01d74b136700c9f022",
             "f7f8a3d881496664ec7e293144d93b8b8e486a04eba830d9cc6500d1b28147f3",
-            3,
+            2,
         ),
         DenseWarmEditCompileGuardTestCase(
             "dense_models_5000_warm_and_one_edit",
@@ -155,7 +155,7 @@ def test_given_dense_project_when_compiling_cold_then_preserves_rules_semantics_
             9 * _GIB // 4,
             "bc6b60e90118d66f3217a5f04f104ae91f657a6356f4d19844614d1ba48b0012",
             "571fe58493234057c7128b4929d3518532731192f6fef85da0496788826bb172",
-            3,
+            2,
         ),
     ),
     ids=lambda case: case.description,
@@ -207,6 +207,8 @@ def test_given_dense_project_when_compiling_warm_and_after_one_edit_then_matches
     batch_hits, entry_hits, misses, bypasses = fresh_process_compile_cache_metrics(edit)
     assert 0 < misses < test_case.model_count
     assert batch_hits + entry_hits + misses == test_case.model_count
+    # The edited model misses once in built-in model rules and once in SQL lint; custom rules
+    # replay their recorded facts, and every other model's built-in findings stay cached.
     assert edit_timings["rule_cache_misses"] == test_case.expected_edit_rule_cache_misses
     oracle_timings: dict[str, int] = cast(
         dict[str, int], measurements["oracle"].payload["compile_timings"]