@@ -16,6 +16,7 @@ pub mod markers;
 pub mod mock_reads;
 pub mod planning;
 pub mod reader_references;
+pub(crate) mod real_model_limit;
 pub mod reference_calls;
 pub mod relation_markers;
 pub mod rendering;
 
@@ -31,6 +31,9 @@ use crate::compiler::_helpers::sql_tests::reader_references::{
     CompilerReads, assertion_ref_targets, calls_reference, unresolvable_call,
     unresolved_reference_error,
 };
+use crate::compiler::_helpers::sql_tests::real_model_limit::{
+    RealModelChain, real_model_limit_message,
+};
 use crate::compiler::_helpers::sql_tests::relation_markers::relation_marker_calls;
 use crate::compiler::_helpers::sql_tests::rendering::{
     RenderRequest, render_comparison_sql, render_dialect, rendered_chain_steps,
@@ -94,6 +97,7 @@ struct ProjectContext {
     patterns: SqlTestPatterns,
     render_dialect: Arc<Dialect>,
     rejects_nested_with: bool,
+    max_real_models: Option<usize>,
     /// Each model query's reserved CTE name, since every test reading the model checks it.
     reserved_ctes: Arc<Mutex<HashMap<String, Option<String>>>>,
 }
@@ -469,7 +473,7 @@ pub(crate) fn resolve_chains(request: SqlTestChainBatch) -> Result<Vec<Vec<Strin
                 let fixtures = classify_fixtures(authored_ctes, expected_ctes, assertion_ctes);
                 let expected_names =
                     chain_root_names(expected_model_names, &reads, &fixtures, &patterns);
-                Ok(topo_sort_model_chain(TopoSortRequest {
+                Ok(topo_sort_model_chain(&TopoSortRequest {
                     expected_names: &expected_names,
                     models: &models,
                     overrides: &model_query_overrides,
@@ -537,6 +541,7 @@ pub(crate) fn plan_batch(request: SqlTestPlanBatch) -> Result<SqlTestPlanBatchOu
         patterns: SqlTestPatterns::new(request.lexical_syntax)?,
         render_dialect,
         rejects_nested_with,
+        max_real_models: request.max_unit_test_real_models,
         reserved_ctes: Arc::new(Mutex::new(HashMap::new())),
     };
     let workers = request.workers.clamp(1, MAX_WORKERS);
@@ -683,6 +688,8 @@ fn plan_direct_test(
         set_difference_operator: context.set_difference_operator.clone(),
         sql_analysis_dialect: Some(context.dialect.clone()),
         probe_step_index: None,
+        actual_relations: HashMap::new(),
+        output_step_index: None,
     };
     Ok(PlannedResponse {
         request,
@@ -755,13 +762,50 @@ fn plan_model_test(
         &fixtures,
         &context.patterns,
     );
-    let ordered_names = topo_sort_model_chain(TopoSortRequest {
+    let topo_request = TopoSortRequest {
         expected_names: &expected_names,
         models: &context.models,
         overrides: &plan.model_query_overrides,
         mock_refs: &fixtures.mock_refs,
         patterns: &context.patterns,
-    });
+    };
+    let ordered_names = topo_sort_model_chain(&topo_request);
+    if let Some(limit) = context.max_real_models {
+        let real_models: Vec<String> = ordered_names
+            .iter()
+            .filter(|name| context.models.contains_key(*name))
+            .cloned()
+            .collect();
+        let chain = RealModelChain {
+            test_name: &plan.test_name,
+            file_label: &plan.file_label,
+            roots: &expected_names,
+            real_models: &real_models,
+            limit,
+        };
+        if let Some(message) =
+            real_model_limit_message(&chain, |name| real_model_inputs(&topo_request, name))
+        {
+            return Ok(PlannedResponse {
+                request: RenderRequest {
+                    chain: Vec::new(),
+                    assertions: Vec::new(),
+                    sql_analysis_enabled: context.sql_analysis_enabled,
+                    set_difference_operator: context.set_difference_operator.clone(),
+                    sql_analysis_dialect: Some(context.dialect.clone()),
+                    probe_step_index: None,
+                    actual_relations: HashMap::new(),
+                    output_step_index: None,
+                },
+                model_names: real_models,
+                warnings: vec![SqlTestPlanWarning {
+                    model_name: None,
+                    severity: "error",
+                    message,
+                }],
+            });
+        }
+    }
     let mut warnings: Vec<SqlTestPlanWarning> = Vec::new();
     let mut reported_missing_mocks: HashSet<(&'static str, String)> = HashSet::new();
     let mut reachable_mocks: HashSet<String> = HashSet::new();
@@ -928,6 +972,8 @@ fn plan_model_test(
         set_difference_operator: context.set_difference_operator.clone(),
         sql_analysis_dialect: Some(context.dialect.clone()),
         probe_step_index: None,
+        actual_relations: HashMap::new(),
+        output_step_index: None,
     };
     Ok(PlannedResponse {
         request,
@@ -1302,28 +1348,31 @@ fn classify_fixtures(
     fixtures
 }
 
-fn topo_sort_model_chain(request: TopoSortRequest<'_>) -> Vec<String> {
+fn topo_sort_model_chain(request: &TopoSortRequest<'_>) -> Vec<String> {
     let mut roots: Vec<String> = request.expected_names.to_vec();
     roots.sort();
-    depth_first_postorder(roots, |name| {
-        let Some(model) = request.models.get(name) else {
-            return Vec::new();
-        };
-        let mut dependencies: Vec<String> = match request.overrides.get(name) {
-            Some(override_sql) => marker_names(
-                &request.patterns.reference,
-                &request.patterns.lexical,
-                override_sql,
-            ),
-            None => model.model_dependencies.clone(),
-        };
-        dependencies.retain(|dependency| {
-            !request.mock_refs.contains_key(dependency) && request.models.contains_key(dependency)
-        });
-        dependencies.sort();
-        dependencies.dedup();
-        dependencies
-    })
+    depth_first_postorder(roots, |name| real_model_inputs(request, name))
+}
+
+/// The unmocked project models one chain model reads, sorted and deduplicated.
+fn real_model_inputs(request: &TopoSortRequest<'_>, name: &str) -> Vec<String> {
+    let Some(model) = request.models.get(name) else {
+        return Vec::new();
+    };
+    let mut dependencies: Vec<String> = match request.overrides.get(name) {
+        Some(override_sql) => marker_names(
+            &request.patterns.reference,
+            &request.patterns.lexical,
+            override_sql,
+        ),
+        None => model.model_dependencies.clone(),
+    };
+    dependencies.retain(|dependency| {
+        !request.mock_refs.contains_key(dependency) && request.models.contains_key(dependency)
+    });
+    dependencies.sort();
+    dependencies.dedup();
+    dependencies
 }
 
 fn analyze_and_resolve_sql(
 
@@ -0,0 +1,136 @@
+//! The compile error for a unit test that executes more real models than the project allows.
+
+use std::collections::{BTreeSet, HashMap, VecDeque};
+
+/// The project setting that bounds how many real models one unit test may execute.
+pub(crate) const REAL_MODEL_LIMIT_KEY: &str = "unit_test_max_real_models";
+
+/// One unit test's chain, measured against the configured real-model limit.
+pub(crate) struct RealModelChain<'a> {
+    pub(crate) test_name: &'a str,
+    pub(crate) file_label: &'a str,
+    pub(crate) roots: &'a [String],
+    pub(crate) real_models: &'a [String],
+    pub(crate) limit: usize,
+}
+
+/// The explanation of an oversized chain, or `None` when the chain is within the limit.
+pub(crate) fn real_model_limit_message<F>(chain: &RealModelChain<'_>, inputs: F) -> Option<String>
+where
+    F: Fn(&str) -> Vec<String>,
+{
+    let count = chain.real_models.len();
+    if count <= chain.limit {
+        return None;
+    }
+    let roots: Vec<&String> = chain
+        .roots
+        .iter()
+        .filter(|root| chain.real_models.contains(root))
+        .collect();
+    let paths = dependency_paths(&roots, &inputs);
+    let width = chain.real_models.iter().map(String::len).max().unwrap_or(0);
+    let mut lines: Vec<String> = vec![
+        format!(
+            "SQL unit test '{}' executes {count} real models (limit {}).",
+            chain.test_name, chain.limit
+        ),
+        format!("  Test file:        {}", chain.file_label),
+        format!(
+            "  Model under test: {}",
+            roots
+                .iter()
+                .map(|root| root.as_str())
+                .collect::<Vec<_>>()
+                .join(", ")
+        ),
+        String::new(),
+        "Real models executed, and the dependency path that pulls each one in:".to_owned(),
+    ];
+    for model in chain.real_models.iter().rev() {
+        let reason = match paths.get(model.as_str()) {
+            Some(path) if path.len() == 1 => "model under test".to_owned(),
+            Some(path) => path.join(" -> "),
+            None => "read by the test".to_owned(),
+        };
+        lines.push(format!("  {model:<width$}  {reason}"));
+    }
+    lines.push(String::new());
+    lines.push("Options:".to_owned());
+    lines.extend(mock_advice(&roots, chain.real_models, &inputs));
+    lines.extend(scenario_advice());
+    lines.extend([
+        "  3. Raise the limit in sqlbuild_project.toml (or sqlbuild_local.toml):".to_owned(),
+        "       [settings]".to_owned(),
+        format!("       {REAL_MODEL_LIMIT_KEY} = {count}"),
+    ]);
+    Some(lines.join("\n"))
+}
+
+/// Why a scenario may suit a multi-model check, as the second option.
+fn scenario_advice() -> [String; 4] {
+    [
+        "  2. If this test checks behaviour across several models, consider a scenario".to_owned(),
+        "     (tests/scenarios/). A unit test inlines every real model into one statement,"
+            .to_owned(),
+        "     so the warehouse compiles and optimises the whole chain at once; a scenario"
+            .to_owned(),
+        "     builds each model step by step, compiling each statement separately.".to_owned(),
+    ]
+}
+
+/// The direct real inputs of each model under test, which a test mocks to cut the chain.
+fn mock_advice<F>(roots: &[&String], real_models: &[String], inputs: &F) -> Vec<String>
+where
+    F: Fn(&str) -> Vec<String>,
+{
+    let mut seen: BTreeSet<String> = BTreeSet::new();
+    let mut mocks: Vec<String> = Vec::new();
+    for root in roots {
+        for input in inputs(root) {
+            let is_root = roots.iter().any(|candidate| **candidate == input);
+            if !is_root && real_models.contains(&input) && seen.insert(input.clone()) {
+                mocks.push(input);
+            }
+        }
+    }
+    if mocks.is_empty() {
+        return vec![
+            "  1. Mock models further up the paths above with __ref__<model> CTEs.".to_owned(),
+        ];
+    }
+    let mut lines: Vec<String> =
+        vec!["  1. Mock the direct inputs of the model under test to cut the chain:".to_owned()];
+    for mock in mocks {
+        lines.push(format!("       __ref__{mock} AS (SELECT ...)"));
+    }
+    lines
+}
+
+/// The shortest read path from a model under test to each real model, breadth first.
+fn dependency_paths<F>(roots: &[&String], inputs: &F) -> HashMap<String, Vec<String>>
+where
+    F: Fn(&str) -> Vec<String>,
+{
+    let mut paths: HashMap<String, Vec<String>> = HashMap::new();
+    let mut queue: VecDeque<String> = VecDeque::new();
+    for root in roots {
+        if !paths.contains_key(root.as_str()) {
+            paths.insert((*root).clone(), vec![(*root).clone()]);
+            queue.push_back((*root).clone());
+        }
+    }
+    while let Some(model) = queue.pop_front() {
+        let path = paths.get(&model).cloned().unwrap_or_default();
+        for input in inputs(&model) {
+            if paths.contains_key(&input) {
+                continue;
+            }
+            let mut extended = path.clone();
+            extended.push(input.clone());
+            paths.insert(input.clone(), extended);
+            queue.push_back(input);
+        }
+    }
+    paths
+}