From 4e29ab43d3edf74c499ffb59a1d409af897aceff Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 8 Oct 2026 16:27:19 +0100
Subject: [PATCH 1/3] feat(native): report reference extraction errors natively

---
 .../_helpers/sqltext/sql_references.rs        |  36 ++-
 .../src/sql_references/_helpers/scan.rs       | 258 ++++++++++++++----
 .../src/sql_references/constants.rs           |  15 +-
 .../main/extract_sql_references.rs            |   4 +-
 .../src/sql_references/models.rs              |  25 +-
 .../src/sql_references/tests/helpers.rs       |  34 ++-
 .../tests/test_extract_sql_references.rs      | 223 ++++++++++++---
 .../corpus/reference_failure_cases.py         |  44 +++
 src/sqlbuild/_native.pyi                      |  12 +-
 .../compiler/compile/_helpers/refs/native.py  |  60 ++--
 .../compile/_helpers/refs/references.py       |  48 +---
 src/sqlbuild/compiler/compile/exceptions.py   |   4 +
 .../compile/_helpers/refs/_test_types.py      |  21 +-
 .../compiler/compile/_helpers/refs/helpers.py | 123 +++++++--
 .../_helpers/refs/test_native_references.py   | 155 +++++++----
 15 files changed, 821 insertions(+), 241 deletions(-)

diff --git a/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs b/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
index f36013ea39..64c754ecb8 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
@@ -3,7 +3,9 @@
 use pyo3::prelude::{Bound, PyModule, PyModuleMethods, PyResult};
 use pyo3::{pyclass, pymethods};
 use sqlbuild_sqltext::sql_references::main::extract_sql_references::extract_sql_references;
-use sqlbuild_sqltext::sql_references::models::{ReferenceExtraction, SqlReference};
+use sqlbuild_sqltext::sql_references::models::{
+    InvalidReferenceCall, ReferenceExtraction, SqlReference,
+};
 use sqlbuild_sqltext::sql_scan::models::LexicalSyntax;
 
 use crate::bindings::_helpers::boundary::panics::compiler_guard;
@@ -11,8 +13,13 @@ use crate::bindings::_helpers::sqltext::lexical_syntax::LexicalSyntaxInput;
 
 /// `(kind, name, package, call_argument_count)` for one reference.
 type ReferenceRow = (&'static str, String, Option<String>, Option<usize>);
-/// The references, or Python's error message; `None` defers the text to Python.
-type ExtractionRow = Option<(Option<Vec<ReferenceRow>>, Option<String>)>;
+/// `(kind, call, start, message, help, corrected_call)` for one rejected call.
+type InvalidCallRow = (&'static str, String, usize, String, String, String);
+/// The references and rejected calls, or Python's error message; `None` defers to Python.
+type ExtractionRow = Option<(
+    Option<(Vec<ReferenceRow>, Vec<InvalidCallRow>)>,
+    Option<String>,
+)>;
 
 /// Extracts references under one adapter's lexical rules, read once per syntax.
 #[pyclass(module = "sqlbuild._native", frozen)]
@@ -29,12 +36,18 @@ impl SqlReferenceScanner {
         }
     }
 
-    /// Return `(references, None)`, `(None, message)` when Python raises, or `None` to defer.
+    /// Return `((references, rejected calls), None)`, `(None, message)`, or `None` to defer.
     fn extract(&self, sql: &str) -> PyResult<ExtractionRow> {
         compiler_guard(|| {
             Ok(match extract_sql_references(sql, &self.syntax) {
-                ReferenceExtraction::Extracted(references) => Some((
-                    Some(references.into_iter().map(reference_row).collect()),
+                ReferenceExtraction::Extracted(scan) => Some((
+                    Some((
+                        scan.references.into_iter().map(reference_row).collect(),
+                        scan.invalid_calls
+                            .into_iter()
+                            .map(invalid_call_row)
+                            .collect(),
+                    )),
                     None,
                 )),
                 ReferenceExtraction::Failed(message) => Some((None, Some(message))),
@@ -53,6 +66,17 @@ fn reference_row(reference: SqlReference) -> ReferenceRow {
     )
 }
 
+fn invalid_call_row(call: InvalidReferenceCall) -> InvalidCallRow {
+    (
+        call.kind,
+        call.call,
+        call.start,
+        call.message,
+        call.help,
+        call.corrected_call,
+    )
+}
+
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_class::<SqlReferenceScanner>()?;
     Ok(())
diff --git a/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs b/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
index 9eac895239..2d6bb53df1 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
@@ -1,11 +1,13 @@
 //! The general reference scan of Python `extract_sql_references`, reproduced exactly.
 
+use sqlbuild_core::text::main::is_python_space::is_python_space;
+
 use crate::sql_references::constants::{
-    DBT_REFERENCE_KIND, NON_CODE_START_BYTES, PYTHON_ASCII_WHITESPACE, QUOTE_BYTES,
-    REFERENCE_CONTEXT, REFERENCE_PREFIXES, TABLE_FUNCTION_CALL_CONTEXT,
+    DBT_REFERENCE_KIND, NAME_QUOTE_BYTES, NON_CODE_START_BYTES, PLACEHOLDER_NAMES, QUOTE_BYTES,
+    QUOTED_NAME_MINIMUM_BYTES, REFERENCE_CONTEXT, REFERENCE_PREFIXES, TABLE_FUNCTION_CALL_CONTEXT,
     TABLE_FUNCTION_REFERENCE_KIND,
 };
-use crate::sql_references::models::SqlReference;
+use crate::sql_references::models::{InvalidReferenceCall, ReferenceScan, SqlReference};
 use crate::sql_references::types::ReferencePrefix;
 use crate::sql_scan::main::dialect_non_code_end::dialect_non_code_end;
 use crate::sql_scan::models::{LexicalSyntax, Unclosed};
@@ -15,7 +17,7 @@ use crate::sql_scan::models::{LexicalSyntax, Unclosed};
 pub(crate) enum Stop {
     /// Python raises `CompileInputError` with this message.
     Failed(String),
-    /// A non-ASCII character decides the outcome, which needs Python's `str` classification.
+    /// A character or construct whose classification only Python reproduces.
     Deferred,
 }
 
@@ -36,9 +38,16 @@ const fn byte_table(bytes: &[u8], underscore: bool) -> [bool; 256] {
     table
 }
 
-/// Return every reference in `sql` in authored order.
-pub(crate) fn scan_references(sql: &[u8], syntax: &LexicalSyntax) -> Scan<Vec<SqlReference>> {
-    let mut references: Vec<SqlReference> = Vec::new();
+/// One parsed call: a reference, or a rejected call without its start.
+enum Parsed {
+    Reference(SqlReference),
+    Invalid(InvalidReferenceCall),
+}
+
+/// Return every reference and rejected call in `sql` in authored order.
+pub(crate) fn scan_references(sql: &[u8], syntax: &LexicalSyntax) -> Scan<ReferenceScan> {
+    let mut scan: ReferenceScan = ReferenceScan::default();
+    let mut offsets: CharOffsets = CharOffsets::default();
     let mut index = 0;
     while index < sql.len() {
         if let Some(end) = non_code_end(sql, index, syntax, REFERENCE_CONTEXT)? {
@@ -59,76 +68,209 @@ pub(crate) fn scan_references(sql: &[u8], syntax: &LexicalSyntax) -> Scan<Vec<Sq
             index += 1;
             continue;
         };
-        let (reference, next) = parse_reference(sql, index, call, syntax)?;
-        references.push(reference);
+        let (parsed, next) = parse_reference(sql, index, call, syntax)?;
+        match parsed {
+            Parsed::Reference(reference) => scan.references.push(reference),
+            Parsed::Invalid(invalid_call) => scan.invalid_calls.push(InvalidReferenceCall {
+                start: offsets.advance(sql, index)?,
+                ..invalid_call
+            }),
+        }
         index = next;
     }
-    Ok(references)
+    Ok(scan)
 }
 
-/// Parse a bare double-quoted reference call at `start`; other forms defer so Python reports P012.
+/// Parse the reference call at `start` as Python `_parse_reference_at` does.
 fn parse_reference(
     sql: &[u8],
     start: usize,
     call: &ReferencePrefix,
     syntax: &LexicalSyntax,
-) -> Scan<(SqlReference, usize)> {
+) -> Scan<(Parsed, usize)> {
     let (prefix, kind) = *call;
     let open = start + prefix.len() - 1;
     let close = matching_paren(sql, open, syntax, REFERENCE_CONTEXT)?;
     let arguments = &sql[open + 1..close];
-    let (name, package) = if kind == DBT_REFERENCE_KIND {
+    let call_text = &sql[start..=close];
+    let names = if kind == DBT_REFERENCE_KIND {
         dbt_reference_names(arguments)?
     } else {
-        (bare_name(arguments)?, None)
+        bare_name(arguments)?.map(|name| (name, None))
+    };
+    let Some((name, package)) = names else {
+        let invalid = invalid_arguments(kind, sql, (start, open, close), syntax)?;
+        return Ok((Parsed::Invalid(invalid), close + 1));
     };
     let mut call_argument_count = None;
     if kind == TABLE_FUNCTION_REFERENCE_KIND {
         let suffix = python_whitespace_end(sql, close + 1)?;
         if sql.get(suffix) != Some(&b'(') {
-            return Err(Stop::Deferred);
+            let corrected_call = format!("{}()", example_call(kind, &[name.as_str()]));
+            let invalid = InvalidReferenceCall {
+                kind,
+                call: utf8(call_text)?,
+                start: 0,
+                message: format!(
+                    "{} must be followed by an argument list",
+                    function_name(kind)
+                ),
+                help: format!(
+                    "pass the function arguments in a second set of parentheses, using () for \
+                     no arguments: {corrected_call}"
+                ),
+                corrected_call,
+            };
+            return Ok((Parsed::Invalid(invalid), close + 1));
         }
         let suffix_close = matching_paren(sql, suffix, syntax, TABLE_FUNCTION_CALL_CONTEXT)?;
         call_argument_count = Some(split_arguments(sql, suffix + 1, suffix_close, syntax)?.len());
     }
     Ok((
-        SqlReference {
+        Parsed::Reference(SqlReference {
             kind,
             name,
             package,
             call_argument_count,
-        },
+        }),
         close + 1,
     ))
 }
 
-/// Python `"([^"]+)"` matched against the whole argument text.
-fn bare_name(arguments: &[u8]) -> Scan<String> {
+/// Python `_invalid_reference_arguments`: the P012 message, help and corrected call.
+fn invalid_arguments(
+    kind: &'static str,
+    sql: &[u8],
+    (start, open, close): (usize, usize, usize),
+    syntax: &LexicalSyntax,
+) -> Scan<InvalidReferenceCall> {
+    let allowed: &[usize] = if kind == DBT_REFERENCE_KIND {
+        &[1, 2]
+    } else {
+        &[1]
+    };
+    let authored: Option<Vec<String>> = authored_names(sql, open + 1, close, syntax)?
+        .filter(|names| allowed.contains(&names.len()));
+    let names: Vec<String> = authored.unwrap_or_else(|| {
+        PLACEHOLDER_NAMES
+            .iter()
+            .find(|(candidate, _)| *candidate == kind)
+            .map_or_else(Vec::new, |(_, names)| {
+                names.iter().map(|name| (*name).to_owned()).collect()
+            })
+    });
+    let name_refs: Vec<&str> = names.iter().map(String::as_str).collect();
+    let mut corrected_call = example_call(kind, &name_refs);
+    if kind == TABLE_FUNCTION_REFERENCE_KIND {
+        corrected_call.push_str("(...)");
+    }
+    let accepted = if kind == DBT_REFERENCE_KIND {
+        "one double-quoted model name, or a double-quoted package name and model name separated \
+         by a comma"
+    } else {
+        "exactly one double-quoted name"
+    };
+    let call = utf8(&sql[start..=close])?;
+    let words: Vec<&str> = call
+        .split(is_python_space)
+        .filter(|word| !word.is_empty())
+        .collect();
+    let prefix = function_name(kind);
+    Ok(InvalidReferenceCall {
+        kind,
+        message: format!("{} is not a valid {prefix}() call", words.join(" ")),
+        help: format!(
+            "{prefix}() takes {accepted}, with no comments or extra spaces inside the \
+             parentheses: {corrected_call}"
+        ),
+        corrected_call,
+        call,
+        start: 0,
+    })
+}
+
+/// Python `_authored_reference_names`: each argument as a plain or quoted name, else `None`.
+fn authored_names(
+    sql: &[u8],
+    start: usize,
+    end: usize,
+    syntax: &LexicalSyntax,
+) -> Scan<Option<Vec<String>>> {
+    let split = match split_arguments(sql, start, end, syntax) {
+        Ok(split) => split,
+        Err(Stop::Failed(_)) => return Ok(None),
+        Err(Stop::Deferred) => return Err(Stop::Deferred),
+    };
+    let mut names: Vec<String> = Vec::with_capacity(split.len());
+    for argument in split {
+        let name: &[u8] = match (argument.first(), argument.last()) {
+            (Some(first), Some(last))
+                if argument.len() >= QUOTED_NAME_MINIMUM_BYTES
+                    && first == last
+                    && NAME_QUOTE_BYTES.contains(first) =>
+            {
+                &argument[1..argument.len() - 1]
+            }
+            _ => &argument,
+        };
+        if !is_authored_name(name) {
+            return Ok(None);
+        }
+        names.push(utf8(name)?);
+    }
+    Ok(Some(names))
+}
+
+/// Python `[A-Za-z_][A-Za-z0-9_.]*` matched against the whole name.
+fn is_authored_name(name: &[u8]) -> bool {
+    name.first()
+        .is_some_and(|first| first.is_ascii_alphabetic() || *first == b'_')
+        && name[1..]
+            .iter()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.'))
+}
+
+/// Python `SqlReferenceKind.function_name`.
+fn function_name(kind: &str) -> String {
+    format!("__{kind}")
+}
+
+/// Python `SqlReferenceKind.example_call(*names, quote='"')`.
+fn example_call(kind: &str, names: &[&str]) -> String {
+    let quoted: Vec<String> = names
+        .iter()
+        .map(|name| format!("\"{}\"", name.replace('"', "\"\"")))
+        .collect();
+    format!("{}({})", function_name(kind), quoted.join(", "))
+}
+
+/// Python `"([^"]+)"` matched against the whole argument text; `None` when it does not match.
+fn bare_name(arguments: &[u8]) -> Scan<Option<String>> {
     match quoted_name_end(arguments, 0) {
-        Some(end) if end == arguments.len() => utf8(&arguments[1..end - 1]),
-        _ => Err(Stop::Deferred),
+        Some(end) if end == arguments.len() => utf8(&arguments[1..end - 1]).map(Some),
+        _ => Ok(None),
     }
 }
 
-/// Python `"([^"]+)"(?:\s*,\s*"([^"]+)")?` matched against the whole argument text.
-fn dbt_reference_names(arguments: &[u8]) -> Scan<(String, Option<String>)> {
+/// Python `"([^"]+)"(?:\s*,\s*"([^"]+)")?` over the whole text: name and package, or `None`.
+fn dbt_reference_names(arguments: &[u8]) -> Scan<Option<(String, Option<String>)>> {
     let Some(first_end) = quoted_name_end(arguments, 0) else {
-        return Err(Stop::Deferred);
+        return Ok(None);
     };
     if first_end == arguments.len() {
-        return Ok((utf8(&arguments[1..first_end - 1])?, None));
+        return Ok(Some((utf8(&arguments[1..first_end - 1])?, None)));
     }
     let separator = python_whitespace_end(arguments, first_end)?;
     if arguments.get(separator) != Some(&b',') {
-        return Err(Stop::Deferred);
+        return Ok(None);
     }
     let second_start = python_whitespace_end(arguments, separator + 1)?;
     match quoted_name_end(arguments, second_start) {
-        Some(end) if end == arguments.len() => Ok((
+        Some(end) if end == arguments.len() => Ok(Some((
             utf8(&arguments[second_start + 1..end - 1])?,
             Some(utf8(&arguments[1..first_end - 1])?),
-        )),
-        _ => Err(Stop::Deferred),
+        ))),
+        _ => Ok(None),
     }
 }
 
@@ -202,40 +344,36 @@ fn split_arguments(
     Ok(arguments)
 }
 
-/// Python `str.strip()`; a non-ASCII character at either edge may be Unicode whitespace.
+/// Python `str.strip()`, which strips `str.isspace()` characters.
 fn python_strip(value: &[u8]) -> Scan<&[u8]> {
-    let start = value
-        .iter()
-        .position(|byte| !PYTHON_ASCII_WHITESPACE.contains(byte))
-        .unwrap_or(value.len());
-    let end = value
-        .iter()
-        .rposition(|byte| !PYTHON_ASCII_WHITESPACE.contains(byte))
-        .map_or(start, |last| last + 1);
-    let stripped = &value[start..end];
-    if stripped.first().is_some_and(|byte| !byte.is_ascii())
-        || stripped.last().is_some_and(|byte| !byte.is_ascii())
-    {
-        return Err(Stop::Deferred);
-    }
-    Ok(stripped)
+    let text: &str = std::str::from_utf8(value).map_err(|_| Stop::Deferred)?;
+    Ok(text.trim_matches(is_python_space).as_bytes())
 }
 
 /// Python `_skip_whitespace`, which skips `str.isspace()` characters.
 fn python_whitespace_end(sql: &[u8], start: usize) -> Scan<usize> {
     let mut index = start;
-    while let Some(byte) = sql.get(index) {
-        if !byte.is_ascii() {
-            return Err(Stop::Deferred);
-        }
-        if !PYTHON_ASCII_WHITESPACE.contains(byte) {
+    while index < sql.len() {
+        let character: char = char_at(sql, index)?;
+        if !is_python_space(character) {
             break;
         }
-        index += 1;
+        index += character.len_utf8();
     }
     Ok(index)
 }
 
+/// The character starting at byte `index` of UTF-8 text.
+fn char_at(text: &[u8], index: usize) -> Scan<char> {
+    let width: usize = (text[index].leading_ones() as usize).max(1);
+    let encoded: &[u8] = text.get(index..index + width).ok_or(Stop::Deferred)?;
+    std::str::from_utf8(encoded)
+        .map_err(|_| Stop::Deferred)?
+        .chars()
+        .next()
+        .ok_or(Stop::Deferred)
+}
+
 fn non_code_end(
     sql: &[u8],
     index: usize,
@@ -289,3 +427,21 @@ fn utf8(value: &[u8]) -> Scan<String> {
 fn failed(message: String) -> Stop {
     Stop::Failed(message)
 }
+
+/// Converts increasing byte offsets of UTF-8 text into Python code-point offsets.
+#[derive(Default)]
+struct CharOffsets {
+    byte: usize,
+    char: usize,
+}
+
+impl CharOffsets {
+    fn advance(&mut self, text: &[u8], byte: usize) -> Scan<usize> {
+        self.char += std::str::from_utf8(&text[self.byte..byte])
+            .map_err(|_| Stop::Deferred)?
+            .chars()
+            .count();
+        self.byte = byte;
+        Ok(self.char)
+    }
+}
diff --git a/crates/sqlbuild-sqltext/src/sql_references/constants.rs b/crates/sqlbuild-sqltext/src/sql_references/constants.rs
index 086b820dc9..5bd768c96e 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/constants.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/constants.rs
@@ -21,5 +21,16 @@ pub(crate) const SUPPORTED_LINE_COMMENT_PREFIXES: [&str; 3] = ["--", "//", "#"];
 pub(crate) const NON_CODE_START_BYTES: &[u8] = b"-/#'\"`$";
 /// Non-code openers whose text an argument keeps; comments become one space.
 pub(crate) const QUOTE_BYTES: &[u8] = b"'\"`$";
-/// Python `str.isspace()` for ASCII characters.
-pub(crate) const PYTHON_ASCII_WHITESPACE: &[u8] = b" \t\n\x0b\x0c\r\x1c\x1d\x1e\x1f";
+/// The shortest argument whose quotes Python strips: the two quotes.
+pub(crate) const QUOTED_NAME_MINIMUM_BYTES: usize = 2;
+/// Quotes Python strips from an authored reference name.
+pub(crate) const NAME_QUOTE_BYTES: &[u8] = b"'\"";
+/// Python's placeholder names for a rejected call's corrected form, by kind.
+pub(crate) const PLACEHOLDER_NAMES: [(&str, &[&str]); 6] = [
+    ("ref", &["model_name"]),
+    ("source", &["source_name"]),
+    ("seed", &["seed_name"]),
+    ("udf", &["function_name"]),
+    ("table_fn", &["function_name"]),
+    ("dbt_ref", &["package_name", "model_name"]),
+];
diff --git a/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs b/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
index 65c55d7084..36067237ea 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
@@ -5,7 +5,7 @@ use crate::sql_references::constants::SUPPORTED_LINE_COMMENT_PREFIXES;
 use crate::sql_references::models::ReferenceExtraction;
 use crate::sql_scan::models::LexicalSyntax;
 
-/// Return the references Python `extract_sql_references` finds in `sql`, or its error message.
+/// Return the references and rejected calls Python finds in `sql`, or its error message.
 pub fn extract_sql_references(sql: &str, syntax: &LexicalSyntax) -> ReferenceExtraction {
     if !syntax
         .line_comment_prefixes
@@ -15,7 +15,7 @@ pub fn extract_sql_references(sql: &str, syntax: &LexicalSyntax) -> ReferenceExt
         return ReferenceExtraction::Deferred;
     }
     match scan_references(sql.as_bytes(), syntax) {
-        Ok(references) => ReferenceExtraction::Extracted(references),
+        Ok(scan) => ReferenceExtraction::Extracted(scan),
         Err(Stop::Failed(message)) => ReferenceExtraction::Failed(message),
         Err(Stop::Deferred) => ReferenceExtraction::Deferred,
     }
diff --git a/crates/sqlbuild-sqltext/src/sql_references/models.rs b/crates/sqlbuild-sqltext/src/sql_references/models.rs
index 0cfa935aa3..4a2b435881 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/models.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/models.rs
@@ -11,11 +11,32 @@ pub struct SqlReference {
     pub call_argument_count: Option<usize>,
 }
 
+/// One reference call compile cannot replace, with Python's P012 message and help.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct InvalidReferenceCall {
+    /// Python `SqlReferenceKind` value of the call.
+    pub kind: &'static str,
+    /// The call text from its prefix through its closing parenthesis.
+    pub call: String,
+    /// The code-point offset of the call in the scanned text.
+    pub start: usize,
+    pub message: String,
+    pub help: String,
+    pub corrected_call: String,
+}
+
+/// The valid references and rejected calls of one SQL text, each in authored order.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ReferenceScan {
+    pub references: Vec<SqlReference>,
+    pub invalid_calls: Vec<InvalidReferenceCall>,
+}
+
 /// The references of one SQL text, Python's error for it, or a deferral to Python.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum ReferenceExtraction {
-    /// Every reference in authored order, exactly as Python extracts them.
-    Extracted(Vec<SqlReference>),
+    /// Every reference and rejected call in authored order, exactly as Python scans them.
+    Extracted(ReferenceScan),
     /// The message of the `CompileInputError` Python raises for this text.
     Failed(String),
     /// A character or syntax rule the scan cannot classify exactly as Python does.
diff --git a/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs b/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
index 2666a4b9a5..c0b7ce9482 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
@@ -1,6 +1,8 @@
 //! Lexical syntaxes and expected references for reference extraction tests.
 
-use crate::sql_references::models::{ReferenceExtraction, SqlReference};
+use crate::sql_references::models::{
+    InvalidReferenceCall, ReferenceExtraction, ReferenceScan, SqlReference,
+};
 use crate::sql_scan::models::LexicalSyntax;
 
 pub(crate) fn generic() -> LexicalSyntax {
@@ -50,7 +52,35 @@ pub(crate) fn dbt_reference(package: &str, name: &str) -> SqlReference {
 }
 
 pub(crate) fn extracted(references: Vec<SqlReference>) -> ReferenceExtraction {
-    ReferenceExtraction::Extracted(references)
+    rejected(references, Vec::new())
+}
+
+pub(crate) fn rejected(
+    references: Vec<SqlReference>,
+    invalid_calls: Vec<InvalidReferenceCall>,
+) -> ReferenceExtraction {
+    ReferenceExtraction::Extracted(ReferenceScan {
+        references,
+        invalid_calls,
+    })
+}
+
+pub(crate) fn invalid_call(
+    kind: &'static str,
+    call: &str,
+    start: usize,
+    message: &str,
+    help: &str,
+    corrected_call: &str,
+) -> InvalidReferenceCall {
+    InvalidReferenceCall {
+        kind,
+        call: call.to_string(),
+        start,
+        message: message.to_string(),
+        help: help.to_string(),
+        corrected_call: corrected_call.to_string(),
+    }
 }
 
 pub(crate) fn failed(message: &str) -> ReferenceExtraction {
diff --git a/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs b/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
index a1c74c301f..90284aebf4 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
@@ -1,8 +1,8 @@
 use crate::sql_references::main::extract_sql_references::extract_sql_references;
 use crate::sql_references::models::ReferenceExtraction;
 use crate::sql_references::tests::helpers::{
-    backslash_hash_comments, dbt_reference, extracted, failed, generic, reference, table_function,
-    unsupported_comments,
+    backslash_hash_comments, dbt_reference, extracted, failed, generic, invalid_call, reference,
+    rejected, table_function, unsupported_comments,
 };
 use crate::sql_references::tests::test_types::ExtractSqlReferencesTestCase;
 
@@ -46,10 +46,21 @@ fn given_sql_when_extracting_references_then_matches_python_scanner() {
             ]),
         },
         ExtractSqlReferencesTestCase {
-            description: "comment inside a reference call defers to python",
+            description: "comment inside a reference call is rejected",
             sql: "SELECT * FROM __ref(\"a\" /* note */ \"b\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(\"a\" /* note */ \"b\")",
+                    14,
+                    "__ref(\"a\" /* note */ \"b\") is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces \
+                     inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
             description: "comments, strings and dollar quotes hide references",
@@ -100,94 +111,234 @@ fn given_sql_when_extracting_references_then_matches_python_scanner() {
             expected_extraction: failed("SQL reference contains an unclosed quoted string"),
         },
         ExtractSqlReferencesTestCase {
-            description: "unquoted name defers to python",
+            description: "an unquoted name is rejected",
             sql: "SELECT * FROM __ref(orders)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(orders)",
+                    14,
+                    "__ref(orders) is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"orders\")",
+                    "__ref(\"orders\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "single quoted name defers to python",
+            description: "a single quoted name is rejected",
             sql: "SELECT * FROM __source('raw_orders')",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "source",
+                    "__source('raw_orders')",
+                    14,
+                    "__source('raw_orders') is not a valid __source() call",
+                    "__source() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __source(\"raw_orders\")",
+                    "__source(\"raw_orders\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "spaces inside the call defer to python",
+            description: "spaces inside the call are rejected",
             sql: "SELECT * FROM __seed( \"regions\" )",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "seed",
+                    "__seed( \"regions\" )",
+                    14,
+                    "__seed( \"regions\" ) is not a valid __seed() call",
+                    "__seed() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __seed(\"regions\")",
+                    "__seed(\"regions\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "empty quoted name defers to python",
+            description: "an empty quoted name is rejected",
             sql: "SELECT * FROM __ref(\"\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(\"\")",
+                    14,
+                    "__ref(\"\") is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "doubled quote inside a name defers to python",
+            description: "a doubled quote inside a name is rejected",
             sql: "SELECT * FROM __ref(\"ord\"\"ers\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(\"ord\"\"ers\")",
+                    14,
+                    "__ref(\"ord\"\"ers\") is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "second name argument defers to python",
+            description: "a second name argument is rejected",
             sql: "SELECT * FROM __ref(\"a\", \"b\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(\"a\", \"b\")",
+                    14,
+                    "__ref(\"a\", \"b\") is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "dbt reference trailing space defers to python",
+            description: "a dbt reference trailing space is rejected",
             sql: "SELECT * FROM __dbt_ref(\"shop\", \"customers\" )",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "dbt_ref",
+                    "__dbt_ref(\"shop\", \"customers\" )",
+                    14,
+                    "__dbt_ref(\"shop\", \"customers\" ) is not a valid __dbt_ref() call",
+                    "__dbt_ref() takes one double-quoted model name, or a double-quoted package name and model name separated by a comma, with no comments or extra spaces inside the parentheses: __dbt_ref(\"shop\", \"customers\")",
+                    "__dbt_ref(\"shop\", \"customers\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "dbt reference third name defers to python",
+            description: "a dbt reference third name is rejected",
             sql: "SELECT * FROM __dbt_ref(\"a\", \"b\", \"c\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "dbt_ref",
+                    "__dbt_ref(\"a\", \"b\", \"c\")",
+                    14,
+                    "__dbt_ref(\"a\", \"b\", \"c\") is not a valid __dbt_ref() call",
+                    "__dbt_ref() takes one double-quoted model name, or a double-quoted package name and model name separated by a comma, with no comments or extra spaces inside the parentheses: __dbt_ref(\"package_name\", \"model_name\")",
+                    "__dbt_ref(\"package_name\", \"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "non-ascii space around a dbt comma defers to python",
+            description: "non-ascii space around a dbt comma is python whitespace",
             sql: "SELECT * FROM __dbt_ref(\"shop\"\u{a0}, \"customers\")",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(vec![dbt_reference("shop", "customers")], vec![]),
         },
         ExtractSqlReferencesTestCase {
-            description: "table function without call suffix defers to python",
+            description: "a table function without a call suffix is rejected",
             sql: "SELECT * FROM __table_fn(\"f\") /* x */ (1)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "table_fn",
+                    "__table_fn(\"f\")",
+                    14,
+                    "__table_fn must be followed by an argument list",
+                    "pass the function arguments in a second set of parentheses, using () for no arguments: __table_fn(\"f\")()",
+                    "__table_fn(\"f\")()",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "single quoted table function name defers to python",
+            description: "a single quoted table function name is rejected",
             sql: "SELECT * FROM __table_fn('f')(1)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "table_fn",
+                    "__table_fn('f')",
+                    14,
+                    "__table_fn('f') is not a valid __table_fn() call",
+                    "__table_fn() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __table_fn(\"f\")(...)",
+                    "__table_fn(\"f\")(...)",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "nested reference name defers to python",
+            description: "a nested reference name is rejected",
             sql: "SELECT * FROM __ref(__ref(\"a\"))",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(__ref(\"a\"))",
+                    14,
+                    "__ref(__ref(\"a\")) is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "non-ascii identifier defers to python",
-            sql: "SELECT * FROM __ref(commandés)",
+            description: "a non-ascii identifier is rejected at its code-point offset",
+            sql: "SELECT '\u{e9}', __ref(command\u{e9}s)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![],
+                vec![invalid_call(
+                    "ref",
+                    "__ref(command\u{e9}s)",
+                    12,
+                    "__ref(command\u{e9}s) is not a valid __ref() call",
+                    "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"model_name\")",
+                    "__ref(\"model_name\")",
+                )],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "an earlier valid reference does not stop a later deferral",
-            sql: "SELECT * FROM __ref(\"a\") JOIN __ref(b)",
+            description: "rejected calls and references keep authored order",
+            sql: "SELECT * FROM __ref(\"a\") JOIN __ref(b) JOIN __udf(c, d)(1)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(
+                vec![reference("ref", "a")],
+                vec![
+                    invalid_call(
+                        "ref",
+                        "__ref(b)",
+                        30,
+                        "__ref(b) is not a valid __ref() call",
+                        "__ref() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __ref(\"b\")",
+                        "__ref(\"b\")",
+                    ),
+                    invalid_call(
+                        "udf",
+                        "__udf(c, d)",
+                        44,
+                        "__udf(c, d) is not a valid __udf() call",
+                        "__udf() takes exactly one double-quoted name, with no comments or extra spaces inside the parentheses: __udf(\"function_name\")",
+                        "__udf(\"function_name\")",
+                    ),
+                ],
+            ),
         },
         ExtractSqlReferencesTestCase {
-            description: "non-ascii whitespace before a call suffix defers",
+            description: "non-ascii whitespace before a call suffix is python whitespace",
             sql: "SELECT * FROM __table_fn(\"f\")\u{a0}(1)",
             syntax: generic(),
-            expected_extraction: ReferenceExtraction::Deferred,
+            expected_extraction: rejected(vec![table_function("f", 1)], vec![]),
         },
         ExtractSqlReferencesTestCase {
             description: "python whitespace includes ascii separators",
diff --git a/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py b/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
index e9ba5e96d3..8de8966c06 100644
--- a/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
+++ b/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
@@ -3,10 +3,17 @@
 from scripts.compiler_differential._helpers.corpus.case_builder import (
     failure_case,
     mart_body_files,
+    staging_files,
 )
 from scripts.compiler_differential.models import FailureCase
 
 _SELECT: str = "SELECT order_id AS customer_id, 1 AS total_amount\n"
+_STAGING_HEADER: str = 'MODEL (\n  description "Staged orders",\n);\n\n'
+_HIDDEN_CALLS: str = (
+    "-- __ref(hidden)\n"
+    "SELECT order_id AS customer_id, 'é __ref(hidden)' AS note, $$__ref(hidden)$$ AS body,\n"
+    "  1 AS total_amount /* __ref(hidden) */\n"
+)
 
 
 def _mart_from(
@@ -78,4 +85,41 @@ def reference_failure_cases() -> tuple[FailureCase, ...]:
             expected_code="P001",
             expected_message="SQL table function call contains an unclosed parenthesis",
         ),
+        failure_case(
+            name="reference-several-rejected-calls-in-one-file",
+            expected_code="P012",
+            expected_message="__ref(stg_orders) is not a valid __ref() call",
+            expected_help=(
+                "__ref() takes exactly one double-quoted name, with no comments or extra spaces "
+                'inside the parentheses: __ref("stg_orders")'
+            ),
+            expected_location=(8, 6),
+            expected_codes=("P012", "P012", "P012"),
+            files=mart_body_files(
+                f"{_HIDDEN_CALLS}FROM __ref(stg_orders)\n"
+                "JOIN __source( 'raw_orders' ) USING (customer_id)\n"
+                'WHERE customer_id IN (SELECT customer_id FROM __seed("a", "b"))\n'
+            ),
+        ),
+        failure_case(
+            name="reference-rejected-calls-across-files",
+            expected_code="P012",
+            expected_message="__dbt_ref(shop, orders, extra) is not a valid __dbt_ref() call",
+            expected_codes=("P012", "P012"),
+            files={
+                **mart_body_files(f"{_SELECT}FROM __dbt_ref(shop, orders, extra)\n"),
+                **staging_files(
+                    f"{_STAGING_HEADER}SELECT order_id, customer_id, amount, status\n"
+                    'FROM __table_fn("table_fn__orders_for") -- no call\n'
+                ),
+            },
+        ),
+        failure_case(
+            name="reference-error-after-rejected-calls",
+            expected_code="P001",
+            expected_message="SQL reference contains an unclosed quoted string",
+            files=mart_body_files(
+                f"{_SELECT}FROM __ref(stg_orders)\nWHERE __ref(other) IS NULL AND note = 'open\n"
+            ),
+        ),
     )
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index af327366cc..fbc54f99b1 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -206,7 +206,17 @@ class SqlReferenceScanner:
     def __init__(self, syntax: dict[str, object]) -> None: ...
     def extract(
         self, sql: str
-    ) -> tuple[list[tuple[str, str, str | None, int | None]], None] | tuple[None, str] | None: ...
+    ) -> (
+        tuple[
+            tuple[
+                list[tuple[str, str, str | None, int | None]],
+                list[tuple[str, str, int, str, str, str]],
+            ],
+            None,
+        ]
+        | tuple[None, str]
+        | None
+    ): ...
 
 def scope_expected_model_names(
     sqls: list[str], syntax: dict[str, object]
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/native.py b/src/sqlbuild/compiler/compile/_helpers/refs/native.py
index ad62a54a17..ddb7ebd5f2 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/native.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/native.py
@@ -5,35 +5,63 @@
 from functools import cache
 
 import sqlbuild._native as _native
-from sqlbuild.compiler.compile.models import CompileSqlReference
+from sqlbuild.compiler.compile.models import (
+    CompileSqlReference,
+    InvalidSqlReferenceCall,
+    SqlReferenceScan,
+)
 from sqlbuild.compiler.references.types import SqlReferenceKind
 from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 
+type _NativeExtraction = (
+    tuple[
+        tuple[
+            list[tuple[str, str, str | None, int | None]],
+            list[tuple[str, str, int, str, str, str]],
+        ],
+        None,
+    ]
+    | tuple[None, str]
+    | None
+)
+
 
 def extract_native_sql_references(
     *, sql: str, syntax: SqlLexicalSyntax
-) -> tuple[CompileSqlReference, ...] | str | None:
-    """Return the references, the message Python raises, or None where Python must extract."""
+) -> SqlReferenceScan | str | None:
+    """Return the references and rejected calls, Python's error message, or None for Python."""
 
     try:
-        extraction: (
-            tuple[list[tuple[str, str, str | None, int | None]], None] | tuple[None, str] | None
-        ) = _scanner(syntax).extract(sql)
+        extraction: _NativeExtraction = _scanner(syntax).extract(sql)
     except UnicodeError:
         return None
     if extraction is None:
         return None
-    references, message = extraction
-    if references is None:
+    scanned, message = extraction
+    if scanned is None:
         return message
-    return tuple(
-        CompileSqlReference(
-            ref_kind=SqlReferenceKind(kind),
-            ref_name=name,
-            ref_package=package,
-            call_argument_count=call_argument_count,
-        )
-        for kind, name, package, call_argument_count in references
+    references, invalid_calls = scanned
+    return SqlReferenceScan(
+        references=tuple(
+            CompileSqlReference(
+                ref_kind=SqlReferenceKind(kind),
+                ref_name=name,
+                ref_package=package,
+                call_argument_count=call_argument_count,
+            )
+            for kind, name, package, call_argument_count in references
+        ),
+        invalid_calls=tuple(
+            InvalidSqlReferenceCall(
+                ref_kind=SqlReferenceKind(kind),
+                call=call,
+                start=start,
+                message=call_message,
+                help=help_text,
+                corrected_call=corrected_call,
+            )
+            for kind, call, start, call_message, help_text, corrected_call in invalid_calls
+        ),
     )
 
 
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/references.py b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
index cf397055bf..7d95d568ea 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/references.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
@@ -16,7 +16,7 @@
     SQL_QUOTE_TOKENS,
     SQL_REFERENCE_NAME_QUOTE_TOKENS,
 )
-from sqlbuild.compiler.compile.exceptions import CompileInputError
+from sqlbuild.compiler.compile.exceptions import CompileInputError, SqlReferenceExtractionError
 from sqlbuild.compiler.compile.models import (
     CompilerDiagnostic,
     CompileSqlReference,
@@ -25,10 +25,8 @@
     SqlReferenceScan,
 )
 from sqlbuild.compiler.compile.types import DiagnosticPhase, DiagnosticSeverity
-from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
-from sqlbuild.compiler.frontier.exceptions import NativeStageMismatchError
 from sqlbuild.compiler.frontier.main.native_stage_enabled import native_stage_enabled
-from sqlbuild.compiler.frontier.types import CompilerEngine, NativeStage
+from sqlbuild.compiler.frontier.types import NativeStage
 from sqlbuild.compiler.references.types import SqlReferenceKind
 from sqlbuild.compiler.sql_analysis.main._find_matching_paren import find_matching_paren
 from sqlbuild.compiler.sql_analysis.main._skip_dialect_non_code import dialect_non_code_end
@@ -99,38 +97,14 @@ def scan_sql_reference_calls(*, sql: str, syntax: SqlLexicalSyntax) -> SqlRefere
 
 
 def _scan_sql_references_natively(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan:
-    """Return the native refs; a native error stands only when Python raises the same error."""
+    """Return the native scan or raise Python's error; Python scans the text native defers."""
 
-    native_references: tuple[CompileSqlReference, ...] | str | None = extract_native_sql_references(
-        sql=sql, syntax=syntax
-    )
-    if native_references is None:
+    native: SqlReferenceScan | str | None = extract_native_sql_references(sql=sql, syntax=syntax)
+    if native is None:
         return _scan_sql_references_with_python(sql=sql, syntax=syntax)
-    if not isinstance(native_references, str):
-        return SqlReferenceScan(references=native_references)
-    try:
-        python_scan: SqlReferenceScan = _scan_sql_references_with_python(sql=sql, syntax=syntax)
-    except CompileInputError as error:
-        if str(error) == native_references:
-            raise
-        raise _reference_mismatch(
-            native_message=native_references, python_outcome=f"raised {str(error)!r}"
-        ) from error
-    raise _reference_mismatch(
-        native_message=native_references,
-        python_outcome=(
-            f"found {len(python_scan.references)} reference(s) and "
-            f"{len(python_scan.invalid_calls)} rejected call(s)"
-        ),
-    )
-
-
-def _reference_mismatch(*, native_message: str, python_outcome: str) -> NativeStageMismatchError:
-    return NativeStageMismatchError(
-        f"native {NativeStage.REFERENCE_EXTRACTION} raised {native_message!r} but the Python "
-        f"compiler {python_outcome}; rerun with {COMPILER_ENGINE_ENV_VAR}="
-        f"{CompilerEngine.PYTHON} and report this mismatch"
-    )
+    if isinstance(native, str):
+        raise SqlReferenceExtractionError(native)
+    return native
 
 
 def report_invalid_reference_calls(
@@ -203,8 +177,10 @@ def _authored_invalid_call_starts(
     if origin is None:
         return {}
     try:
-        authored: SqlReferenceScan = _scan_sql_references_with_python(
-            sql=origin.contents, syntax=syntax
+        authored: SqlReferenceScan = (
+            _scan_sql_references_natively(sql=origin.contents, syntax=syntax)
+            if native_stage_enabled(NativeStage.REFERENCE_EXTRACTION)
+            else _scan_sql_references_with_python(sql=origin.contents, syntax=syntax)
         )
     except CompileInputError:
         return {}
diff --git a/src/sqlbuild/compiler/compile/exceptions.py b/src/sqlbuild/compiler/compile/exceptions.py
index 9d7946a898..3fe13453af 100644
--- a/src/sqlbuild/compiler/compile/exceptions.py
+++ b/src/sqlbuild/compiler/compile/exceptions.py
@@ -67,5 +67,9 @@ class CompactAnalysisInputError(CompileInputError):
     """Raised when compact SQL-analysis batch inputs are inconsistent."""
 
 
+class SqlReferenceExtractionError(CompileInputError):
+    """Raised when SQL text cannot be scanned for its reference calls."""
+
+
 class AnalysisCacheEntryError(ValueError):
     """Raised when a persisted model analysis cache entry is invalid."""
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
index c0c70c7230..aefa85830c 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
@@ -11,7 +11,8 @@ class CraftedReferenceParityTestCase:
 
     description: str
     sql: str
-    expected_deferred: bool = False
+    expected_rejected: int = 0
+    expected_failed: int = 0
 
 
 @dataclass(frozen=True)
@@ -42,13 +43,23 @@ class ReferenceEngineTestCase:
 
 
 @dataclass(frozen=True)
-class ReferenceMismatchTestCase:
-    """A native error that Python does not raise identically."""
+class NativeReferenceErrorTestCase:
+    """Reference SQL whose native error must stand without a Python re-scan."""
 
     description: str
     sql: str
-    native_message: str
-    expected_python_outcome: str
+    expected_message: str
+
+
+@dataclass(frozen=True)
+class ReferenceDiagnosticParityTestCase:
+    """Generated authored files whose rejected calls both engines must report identically."""
+
+    description: str
+    syntax: str
+    seed: int
+    count: int
+    expected_minimum_located: int
 
 
 @dataclass(frozen=True)
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
index 15a36f2bee..7c94059efc 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
@@ -5,20 +5,34 @@
 import random
 from dataclasses import dataclass
 from itertools import chain, compress
+from operator import attrgetter, itemgetter
+from pathlib import Path
 from typing import cast
 
+import pytest
+
 from sqlbuild.adapters.bigquery.classes.bigquery_adapter import BigQueryAdapter
 from sqlbuild.adapters.databricks.classes.databricks_adapter import DatabricksAdapter
 from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter
 from sqlbuild.adapters.postgres.classes.postgres_adapter import PostgresAdapter
 from sqlbuild.adapters.snowflake.classes.snowflake_adapter import SnowflakeAdapter
 from sqlbuild.adapters.sqlserver.classes.sqlserver_adapter import SqlServerAdapter
+from sqlbuild.compiler.compile._helpers.diagnostics.collector import collect_compile_diagnostics
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
 from sqlbuild.compiler.compile._helpers.refs.references import (
     _scan_sql_references_with_python,
+    extract_sql_references,
 )
 from sqlbuild.compiler.compile.exceptions import CompileInputError
-from sqlbuild.compiler.compile.models import CompileSqlReference, SqlReferenceScan
+from sqlbuild.compiler.compile.models import (
+    CompilerDiagnostic,
+    CompileSqlReference,
+    SqlReferenceOrigin,
+    SqlReferenceScan,
+)
+from sqlbuild.compiler.compile.types import CompiledResourceType
+from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
+from sqlbuild.compiler.frontier.types import CompilerEngine
 from sqlbuild.compiler.references.types import SqlReferenceKind
 from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from tests.integration.src.sqlbuild.compiler.helpers import mismatches
@@ -152,7 +166,7 @@ def _pick(*, rng: random.Random, usual: tuple[str, ...], rare: tuple[str, ...])
 
 @dataclass(frozen=True)
 class ReferenceParity:
-    """How native extraction compared with Python over one corpus."""
+    """How native extraction compared with Python over one corpus; counts are native's."""
 
     mismatches: list[tuple[object, object, object]]
     extracted: int
@@ -186,34 +200,26 @@ def _generated_reference_call(*, rng: random.Random) -> str:
     )
 
 
-def python_outcome(
-    *, sql: str, syntax: SqlLexicalSyntax
-) -> tuple[CompileSqlReference, ...] | str | None:
-    """Return the oracle's references, its error message, or None when it rejects a call."""
+def python_outcome(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan | str:
+    """Return the oracle's references and rejected calls, or its error message."""
 
     try:
-        scan: SqlReferenceScan = _scan_sql_references_with_python(sql=sql, syntax=syntax)
+        return _scan_sql_references_with_python(sql=sql, syntax=syntax)
     except CompileInputError as error:
         return str(error)
-    return (scan.references, None)[bool(scan.invalid_calls)]
 
 
 def reference_parity(*, sqls: list[str], syntax: SqlLexicalSyntax) -> ReferenceParity:
-    """Compare native extraction with Python; native must defer every SQL Python rejects."""
+    """Compare native extraction, rejected calls and errors with Python wherever native scans."""
 
-    native: list[tuple[CompileSqlReference, ...] | str | None] = [
+    native: list[SqlReferenceScan | str | None] = [
         extract_native_sql_references(sql=sql, syntax=syntax) for sql in sqls
     ]
-    python: list[tuple[CompileSqlReference, ...] | str | None] = [
-        python_outcome(sql=sql, syntax=syntax) for sql in sqls
-    ]
-    scanned: list[bool] = [
-        native_outcome is not None or python_result is None
-        for native_outcome, python_result in zip(native, python, strict=True)
-    ]
-    extracted: list[tuple[CompileSqlReference, ...]] = cast(
-        list[tuple[CompileSqlReference, ...]],
-        list(compress(native, [isinstance(outcome, tuple) for outcome in native])),
+    python: list[SqlReferenceScan | str] = [python_outcome(sql=sql, syntax=syntax) for sql in sqls]
+    scanned: list[bool] = [outcome is not None for outcome in native]
+    extracted: list[SqlReferenceScan] = cast(
+        list[SqlReferenceScan],
+        list(compress(native, [isinstance(outcome, SqlReferenceScan) for outcome in native])),
     )
     return ReferenceParity(
         mismatches=mismatches(
@@ -221,12 +227,81 @@ def reference_parity(*, sqls: list[str], syntax: SqlLexicalSyntax) -> ReferenceP
             expected=list(compress(python, scanned)),
             actual=list(compress(native, scanned)),
         ),
-        extracted=len(extracted),
+        extracted=sum(not scan.invalid_calls for scan in extracted),
         failed=sum(isinstance(outcome, str) for outcome in native),
-        deferred=native.count(None) - python.count(None),
-        rejected=python.count(None),
+        deferred=native.count(None),
+        rejected=sum(len(scan.invalid_calls) for scan in extracted),
         table_functions=sum(
             reference.ref_kind is SqlReferenceKind.TABLE_FUNCTION
-            for reference in chain.from_iterable(extracted)
+            for reference in chain.from_iterable(map(attrgetter("references"), extracted))
         ),
     )
+
+
+_FILE_HEADERS: tuple[str, ...] = (
+    'MODEL (description "Orders");\n\n',
+    'MODEL (\n  description "Orders",\n);\n-- notes\n\n',
+    'MODEL (description "Ordres é");\n/* commentaire */\n',
+)
+
+
+def generated_reference_files(*, rng: random.Random, count: int) -> list[tuple[str, str]]:
+    """Return seeded authored files as `(relative path, SQL body)` pairs."""
+
+    return [
+        (f"models/area_{index % 7}/orders_{index}.sql", body)
+        for index, body in enumerate(generated_reference_sqls(rng=rng, count=count))
+    ]
+
+
+def reported_reference_outcomes(
+    *,
+    files: list[tuple[str, str]],
+    syntax: SqlLexicalSyntax,
+    engine: CompilerEngine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> list[object]:
+    """Extract each file's body under `engine`, keeping its references, diagnostics and error."""
+
+    monkeypatch.setenv(COMPILER_ENGINE_ENV_VAR, engine.value)
+    return [
+        _reported_reference_outcome(
+            origin=SqlReferenceOrigin(
+                file_path=Path("/project") / relative_path,
+                relative_path=Path(relative_path),
+                contents=_FILE_HEADERS[index % len(_FILE_HEADERS)] + body,
+                resource_type=CompiledResourceType.MODEL,
+                resource_name=Path(relative_path).stem,
+            ),
+            sql=body,
+            syntax=syntax,
+        )
+        for index, (relative_path, body) in enumerate(files)
+    ]
+
+
+def _reported_reference_outcome(
+    *, origin: SqlReferenceOrigin, sql: str, syntax: SqlLexicalSyntax
+) -> tuple[tuple[CompileSqlReference, ...], tuple[CompilerDiagnostic, ...], str, str | None]:
+    references: tuple[CompileSqlReference, ...] = ()
+    with collect_compile_diagnostics() as collected:
+        try:
+            references = extract_sql_references(sql=sql, syntax=syntax, origin=origin)
+        except CompileInputError as error:
+            return (references, tuple(collected.diagnostics), str(error), error.code)
+    return (references, tuple(collected.diagnostics), "", None)
+
+
+def located_diagnostic_count(*, outcomes: list[object]) -> int:
+    """Count the reported diagnostics that carry an authored location."""
+
+    diagnostics: list[tuple[CompilerDiagnostic, ...]] = list(
+        map(itemgetter(1), cast(list[tuple[object, tuple[CompilerDiagnostic, ...]]], outcomes))
+    )
+    return sum(diagnostic.location is not None for diagnostic in chain.from_iterable(diagnostics))
+
+
+def python_scan_not_expected(**_: object) -> SqlReferenceScan:
+    """A replacement for the Python scanner that fails the test when it is called."""
+
+    raise AssertionError("the Python reference scanner ran")
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
index e2940592d7..1e7268db92 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
@@ -1,4 +1,4 @@
-"""Native reference extraction returns Python's references or error exactly, or defers to Python."""
+"""Native reference extraction returns Python's references, rejected calls and errors exactly."""
 
 from __future__ import annotations
 
@@ -13,25 +13,30 @@
 from sqlbuild.compiler.compile._helpers.refs import references
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
 from sqlbuild.compiler.compile._helpers.refs.references import extract_sql_references
-from sqlbuild.compiler.compile.exceptions import CompileInputError
-from sqlbuild.compiler.compile.models import CompileSqlReference
+from sqlbuild.compiler.compile.exceptions import CompileInputError, SqlReferenceExtractionError
+from sqlbuild.compiler.compile.models import SqlReferenceScan
 from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
-from sqlbuild.compiler.frontier.exceptions import NativeStageMismatchError
 from sqlbuild.compiler.frontier.types import CompilerEngine
 from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 from tests.integration.src.sqlbuild.compiler.compile._helpers.refs._test_types import (
     CraftedReferenceParityTestCase,
     GeneratedReferenceParityTestCase,
+    NativeReferenceErrorTestCase,
+    ReferenceDiagnosticParityTestCase,
     ReferenceEngineTestCase,
-    ReferenceMismatchTestCase,
     ReferenceScanBoundTestCase,
 )
 from tests.integration.src.sqlbuild.compiler.compile._helpers.refs.helpers import (
     LEXICAL_SYNTAXES,
     ReferenceParity,
+    generated_reference_files,
     generated_reference_sqls,
+    located_diagnostic_count,
+    python_scan_not_expected,
     reference_parity,
+    reported_reference_outcomes,
 )
+from tests.integration.src.sqlbuild.compiler.helpers import mismatches
 
 _GENERIC_SYNTAX: SqlLexicalSyntax = LEXICAL_SYNTAXES["generic"]
 _ENGINE_SQLS: tuple[
@@ -66,7 +71,7 @@
 )
 _NESTED_CALLS: str = '__table_fn("orders_for")(' * 1_000 + "1" + ")" * 1_000
 _OUTCOME_SUMMARIES: dict[type, Callable[[Any], int | str]] = {
-    tuple: len,
+    SqlReferenceScan: lambda scan: len(scan.references) + len(scan.invalid_calls),
     str: str,
     type(None): repr,
 }
@@ -112,12 +117,12 @@
                 "SELECT 'it\\'s __ref(a)', E'x\\'__ref(b)', r'C:\\' AS p, '''__ref(c)''', "
                 '/* x /* y */ __ref(d) */ 1 # __ref(e)\n// __ref(f)\nFROM __ref("g")'
             ),
-            expected_deferred=True,
+            expected_rejected=4,
         ),
         CraftedReferenceParityTestCase(
             description="comments_inside_calls_are_rejected",
             sql='SELECT * FROM __ref("a" /* note */ "b") JOIN __ref(-- c\n"orders")',
-            expected_deferred=True,
+            expected_rejected=2,
         ),
         CraftedReferenceParityTestCase(
             description="non_ascii_quoted_names_and_text",
@@ -126,69 +131,71 @@
         CraftedReferenceParityTestCase(
             description="wrong_reference_argument_count_is_rejected",
             sql="SELECT * FROM __ref(orders, customers)",
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
             description="unquoted_and_single_quoted_names_are_rejected",
             sql="SELECT * FROM __ref(orders) JOIN __source('raw_orders')",
-            expected_deferred=True,
+            expected_rejected=2,
         ),
         CraftedReferenceParityTestCase(
             description="spaces_inside_calls_are_rejected",
             sql='SELECT * FROM __seed( "regions" ) JOIN __dbt_ref("shop", "orders" )',
-            expected_deferred=True,
+            expected_rejected=2,
         ),
         CraftedReferenceParityTestCase(
-            description="unclosed_text_after_a_rejected_call_is_rejected",
+            description="unclosed_text_after_a_rejected_call_fails",
             sql="SELECT * FROM __ref(orders) WHERE note = 'unclosed",
-            expected_deferred=True,
+            expected_failed=1,
         ),
         CraftedReferenceParityTestCase(
             description="unclosed_text_before_a_rejected_call_fails",
             sql='SELECT * FROM __ref("orders" WHERE __ref(customers)',
+            expected_failed=1,
         ),
         CraftedReferenceParityTestCase(
             description="wrong_dbt_reference_argument_count_is_rejected",
             sql="SELECT * FROM __dbt_ref(shop, orders, extra)",
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
             description="empty_table_function_call_argument",
             sql='SELECT * FROM __table_fn("orders_for")(1,,2)',
+            expected_failed=1,
         ),
         CraftedReferenceParityTestCase(
             description="table_function_without_call_is_rejected",
             sql='SELECT * FROM __table_fn("orders_for") /* gap */ (1)',
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
             description="unquoted_table_function_name_is_rejected",
             sql="SELECT * FROM __table_fn(orders_for)(1)",
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
             description="unclosed_table_function_call",
             sql='SELECT * FROM __table_fn("orders_for")(1',
+            expected_failed=1,
         ),
         CraftedReferenceParityTestCase(
             description="nested_reference_as_name_is_rejected",
             sql="SELECT * FROM __ref(__ref(orders))",
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
-            description="non_ascii_identifier_defers",
+            description="non_ascii_identifier_is_rejected",
             sql="SELECT * FROM __ref(commandés)",
-            expected_deferred=True,
+            expected_rejected=1,
         ),
         CraftedReferenceParityTestCase(
-            description="unicode_whitespace_before_call_defers",
+            description="unicode_whitespace_before_call_is_skipped",
             sql='SELECT * FROM __table_fn("orders_for")\u00a0(1)',
-            expected_deferred=True,
         ),
     ],
     ids=lambda case: case.description,
 )
-def test_given_crafted_reference_sql_when_extracting_natively_then_matches_python_or_defers(
+def test_given_crafted_reference_sql_when_extracting_natively_then_matches_python(
     test_case: CraftedReferenceParityTestCase,
 ) -> None:
     parity: ReferenceParity = reference_parity(sqls=[test_case.sql], syntax=_GENERIC_SYNTAX)
@@ -197,8 +204,12 @@ def test_given_crafted_reference_sql_when_extracting_natively_then_matches_pytho
         for syntax in LEXICAL_SYNTAXES.values()
     ]
 
-    assert dialect_mismatches == [[]] * len(LEXICAL_SYNTAXES)
-    assert parity.deferred + parity.rejected == int(test_case.expected_deferred)
+    assert (dialect_mismatches, parity.deferred, parity.rejected, parity.failed) == (
+        [[]] * len(LEXICAL_SYNTAXES),
+        0,
+        test_case.expected_rejected,
+        test_case.expected_failed,
+    )
 
 
 @pytest.mark.parametrize(
@@ -212,14 +223,14 @@ def test_given_crafted_reference_sql_when_extracting_natively_then_matches_pytho
             expected_minimum_extracted=700,
             expected_minimum_failed=500,
             expected_minimum_table_functions=300,
-            expected_minimum_rejected=1000,
-            expected_maximum_deferred=400,
+            expected_minimum_rejected=1400,
+            expected_maximum_deferred=0,
         )
         for offset, syntax in enumerate(LEXICAL_SYNTAXES)
     ],
     ids=lambda case: case.description,
 )
-def test_given_generated_reference_sql_when_extracting_natively_then_matches_python_or_defers(
+def test_given_generated_reference_sql_when_extracting_natively_then_matches_python(
     test_case: GeneratedReferenceParityTestCase,
 ) -> None:
     sqls: list[str] = generated_reference_sqls(
@@ -260,13 +271,11 @@ def test_given_engine_when_extracting_references_then_only_preview_runs_native_s
 ) -> None:
     monkeypatch.setenv(COMPILER_ENGINE_ENV_VAR, test_case.engine.value)
     native_calls: list[str] = []
-    native_extract: Callable[..., tuple[CompileSqlReference, ...] | str | None] = (
+    native_extract: Callable[..., SqlReferenceScan | str | None] = (
         references.extract_native_sql_references
     )
 
-    def counting_extract(
-        *, sql: str, syntax: SqlLexicalSyntax
-    ) -> tuple[CompileSqlReference, ...] | str | None:
+    def counting_extract(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan | str | None:
         native_calls.append(sql)
         return native_extract(sql=sql, syntax=syntax)
 
@@ -292,43 +301,73 @@ def counting_extract(
 @pytest.mark.parametrize(
     "test_case",
     [
-        ReferenceMismatchTestCase(
-            description="python_extracts_references",
-            sql='SELECT * FROM __ref("orders")',
-            native_message="SQL reference contains an empty argument",
-            expected_python_outcome="found 1 reference(s) and 0 rejected call(s)",
+        NativeReferenceErrorTestCase(
+            description="unclosed_reference_call",
+            sql='SELECT * FROM __ref("orders"',
+            expected_message="SQL reference contains an unclosed parenthesis",
         ),
-        ReferenceMismatchTestCase(
-            description="python_rejects_a_call",
-            sql="SELECT * FROM __ref(orders)",
-            native_message="SQL reference contains an empty argument",
-            expected_python_outcome="found 0 reference(s) and 1 rejected call(s)",
+        NativeReferenceErrorTestCase(
+            description="empty_table_function_argument_after_a_rejected_call",
+            sql='SELECT * FROM __ref(orders) JOIN __table_fn("orders_for")(1,,2)',
+            expected_message="SQL reference contains an empty argument",
         ),
-        ReferenceMismatchTestCase(
-            description="python_raises_a_different_error",
-            sql='SELECT * FROM __ref("orders"',
-            native_message="SQL reference contains an empty argument",
-            expected_python_outcome="raised 'SQL reference contains an unclosed parenthesis'",
+        NativeReferenceErrorTestCase(
+            description="unclosed_quote_after_references",
+            sql='SELECT * FROM __ref("orders") WHERE note = \'open',
+            expected_message="SQL reference contains an unclosed quoted string",
         ),
     ],
     ids=lambda case: case.description,
 )
-def test_given_native_error_python_disagrees_with_when_extracting_then_raises_mismatch(
-    test_case: ReferenceMismatchTestCase, monkeypatch: pytest.MonkeyPatch
+def test_given_native_reference_error_when_extracting_then_raised_without_python_rescan(
+    test_case: NativeReferenceErrorTestCase, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     monkeypatch.setenv(COMPILER_ENGINE_ENV_VAR, CompilerEngine.NATIVE_PREVIEW.value)
+    monkeypatch.setattr(references, "_scan_sql_references_with_python", python_scan_not_expected)
 
-    def failing_extract(*, sql: str, syntax: SqlLexicalSyntax) -> str:
-        del sql, syntax
-        return test_case.native_message
+    with pytest.raises(SqlReferenceExtractionError) as raised:
+        _ = extract_sql_references(sql=test_case.sql, syntax=_GENERIC_SYNTAX)
 
-    monkeypatch.setattr(references, "extract_native_sql_references", failing_extract)
+    assert (str(raised.value), raised.value.code, raised.value.help) == (
+        test_case.expected_message,
+        "P001",
+        None,
+    )
 
-    with pytest.raises(NativeStageMismatchError) as raised:
-        _ = extract_sql_references(sql=test_case.sql, syntax=_GENERIC_SYNTAX)
 
-    assert test_case.expected_python_outcome in str(raised.value)
-    assert repr(test_case.native_message) in str(raised.value)
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ReferenceDiagnosticParityTestCase(
+            description=f"{syntax}_syntax",
+            syntax=syntax,
+            seed=20261009 + offset,
+            count=400,
+            expected_minimum_located=150,
+        )
+        for offset, syntax in enumerate(LEXICAL_SYNTAXES)
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_generated_files_when_reporting_rejected_calls_then_engines_report_identically(
+    test_case: ReferenceDiagnosticParityTestCase, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    files: list[tuple[str, str]] = generated_reference_files(
+        rng=random.Random(test_case.seed), count=test_case.count
+    )
+    syntax: SqlLexicalSyntax = LEXICAL_SYNTAXES[test_case.syntax]
+
+    python: list[object] = reported_reference_outcomes(
+        files=files, syntax=syntax, engine=CompilerEngine.PYTHON, monkeypatch=monkeypatch
+    )
+    preview: list[object] = reported_reference_outcomes(
+        files=files, syntax=syntax, engine=CompilerEngine.NATIVE_PREVIEW, monkeypatch=monkeypatch
+    )
+
+    assert (
+        mismatches(inputs=[body for _, body in files], expected=python, actual=preview),
+        located_diagnostic_count(outcomes=python) >= test_case.expected_minimum_located,
+    ) == ([], True)
 
 
 @pytest.mark.parametrize(
@@ -361,7 +400,7 @@ def failing_extract(*, sql: str, syntax: SqlLexicalSyntax) -> str:
         ReferenceScanBoundTestCase(
             description="rejected_call_after_fifty_thousand_references",
             sql=_REJECTED_AFTER_MANY,
-            expected_outcome="None",
+            expected_outcome=50_001,
             expected_maximum_seconds=0.5,
         ),
     ],
@@ -371,7 +410,7 @@ def test_given_worst_case_reference_sql_when_extracting_natively_then_finishes_q
     test_case: ReferenceScanBoundTestCase,
 ) -> None:
     started: float = time.thread_time()
-    outcome: tuple[CompileSqlReference, ...] | str | None = extract_native_sql_references(
+    outcome: SqlReferenceScan | str | None = extract_native_sql_references(
         sql=test_case.sql, syntax=_GENERIC_SYNTAX
     )
     elapsed: float = time.thread_time() - started

From 2cff80cabbefbc62c101d0ba37650ab9631d701b Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 8 Oct 2026 18:29:00 +0100
Subject: [PATCH 2/3] feat(native): locate reference scan errors and mark them
 bridge independent

---
 .../_helpers/sqltext/sql_references.rs        | 10 +-
 .../src/sql_references/_helpers/scan.rs       | 32 +++++-
 .../main/extract_sql_references.rs            | 16 ++-
 .../src/sql_references/models.rs              | 11 ++-
 .../src/sql_references/tests/helpers.rs       |  9 +-
 .../tests/test_extract_sql_references.rs      | 23 ++++-
 .../corpus/reference_failure_cases.py         | 12 ++-
 src/sqlbuild/_native.pyi                      |  2 +-
 .../compiler/compile/_helpers/refs/cache.py   |  4 +-
 .../compiler/compile/_helpers/refs/native.py  | 11 ++-
 .../compile/_helpers/refs/references.py       | 95 ++++++++++++------
 src/sqlbuild/compiler/compile/exceptions.py   |  4 -
 src/sqlbuild/compiler/compile/types.py        |  3 +
 .../cli/commands/main/compile/_test_types.py  | 10 ++
 .../compile/test_native_reference_errors.py   | 97 +++++++++++++++++++
 .../compile/_helpers/refs/_test_types.py      |  4 +-
 .../compiler/compile/_helpers/refs/helpers.py | 25 ++---
 .../_helpers/refs/test_native_references.py   | 78 +++++++++++----
 18 files changed, 346 insertions(+), 100 deletions(-)
 create mode 100644 tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py

diff --git a/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs b/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
index 64c754ecb8..b16159e080 100644
--- a/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
+++ b/crates/sqlbuild-python/src/bindings/_helpers/sqltext/sql_references.rs
@@ -15,10 +15,10 @@ use crate::bindings::_helpers::sqltext::lexical_syntax::LexicalSyntaxInput;
 type ReferenceRow = (&'static str, String, Option<String>, Option<usize>);
 /// `(kind, call, start, message, help, corrected_call)` for one rejected call.
 type InvalidCallRow = (&'static str, String, usize, String, String, String);
-/// The references and rejected calls, or Python's error message; `None` defers to Python.
+/// The references and rejected calls, or Python's error and its start; `None` defers.
 type ExtractionRow = Option<(
     Option<(Vec<ReferenceRow>, Vec<InvalidCallRow>)>,
-    Option<String>,
+    Option<(String, usize)>,
 )>;
 
 /// Extracts references under one adapter's lexical rules, read once per syntax.
@@ -36,7 +36,7 @@ impl SqlReferenceScanner {
         }
     }
 
-    /// Return `((references, rejected calls), None)`, `(None, message)`, or `None` to defer.
+    /// Return `((references, rejected calls), None)`, `(None, (message, start))`, or `None`.
     fn extract(&self, sql: &str) -> PyResult<ExtractionRow> {
         compiler_guard(|| {
             Ok(match extract_sql_references(sql, &self.syntax) {
@@ -50,7 +50,9 @@ impl SqlReferenceScanner {
                     )),
                     None,
                 )),
-                ReferenceExtraction::Failed(message) => Some((None, Some(message))),
+                ReferenceExtraction::Failed(failure) => {
+                    Some((None, Some((failure.message, failure.start))))
+                }
                 ReferenceExtraction::Deferred => None,
             })
         })
diff --git a/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs b/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
index 2d6bb53df1..733926471e 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/_helpers/scan.rs
@@ -23,6 +23,14 @@ pub(crate) enum Stop {
 
 type Scan<T> = Result<T, Stop>;
 
+/// Why the scan of one text stopped, with the byte offset Python's error points at.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum Stopped {
+    /// Python raises this message for the quote or comment, or the call, starting at the offset.
+    Failed(String, usize),
+    Deferred,
+}
+
 const NON_CODE_START: [bool; 256] = byte_table(NON_CODE_START_BYTES, false);
 const REFERENCE_SCAN_START: [bool; 256] = byte_table(NON_CODE_START_BYTES, true);
 
@@ -45,12 +53,17 @@ enum Parsed {
 }
 
 /// Return every reference and rejected call in `sql` in authored order.
-pub(crate) fn scan_references(sql: &[u8], syntax: &LexicalSyntax) -> Scan<ReferenceScan> {
+pub(crate) fn scan_references(
+    sql: &[u8],
+    syntax: &LexicalSyntax,
+) -> Result<ReferenceScan, Stopped> {
     let mut scan: ReferenceScan = ReferenceScan::default();
     let mut offsets: CharOffsets = CharOffsets::default();
     let mut index = 0;
     while index < sql.len() {
-        if let Some(end) = non_code_end(sql, index, syntax, REFERENCE_CONTEXT)? {
+        if let Some(end) = non_code_end(sql, index, syntax, REFERENCE_CONTEXT)
+            .map_err(|stop| stopped_at(stop, index))?
+        {
             index = end;
             continue;
         }
@@ -68,11 +81,14 @@ pub(crate) fn scan_references(sql: &[u8], syntax: &LexicalSyntax) -> Scan<Refere
             index += 1;
             continue;
         };
-        let (parsed, next) = parse_reference(sql, index, call, syntax)?;
+        let (parsed, next) =
+            parse_reference(sql, index, call, syntax).map_err(|stop| stopped_at(stop, index))?;
         match parsed {
             Parsed::Reference(reference) => scan.references.push(reference),
             Parsed::Invalid(invalid_call) => scan.invalid_calls.push(InvalidReferenceCall {
-                start: offsets.advance(sql, index)?,
+                start: offsets
+                    .advance(sql, index)
+                    .map_err(|stop| stopped_at(stop, index))?,
                 ..invalid_call
             }),
         }
@@ -445,3 +461,11 @@ impl CharOffsets {
         Ok(self.char)
     }
 }
+
+/// Point a stop at `start`: the top-level quote or comment, or the call being parsed.
+fn stopped_at(stop: Stop, start: usize) -> Stopped {
+    match stop {
+        Stop::Failed(message) => Stopped::Failed(message, start),
+        Stop::Deferred => Stopped::Deferred,
+    }
+}
diff --git a/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs b/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
index 36067237ea..5bcc0c0c80 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/main/extract_sql_references.rs
@@ -1,8 +1,8 @@
 //! Logical SQL references under one adapter's lexical rules, or a deferral to Python.
 
-use crate::sql_references::_helpers::scan::{Stop, scan_references};
+use crate::sql_references::_helpers::scan::{Stopped, scan_references};
 use crate::sql_references::constants::SUPPORTED_LINE_COMMENT_PREFIXES;
-use crate::sql_references::models::ReferenceExtraction;
+use crate::sql_references::models::{ReferenceExtraction, ReferenceScanFailure};
 use crate::sql_scan::models::LexicalSyntax;
 
 /// Return the references and rejected calls Python finds in `sql`, or its error message.
@@ -16,7 +16,15 @@ pub fn extract_sql_references(sql: &str, syntax: &LexicalSyntax) -> ReferenceExt
     }
     match scan_references(sql.as_bytes(), syntax) {
         Ok(scan) => ReferenceExtraction::Extracted(scan),
-        Err(Stop::Failed(message)) => ReferenceExtraction::Failed(message),
-        Err(Stop::Deferred) => ReferenceExtraction::Deferred,
+        Err(Stopped::Failed(message, start)) => {
+            sql.get(..start)
+                .map_or(ReferenceExtraction::Deferred, |prefix| {
+                    ReferenceExtraction::Failed(ReferenceScanFailure {
+                        message,
+                        start: prefix.chars().count(),
+                    })
+                })
+        }
+        Err(Stopped::Deferred) => ReferenceExtraction::Deferred,
     }
 }
diff --git a/crates/sqlbuild-sqltext/src/sql_references/models.rs b/crates/sqlbuild-sqltext/src/sql_references/models.rs
index 4a2b435881..72acbeb3ce 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/models.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/models.rs
@@ -32,13 +32,20 @@ pub struct ReferenceScan {
     pub invalid_calls: Vec<InvalidReferenceCall>,
 }
 
+/// Python's reference scan error and the code-point start of the quote, comment or call.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ReferenceScanFailure {
+    pub message: String,
+    pub start: usize,
+}
+
 /// The references of one SQL text, Python's error for it, or a deferral to Python.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum ReferenceExtraction {
     /// Every reference and rejected call in authored order, exactly as Python scans them.
     Extracted(ReferenceScan),
-    /// The message of the `CompileInputError` Python raises for this text.
-    Failed(String),
+    /// The `CompileInputError` Python raises for this text.
+    Failed(ReferenceScanFailure),
     /// A character or syntax rule the scan cannot classify exactly as Python does.
     Deferred,
 }
diff --git a/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs b/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
index c0b7ce9482..794647a5be 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/tests/helpers.rs
@@ -1,7 +1,7 @@
 //! Lexical syntaxes and expected references for reference extraction tests.
 
 use crate::sql_references::models::{
-    InvalidReferenceCall, ReferenceExtraction, ReferenceScan, SqlReference,
+    InvalidReferenceCall, ReferenceExtraction, ReferenceScan, ReferenceScanFailure, SqlReference,
 };
 use crate::sql_scan::models::LexicalSyntax;
 
@@ -83,6 +83,9 @@ pub(crate) fn invalid_call(
     }
 }
 
-pub(crate) fn failed(message: &str) -> ReferenceExtraction {
-    ReferenceExtraction::Failed(message.to_string())
+pub(crate) fn failed(message: &str, start: usize) -> ReferenceExtraction {
+    ReferenceExtraction::Failed(ReferenceScanFailure {
+        message: message.to_string(),
+        start,
+    })
 }
diff --git a/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs b/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
index 90284aebf4..2abecb7591 100644
--- a/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
+++ b/crates/sqlbuild-sqltext/src/sql_references/tests/test_extract_sql_references.rs
@@ -90,25 +90,40 @@ fn given_sql_when_extracting_references_then_matches_python_scanner() {
             description: "empty table function call argument",
             sql: "SELECT * FROM __table_fn(\"f\")(1,,2)",
             syntax: generic(),
-            expected_extraction: failed("SQL reference contains an empty argument"),
+            expected_extraction: failed("SQL reference contains an empty argument", 14),
         },
         ExtractSqlReferencesTestCase {
             description: "unclosed table function call",
             sql: "SELECT * FROM __table_fn(\"f\")(1",
             syntax: generic(),
-            expected_extraction: failed("SQL table function call contains an unclosed parenthesis"),
+            expected_extraction: failed(
+                "SQL table function call contains an unclosed parenthesis",
+                14,
+            ),
         },
         ExtractSqlReferencesTestCase {
             description: "unclosed reference",
             sql: "SELECT * FROM __ref(orders",
             syntax: generic(),
-            expected_extraction: failed("SQL reference contains an unclosed parenthesis"),
+            expected_extraction: failed("SQL reference contains an unclosed parenthesis", 14),
+        },
+        ExtractSqlReferencesTestCase {
+            description: "an unclosed quote inside a call points at the call",
+            sql: "SELECT 'é', __ref(\"a\" || 'open)",
+            syntax: generic(),
+            expected_extraction: failed("SQL reference contains an unclosed quoted string", 12),
+        },
+        ExtractSqlReferencesTestCase {
+            description: "an unclosed block comment after references points at the comment",
+            sql: "SELECT __ref(\"a\") /* é */ /* open",
+            syntax: generic(),
+            expected_extraction: failed("SQL reference contains an unclosed block comment", 26),
         },
         ExtractSqlReferencesTestCase {
             description: "unclosed quoted text",
             sql: "SELECT 'orders",
             syntax: generic(),
-            expected_extraction: failed("SQL reference contains an unclosed quoted string"),
+            expected_extraction: failed("SQL reference contains an unclosed quoted string", 7),
         },
         ExtractSqlReferencesTestCase {
             description: "an unquoted name is rejected",
diff --git a/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py b/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
index 8de8966c06..1f88f333ff 100644
--- a/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
+++ b/scripts/compiler_differential/_helpers/corpus/reference_failure_cases.py
@@ -5,8 +5,10 @@
     mart_body_files,
     staging_files,
 )
+from scripts.compiler_differential.constants import FAILURE_MART_PATH
 from scripts.compiler_differential.models import FailureCase
 
+_MART: str = FAILURE_MART_PATH
 _SELECT: str = "SELECT order_id AS customer_id, 1 AS total_amount\n"
 _STAGING_HEADER: str = 'MODEL (\n  description "Staged orders",\n);\n\n'
 _HIDDEN_CALLS: str = (
@@ -59,7 +61,7 @@ def reference_failure_cases() -> tuple[FailureCase, ...]:
             name="reference-unclosed-call",
             from_clause='__ref("stg_orders"',
             expected_code="P001",
-            expected_message="SQL reference contains an unclosed parenthesis",
+            expected_message=f"{_MART}:6:6: SQL reference contains an unclosed parenthesis",
         ),
         _mart_from(
             name="table-function-without-arguments",
@@ -77,13 +79,15 @@ def reference_failure_cases() -> tuple[FailureCase, ...]:
             name="table-function-empty-call-argument",
             from_clause='__table_fn("table_fn__orders_for")(1, , 2)',
             expected_code="P001",
-            expected_message="SQL reference contains an empty argument",
+            expected_message=f"{_MART}:6:6: SQL reference contains an empty argument",
         ),
         _mart_from(
             name="table-function-unclosed-call",
             from_clause='__table_fn("table_fn__orders_for")(1 -- )',
             expected_code="P001",
-            expected_message="SQL table function call contains an unclosed parenthesis",
+            expected_message=(
+                f"{_MART}:6:6: SQL table function call contains an unclosed parenthesis"
+            ),
         ),
         failure_case(
             name="reference-several-rejected-calls-in-one-file",
@@ -117,7 +121,7 @@ def reference_failure_cases() -> tuple[FailureCase, ...]:
         failure_case(
             name="reference-error-after-rejected-calls",
             expected_code="P001",
-            expected_message="SQL reference contains an unclosed quoted string",
+            expected_message=f"{_MART}:7:39: SQL reference contains an unclosed quoted string",
             files=mart_body_files(
                 f"{_SELECT}FROM __ref(stg_orders)\nWHERE __ref(other) IS NULL AND note = 'open\n"
             ),
diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi
index fbc54f99b1..bb2cb7e6ef 100644
--- a/src/sqlbuild/_native.pyi
+++ b/src/sqlbuild/_native.pyi
@@ -214,7 +214,7 @@ class SqlReferenceScanner:
             ],
             None,
         ]
-        | tuple[None, str]
+        | tuple[None, tuple[str, int]]
         | None
     ): ...
 
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/cache.py b/src/sqlbuild/compiler/compile/_helpers/refs/cache.py
index 1abcdf65f4..a47ec6bca1 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/cache.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/cache.py
@@ -120,7 +120,9 @@ def references(self, *, sql: str, origin: SqlReferenceOrigin | None) -> SqlRefer
                 self._disable()
                 connection = None
 
-        scan: SqlReferenceScan = scan_sql_reference_calls(sql=sql, syntax=self._syntax)
+        scan: SqlReferenceScan = scan_sql_reference_calls(
+            sql=sql, syntax=self._syntax, origin=origin
+        )
         report_invalid_reference_calls(
             invalid_calls=scan.invalid_calls, origin=origin, syntax=self._syntax
         )
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/native.py b/src/sqlbuild/compiler/compile/_helpers/refs/native.py
index ddb7ebd5f2..3e1f24cbb9 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/native.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/native.py
@@ -10,6 +10,7 @@
     InvalidSqlReferenceCall,
     SqlReferenceScan,
 )
+from sqlbuild.compiler.compile.types import SqlReferenceScanFailure
 from sqlbuild.compiler.references.types import SqlReferenceKind
 from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
 
@@ -21,15 +22,15 @@
         ],
         None,
     ]
-    | tuple[None, str]
+    | tuple[None, tuple[str, int]]
     | None
 )
 
 
 def extract_native_sql_references(
     *, sql: str, syntax: SqlLexicalSyntax
-) -> SqlReferenceScan | str | None:
-    """Return the references and rejected calls, Python's error message, or None for Python."""
+) -> SqlReferenceScan | SqlReferenceScanFailure | None:
+    """Return the references and rejected calls, Python's located error, or None for Python."""
 
     try:
         extraction: _NativeExtraction = _scanner(syntax).extract(sql)
@@ -37,9 +38,9 @@ def extract_native_sql_references(
         return None
     if extraction is None:
         return None
-    scanned, message = extraction
+    scanned, failure = extraction
     if scanned is None:
-        return message
+        return failure
     references, invalid_calls = scanned
     return SqlReferenceScan(
         references=tuple(
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/references.py b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
index 7d95d568ea..3cfeac5742 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/references.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
@@ -16,7 +16,7 @@
     SQL_QUOTE_TOKENS,
     SQL_REFERENCE_NAME_QUOTE_TOKENS,
 )
-from sqlbuild.compiler.compile.exceptions import CompileInputError, SqlReferenceExtractionError
+from sqlbuild.compiler.compile.exceptions import CompileInputError
 from sqlbuild.compiler.compile.models import (
     CompilerDiagnostic,
     CompileSqlReference,
@@ -24,7 +24,11 @@
     SqlReferenceOrigin,
     SqlReferenceScan,
 )
-from sqlbuild.compiler.compile.types import DiagnosticPhase, DiagnosticSeverity
+from sqlbuild.compiler.compile.types import (
+    DiagnosticPhase,
+    DiagnosticSeverity,
+    SqlReferenceScanFailure,
+)
 from sqlbuild.compiler.frontier.main.native_stage_enabled import native_stage_enabled
 from sqlbuild.compiler.frontier.types import NativeStage
 from sqlbuild.compiler.references.types import SqlReferenceKind
@@ -66,18 +70,34 @@ def extract_sql_references(
 ) -> tuple[CompileSqlReference, ...]:
     """Return logical SQL refs outside comments and quotes, reporting rejected calls as P012."""
 
-    scan: SqlReferenceScan = scan_sql_reference_calls(sql=sql, syntax=syntax)
+    scan: SqlReferenceScan = scan_sql_reference_calls(sql=sql, syntax=syntax, origin=origin)
     report_invalid_reference_calls(invalid_calls=scan.invalid_calls, origin=origin, syntax=syntax)
     return scan.references
 
 
-def scan_sql_reference_calls(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan:
-    """Return valid references and rejected reference calls without reporting either."""
+def scan_sql_reference_calls(
+    *, sql: str, syntax: SqlLexicalSyntax, origin: SqlReferenceOrigin | None = None
+) -> SqlReferenceScan:
+    """Return valid references and rejected calls; a scan error is located in `origin`."""
+
+    outcome: SqlReferenceScan | SqlReferenceScanFailure = _reference_scan_outcome(
+        sql=sql, syntax=syntax
+    )
+    if isinstance(outcome, SqlReferenceScan):
+        return outcome
+    raise _located_scan_error(message=outcome[0], origin=origin, syntax=syntax)
+
 
+def _reference_scan_outcome(
+    *, sql: str, syntax: SqlLexicalSyntax
+) -> SqlReferenceScan | SqlReferenceScanFailure:
     if native_stage_enabled(NativeStage.REFERENCE_EXTRACTION):
-        return _scan_sql_references_natively(sql=sql, syntax=syntax)
+        native: SqlReferenceScan | SqlReferenceScanFailure | None = extract_native_sql_references(
+            sql=sql, syntax=syntax
+        )
+        return native if native is not None else python_reference_scan(sql=sql, syntax=syntax)
     if syntax.reads_differently_from_generic(sql):
-        return _scan_sql_references_with_python(sql=sql, syntax=syntax)
+        return python_reference_scan(sql=sql, syntax=syntax)
     native_references: list[tuple[str, str, str | None, int | None]] | None = (
         _native.extract_static_sql_references(sql)
     )
@@ -93,18 +113,26 @@ def scan_sql_reference_calls(*, sql: str, syntax: SqlLexicalSyntax) -> SqlRefere
                 for kind, name, package, call_argument_count in native_references
             )
         )
-    return _scan_sql_references_with_python(sql=sql, syntax=syntax)
+    return python_reference_scan(sql=sql, syntax=syntax)
 
 
-def _scan_sql_references_natively(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan:
-    """Return the native scan or raise Python's error; Python scans the text native defers."""
+def _located_scan_error(
+    *, message: str, origin: SqlReferenceOrigin | None, syntax: SqlLexicalSyntax
+) -> CompileInputError:
+    """Locate a scan error at the authored quote, comment or call that raises the same error."""
 
-    native: SqlReferenceScan | str | None = extract_native_sql_references(sql=sql, syntax=syntax)
-    if native is None:
-        return _scan_sql_references_with_python(sql=sql, syntax=syntax)
-    if isinstance(native, str):
-        raise SqlReferenceExtractionError(native)
-    return native
+    if origin is None:
+        return CompileInputError(message)
+    authored: SqlReferenceScan | SqlReferenceScanFailure = _reference_scan_outcome(
+        sql=origin.contents, syntax=syntax
+    )
+    path: str = origin.relative_path.as_posix()
+    if isinstance(authored, SqlReferenceScan) or authored[0] != message:
+        return CompileInputError(f"{path}: {message}")
+    start: int = authored[1]
+    line: int = origin.contents.count("\n", 0, start) + 1
+    column: int = start - (origin.contents.rfind("\n", 0, start) + 1) + 1
+    return CompileInputError(f"{path}:{line}:{column}: {message}", bridge_independent=True)
 
 
 def report_invalid_reference_calls(
@@ -176,13 +204,10 @@ def _authored_invalid_call_starts(
 ) -> dict[str, list[int]]:
     if origin is None:
         return {}
-    try:
-        authored: SqlReferenceScan = (
-            _scan_sql_references_natively(sql=origin.contents, syntax=syntax)
-            if native_stage_enabled(NativeStage.REFERENCE_EXTRACTION)
-            else _scan_sql_references_with_python(sql=origin.contents, syntax=syntax)
-        )
-    except CompileInputError:
+    authored: SqlReferenceScan | SqlReferenceScanFailure = _reference_scan_outcome(
+        sql=origin.contents, syntax=syntax
+    )
+    if not isinstance(authored, SqlReferenceScan):
         return {}
     starts: dict[str, list[int]] = {}
     for invalid_call in authored.invalid_calls:
@@ -203,8 +228,10 @@ def reference_call_location(*, path: Path, text: str, start: int, call: str) ->
     )
 
 
-def _scan_sql_references_with_python(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan:
-    """Return logical SQL refs through the authoritative general scanner."""
+def python_reference_scan(
+    *, sql: str, syntax: SqlLexicalSyntax
+) -> SqlReferenceScan | SqlReferenceScanFailure:
+    """Scan with the Python scanner; an error carries the start of its quote, comment or call."""
 
     references: list[CompileSqlReference] = []
     invalid_calls: list[InvalidSqlReferenceCall] = []
@@ -214,16 +241,22 @@ def _scan_sql_references_with_python(*, sql: str, syntax: SqlLexicalSyntax) -> S
         index = _next_reference_scan_position(sql=sql, start=index)
         if index >= length:
             break
-        non_code_end: int | None = dialect_non_code_end(
-            sql=sql, start=index, syntax=syntax, context=_CONTEXT
-        )
+        try:
+            non_code_end: int | None = dialect_non_code_end(
+                sql=sql, start=index, syntax=syntax, context=_CONTEXT
+            )
+        except CompileInputError as error:
+            return (error.message, index)
         if non_code_end is not None:
             index = non_code_end
             continue
 
-        parsed: tuple[CompileSqlReference | InvalidSqlReferenceCall, int] | None = (
-            _parse_reference_at(sql=sql, start=index, syntax=syntax)
-        )
+        try:
+            parsed: tuple[CompileSqlReference | InvalidSqlReferenceCall, int] | None = (
+                _parse_reference_at(sql=sql, start=index, syntax=syntax)
+            )
+        except CompileInputError as error:
+            return (error.message, index)
         if parsed is None:
             index += 1
             continue
diff --git a/src/sqlbuild/compiler/compile/exceptions.py b/src/sqlbuild/compiler/compile/exceptions.py
index 3fe13453af..9d7946a898 100644
--- a/src/sqlbuild/compiler/compile/exceptions.py
+++ b/src/sqlbuild/compiler/compile/exceptions.py
@@ -67,9 +67,5 @@ class CompactAnalysisInputError(CompileInputError):
     """Raised when compact SQL-analysis batch inputs are inconsistent."""
 
 
-class SqlReferenceExtractionError(CompileInputError):
-    """Raised when SQL text cannot be scanned for its reference calls."""
-
-
 class AnalysisCacheEntryError(ValueError):
     """Raised when a persisted model analysis cache entry is invalid."""
diff --git a/src/sqlbuild/compiler/compile/types.py b/src/sqlbuild/compiler/compile/types.py
index b8e69b9a86..cb9e238230 100644
--- a/src/sqlbuild/compiler/compile/types.py
+++ b/src/sqlbuild/compiler/compile/types.py
@@ -123,3 +123,6 @@ class SqlTestMode(StrEnum):
     MACRO = "macro"
     UDF = "udf"
     TABLE_FN = "table_fn"
+
+
+type SqlReferenceScanFailure = tuple[str, int]
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
index bc3989c839..9084c6c61e 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
@@ -908,6 +908,16 @@ class NativeDeclarationErrorTestCase:
     expected_macro_calls: int
 
 
+@dataclass(frozen=True)
+class ReferenceScanErrorTestCase:
+    """A model whose reference scan error every engine must locate and type the same way."""
+
+    description: str
+    mart_body: str
+    expected_message: str
+    expected_error_types: tuple[str, ...]
+
+
 @dataclass(frozen=True)
 class DeclarationErrorLifecycleTestCase:
     """A failing project whose invocation lifecycle every engine must report the same way."""
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py
new file mode 100644
index 0000000000..181623e368
--- /dev/null
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py
@@ -0,0 +1,97 @@
+"""Every engine locates reference scan errors identically and reports them as CompileInputError."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+import pytest
+
+from scripts.compiler_differential.constants import FAILURE_BASE_FILES, FAILURE_MART_PATH
+from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
+    ReferenceScanErrorTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
+    CompileReuseRun,
+    lifecycle_error_type,
+    report_without_engine,
+    run_reuse_compile,
+    stderr_without_durations,
+    write_project_file,
+)
+
+_ENGINES: tuple[str, str, str] = ("python", "native", "native-preview")
+_MART_HEADER: str = 'MODEL (\n  description "Order totals per customer",\n);\n\n'
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        ReferenceScanErrorTestCase(
+            description="unclosed_reference_call",
+            mart_body='SELECT customer_id, 1 AS total_amount\nFROM __ref("stg_orders"\n',
+            expected_message=(
+                f"{FAILURE_MART_PATH}:6:6: SQL reference contains an unclosed parenthesis"
+            ),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="unclosed_quote_after_a_rejected_call",
+            mart_body=(
+                "SELECT customer_id, 1 AS total_amount\nFROM __ref(stg_orders)\n"
+                "WHERE 'é' <> 'open\n"
+            ),
+            expected_message=(
+                f"{FAILURE_MART_PATH}:7:14: SQL reference contains an unclosed quoted string"
+            ),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="empty_table_function_argument",
+            mart_body=(
+                "SELECT customer_id, 1 AS total_amount\n"
+                'FROM __ref("stg_orders") JOIN __table_fn("orders_for")(1, , 2) USING (customer_id)\n'
+            ),
+            expected_message=(
+                f"{FAILURE_MART_PATH}:6:31: SQL reference contains an empty argument"
+            ),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_reference_scan_error_when_compiling_with_each_engine_then_located_and_typed_alike(
+    test_case: ReferenceScanErrorTestCase, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    project_dir: Path = tmp_path / "orders"
+    runs: list[CompileReuseRun] = []
+    error_types: list[str] = []
+    for engine in _ENGINES:
+        shutil.rmtree(project_dir, ignore_errors=True)
+        for relative_path, contents in {
+            **FAILURE_BASE_FILES,
+            FAILURE_MART_PATH: _MART_HEADER + test_case.mart_body,
+        }.items():
+            write_project_file(project_dir, relative_path, contents)
+        runs.append(
+            run_reuse_compile(project_dir=project_dir, global_args=("--compiler-engine", engine))
+        )
+        error_types.append(
+            lifecycle_error_type(project_dir=project_dir, engine=engine, monkeypatch=monkeypatch)
+        )
+
+    assert (
+        tuple(run.returncode for run in runs),
+        {report_without_engine(run) for run in runs} == {report_without_engine(runs[0])},
+        {stderr_without_durations(stderr=run.stderr) for run in runs}
+        == {stderr_without_durations(stderr=runs[0].stderr)},
+        test_case.expected_message in runs[0].report,
+        tuple(error_types),
+    ) == ((1, 1, 1), True, True, True, test_case.expected_error_types), (
+        runs[0].report,
+        runs[2].report,
+    )
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
index aefa85830c..5c6c735fd4 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
@@ -44,11 +44,13 @@ class ReferenceEngineTestCase:
 
 @dataclass(frozen=True)
 class NativeReferenceErrorTestCase:
-    """Reference SQL whose native error must stand without a Python re-scan."""
+    """Reference SQL whose native error must stand, located, without a Python re-scan."""
 
     description: str
     sql: str
+    contents: str
     expected_message: str
+    expected_bridge_independent: bool
 
 
 @dataclass(frozen=True)
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
index 7c94059efc..4b8d376308 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
@@ -20,8 +20,8 @@
 from sqlbuild.compiler.compile._helpers.diagnostics.collector import collect_compile_diagnostics
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
 from sqlbuild.compiler.compile._helpers.refs.references import (
-    _scan_sql_references_with_python,
     extract_sql_references,
+    python_reference_scan,
 )
 from sqlbuild.compiler.compile.exceptions import CompileInputError
 from sqlbuild.compiler.compile.models import (
@@ -30,7 +30,7 @@
     SqlReferenceOrigin,
     SqlReferenceScan,
 )
-from sqlbuild.compiler.compile.types import CompiledResourceType
+from sqlbuild.compiler.compile.types import CompiledResourceType, SqlReferenceScanFailure
 from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
 from sqlbuild.compiler.frontier.types import CompilerEngine
 from sqlbuild.compiler.references.types import SqlReferenceKind
@@ -200,22 +200,23 @@ def _generated_reference_call(*, rng: random.Random) -> str:
     )
 
 
-def python_outcome(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan | str:
-    """Return the oracle's references and rejected calls, or its error message."""
+def python_outcome(
+    *, sql: str, syntax: SqlLexicalSyntax
+) -> SqlReferenceScan | SqlReferenceScanFailure:
+    """Return the oracle's references and rejected calls, or its error and where it points."""
 
-    try:
-        return _scan_sql_references_with_python(sql=sql, syntax=syntax)
-    except CompileInputError as error:
-        return str(error)
+    return python_reference_scan(sql=sql, syntax=syntax)
 
 
 def reference_parity(*, sqls: list[str], syntax: SqlLexicalSyntax) -> ReferenceParity:
     """Compare native extraction, rejected calls and errors with Python wherever native scans."""
 
-    native: list[SqlReferenceScan | str | None] = [
+    native: list[SqlReferenceScan | SqlReferenceScanFailure | None] = [
         extract_native_sql_references(sql=sql, syntax=syntax) for sql in sqls
     ]
-    python: list[SqlReferenceScan | str] = [python_outcome(sql=sql, syntax=syntax) for sql in sqls]
+    python: list[SqlReferenceScan | SqlReferenceScanFailure] = [
+        python_outcome(sql=sql, syntax=syntax) for sql in sqls
+    ]
     scanned: list[bool] = [outcome is not None for outcome in native]
     extracted: list[SqlReferenceScan] = cast(
         list[SqlReferenceScan],
@@ -228,7 +229,7 @@ def reference_parity(*, sqls: list[str], syntax: SqlLexicalSyntax) -> ReferenceP
             actual=list(compress(native, scanned)),
         ),
         extracted=sum(not scan.invalid_calls for scan in extracted),
-        failed=sum(isinstance(outcome, str) for outcome in native),
+        failed=sum(isinstance(outcome, tuple) for outcome in native),
         deferred=native.count(None),
         rejected=sum(len(scan.invalid_calls) for scan in extracted),
         table_functions=sum(
@@ -301,7 +302,7 @@ def located_diagnostic_count(*, outcomes: list[object]) -> int:
     return sum(diagnostic.location is not None for diagnostic in chain.from_iterable(diagnostics))
 
 
-def python_scan_not_expected(**_: object) -> SqlReferenceScan:
+def python_scan_not_expected(**_: object) -> SqlReferenceScan | SqlReferenceScanFailure:
     """A replacement for the Python scanner that fails the test when it is called."""
 
     raise AssertionError("the Python reference scanner ran")
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
index 1e7268db92..feac088e67 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
@@ -6,6 +6,8 @@
 import time
 from collections.abc import Callable
 from itertools import product
+from operator import itemgetter
+from pathlib import Path
 from typing import Any
 
 import pytest
@@ -13,8 +15,9 @@
 from sqlbuild.compiler.compile._helpers.refs import references
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
 from sqlbuild.compiler.compile._helpers.refs.references import extract_sql_references
-from sqlbuild.compiler.compile.exceptions import CompileInputError, SqlReferenceExtractionError
-from sqlbuild.compiler.compile.models import SqlReferenceScan
+from sqlbuild.compiler.compile.exceptions import CompileInputError
+from sqlbuild.compiler.compile.models import SqlReferenceOrigin, SqlReferenceScan
+from sqlbuild.compiler.compile.types import SqlReferenceScanFailure
 from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
 from sqlbuild.compiler.frontier.types import CompilerEngine
 from sqlbuild.compiler.sql_analysis.models import SqlLexicalSyntax
@@ -72,7 +75,7 @@
 _NESTED_CALLS: str = '__table_fn("orders_for")(' * 1_000 + "1" + ")" * 1_000
 _OUTCOME_SUMMARIES: dict[type, Callable[[Any], int | str]] = {
     SqlReferenceScan: lambda scan: len(scan.references) + len(scan.invalid_calls),
-    str: str,
+    tuple: itemgetter(0),
     type(None): repr,
 }
 _MANY_REFERENCES: str = "SELECT 1 FROM " + " JOIN ".join(
@@ -271,11 +274,13 @@ def test_given_engine_when_extracting_references_then_only_preview_runs_native_s
 ) -> None:
     monkeypatch.setenv(COMPILER_ENGINE_ENV_VAR, test_case.engine.value)
     native_calls: list[str] = []
-    native_extract: Callable[..., SqlReferenceScan | str | None] = (
+    native_extract: Callable[..., SqlReferenceScan | SqlReferenceScanFailure | None] = (
         references.extract_native_sql_references
     )
 
-    def counting_extract(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan | str | None:
+    def counting_extract(
+        *, sql: str, syntax: SqlLexicalSyntax
+    ) -> SqlReferenceScan | SqlReferenceScanFailure | None:
         native_calls.append(sql)
         return native_extract(sql=sql, syntax=syntax)
 
@@ -302,36 +307,69 @@ def counting_extract(*, sql: str, syntax: SqlLexicalSyntax) -> SqlReferenceScan
     "test_case",
     [
         NativeReferenceErrorTestCase(
-            description="unclosed_reference_call",
-            sql='SELECT * FROM __ref("orders"',
-            expected_message="SQL reference contains an unclosed parenthesis",
+            description="unclosed_reference_call_points_at_the_call",
+            sql='SELECT *\nFROM __ref("orders"',
+            contents='MODEL (description "Orders");\n\nSELECT *\nFROM __ref("orders"',
+            expected_message=(
+                "models/orders.sql:4:6: SQL reference contains an unclosed parenthesis"
+            ),
+            expected_bridge_independent=True,
         ),
         NativeReferenceErrorTestCase(
             description="empty_table_function_argument_after_a_rejected_call",
-            sql='SELECT * FROM __ref(orders) JOIN __table_fn("orders_for")(1,,2)',
-            expected_message="SQL reference contains an empty argument",
+            sql='SELECT * FROM __ref(orders)\nJOIN __table_fn("orders_for")(1,,2)',
+            contents='SELECT * FROM __ref(orders)\nJOIN __table_fn("orders_for")(1,,2)',
+            expected_message="models/orders.sql:2:6: SQL reference contains an empty argument",
+            expected_bridge_independent=True,
+        ),
+        NativeReferenceErrorTestCase(
+            description="unclosed_quote_points_at_the_quote_in_code_points",
+            sql="SELECT 'é', __ref(\"orders\") WHERE note = 'open",
+            contents="-- é\nSELECT 'é', __ref(\"orders\") WHERE note = 'open",
+            expected_message=(
+                "models/orders.sql:2:42: SQL reference contains an unclosed quoted string"
+            ),
+            expected_bridge_independent=True,
         ),
         NativeReferenceErrorTestCase(
-            description="unclosed_quote_after_references",
-            sql='SELECT * FROM __ref("orders") WHERE note = \'open',
-            expected_message="SQL reference contains an unclosed quoted string",
+            description="error_only_in_expanded_sql_names_the_file",
+            sql='SELECT * FROM __ref("orders") /* expanded',
+            contents='SELECT * FROM __ref("orders") @expand()',
+            expected_message="models/orders.sql: SQL reference contains an unclosed block comment",
+            expected_bridge_independent=False,
         ),
     ],
     ids=lambda case: case.description,
 )
-def test_given_native_reference_error_when_extracting_then_raised_without_python_rescan(
+def test_given_native_reference_error_when_extracting_then_raised_located_without_python_rescan(
     test_case: NativeReferenceErrorTestCase, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     monkeypatch.setenv(COMPILER_ENGINE_ENV_VAR, CompilerEngine.NATIVE_PREVIEW.value)
-    monkeypatch.setattr(references, "_scan_sql_references_with_python", python_scan_not_expected)
-
-    with pytest.raises(SqlReferenceExtractionError) as raised:
-        _ = extract_sql_references(sql=test_case.sql, syntax=_GENERIC_SYNTAX)
+    monkeypatch.setattr(references, "python_reference_scan", python_scan_not_expected)
+
+    with pytest.raises(CompileInputError) as raised:
+        _ = extract_sql_references(
+            sql=test_case.sql,
+            syntax=_GENERIC_SYNTAX,
+            origin=SqlReferenceOrigin(
+                file_path=Path("/project/models/orders.sql"),
+                relative_path=Path("models/orders.sql"),
+                contents=test_case.contents,
+            ),
+        )
 
-    assert (str(raised.value), raised.value.code, raised.value.help) == (
+    assert (
+        type(raised.value),
+        str(raised.value),
+        raised.value.code,
+        raised.value.help,
+        raised.value.bridge_independent,
+    ) == (
+        CompileInputError,
         test_case.expected_message,
         "P001",
         None,
+        test_case.expected_bridge_independent,
     )
 
 
@@ -410,7 +448,7 @@ def test_given_worst_case_reference_sql_when_extracting_natively_then_finishes_q
     test_case: ReferenceScanBoundTestCase,
 ) -> None:
     started: float = time.thread_time()
-    outcome: SqlReferenceScan | str | None = extract_native_sql_references(
+    outcome: SqlReferenceScan | SqlReferenceScanFailure | None = extract_native_sql_references(
         sql=test_case.sql, syntax=_GENERIC_SYNTAX
     )
     elapsed: float = time.thread_time() - started

From a1361e0dbfedd45105622080ac09b09525bb6f31 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Thu, 8 Oct 2026 20:43:46 +0100
Subject: [PATCH 3/3] fix(native): locate reference scan errors through the
 expansion source map

---
 .../compile/_helpers/attachment/core.py       |  27 +++-
 .../compile/_helpers/refs/references.py       |  87 ++++++++--
 src/sqlbuild/compiler/compile/models.py       |   9 ++
 .../discovery/main/_model_query_start.py      |  17 ++
 .../cli/commands/main/compile/_test_types.py  |   5 +-
 .../compile/test_native_reference_errors.py   | 151 ++++++++++++++++--
 .../compile/_helpers/refs/_test_types.py      |   5 +-
 .../compiler/compile/_helpers/refs/helpers.py |  21 +++
 .../_helpers/refs/test_native_references.py   |  73 +++++++--
 9 files changed, 357 insertions(+), 38 deletions(-)
 create mode 100644 src/sqlbuild/compiler/discovery/main/_model_query_start.py

diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/core.py b/src/sqlbuild/compiler/compile/_helpers/attachment/core.py
index 0e900cedab..183c189b3c 100644
--- a/src/sqlbuild/compiler/compile/_helpers/attachment/core.py
+++ b/src/sqlbuild/compiler/compile/_helpers/attachment/core.py
@@ -5,6 +5,7 @@
 import inspect
 import re
 from dataclasses import dataclass, field, fields, replace
+from functools import partial
 from inspect import Parameter, Signature
 from pathlib import Path
 from typing import Any, cast
@@ -122,9 +123,11 @@
     SqlAnalysisOptOutRequest,
     SqlReferenceOrigin,
     SqlReferenceScan,
+    SqlReferenceSourceMap,
 )
 from sqlbuild.compiler.compile.types import CompiledResourceType, SqlReferenceExtractor
 from sqlbuild.compiler.discovery.constants import PROJECT_CONFIG_FILENAME
+from sqlbuild.compiler.discovery.main._model_query_start import get_model_query_start
 from sqlbuild.compiler.discovery.models import (
     ConstantDeclaration,
     DiscoveredHookFunction,
@@ -780,10 +783,23 @@ def _build_model_input(
         sql_validation_placeholders=sql_validation_placeholders,
         model_schema_columns=model_schema_columns,
         argument_references=macro_expansion.argument_references,
+        reference_source_map=(
+            SqlReferenceSourceMap(
+                body_start=partial(
+                    get_model_query_start,
+                    contents=model_file.contents,
+                    query_sql=model_file.query_sql,
+                ),
+                passes=(declaration_expansion.spans, macro_expansion.spans),
+            )
+            if var_substituted_sql == model_file.query_sql
+            and expanded_query_sql == macro_expansion.sql
+            else None
+        ),
     )
     hook_expansion: HookExpansionResult = expand_model_hook_macros_result(
         values=effective_config.values,
-        model_file=_model_reference_origin(model_file),
+        model_file=_model_reference_origin(model_file=model_file),
         effective_vars=effective_vars,
         context_values=build_model_context_values(
             values=effective_config.values,
@@ -951,13 +967,16 @@ def _hook_expanded_config(
     )
 
 
-def _model_reference_origin(model_file: DiscoveredSqlModelFile) -> SqlReferenceOrigin:
+def _model_reference_origin(
+    *, model_file: DiscoveredSqlModelFile, source_map: SqlReferenceSourceMap | None = None
+) -> SqlReferenceOrigin:
     return SqlReferenceOrigin(
         file_path=model_file.file_path,
         relative_path=model_file.relative_path,
         contents=model_file.contents,
         resource_type=CompiledResourceType.MODEL,
         resource_name=model_file.file_path.stem,
+        source_map=source_map,
     )
 
 
@@ -970,6 +989,7 @@ def _validate_model_input(
     sql_validation_placeholders: dict[str, str] | None,
     model_schema_columns: tuple[SchemaColumn, ...] | None,
     argument_references: tuple[CompileSqlReference, ...],
+    reference_source_map: SqlReferenceSourceMap | None,
 ) -> tuple[bool, tuple[CompileSqlReference, ...], SourceLocation | None]:
     model_name: str = model_file.file_path.stem
     sql_validation_enabled: bool = _model_sql_validation_gate(
@@ -1006,7 +1026,8 @@ def _validate_model_input(
             placeholders=sql_validation_placeholders,
         )
     reference_scan: SqlReferenceScan = context.extract_references(
-        sql=expanded_query_sql, origin=_model_reference_origin(model_file)
+        sql=expanded_query_sql,
+        origin=_model_reference_origin(model_file=model_file, source_map=reference_source_map),
     )
     references: tuple[CompileSqlReference, ...] = merge_call_site_references(
         references=reference_scan.references,
diff --git a/src/sqlbuild/compiler/compile/_helpers/refs/references.py b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
index 3cfeac5742..5fd0299871 100644
--- a/src/sqlbuild/compiler/compile/_helpers/refs/references.py
+++ b/src/sqlbuild/compiler/compile/_helpers/refs/references.py
@@ -8,6 +8,7 @@
 import sqlbuild._native as _native
 from sqlbuild.compiler.compile._helpers.diagnostics.collector import report_compile_diagnostic
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
+from sqlbuild.compiler.compile._helpers.render.spans import map_through_passes
 from sqlbuild.compiler.compile.constants import (
     REFERENCE_CALL_SYNTAX_CODE,
     SQL_ARGUMENT_SEPARATOR_TOKEN,
@@ -20,9 +21,12 @@
 from sqlbuild.compiler.compile.models import (
     CompilerDiagnostic,
     CompileSqlReference,
+    ExpansionSpan,
     InvalidSqlReferenceCall,
+    MappedOffset,
     SqlReferenceOrigin,
     SqlReferenceScan,
+    SqlReferenceSourceMap,
 )
 from sqlbuild.compiler.compile.types import (
     DiagnosticPhase,
@@ -85,7 +89,7 @@ def scan_sql_reference_calls(
     )
     if isinstance(outcome, SqlReferenceScan):
         return outcome
-    raise _located_scan_error(message=outcome[0], origin=origin, syntax=syntax)
+    raise _located_scan_error(failure=outcome, sql=sql, origin=origin, syntax=syntax)
 
 
 def _reference_scan_outcome(
@@ -117,24 +121,89 @@ def _reference_scan_outcome(
 
 
 def _located_scan_error(
-    *, message: str, origin: SqlReferenceOrigin | None, syntax: SqlLexicalSyntax
+    *,
+    failure: SqlReferenceScanFailure,
+    sql: str,
+    origin: SqlReferenceOrigin | None,
+    syntax: SqlLexicalSyntax,
 ) -> CompileInputError:
-    """Locate a scan error at the authored quote, comment or call that raises the same error."""
+    """Locate a scan error in authored text; unmapped or expansion-shaped faults stay unlocated."""
 
+    message, start = failure
     if origin is None:
         return CompileInputError(message)
-    authored: SqlReferenceScan | SqlReferenceScanFailure = _reference_scan_outcome(
-        sql=origin.contents, syntax=syntax
+    authored_offset: int | None = _authored_fault_offset(
+        failure=failure, sql=sql, origin=origin, syntax=syntax
     )
     path: str = origin.relative_path.as_posix()
-    if isinstance(authored, SqlReferenceScan) or authored[0] != message:
+    if authored_offset is None:
         return CompileInputError(f"{path}: {message}")
-    start: int = authored[1]
-    line: int = origin.contents.count("\n", 0, start) + 1
-    column: int = start - (origin.contents.rfind("\n", 0, start) + 1) + 1
+    contents: str = origin.contents
+    line: int = contents.count("\n", 0, authored_offset) + 1
+    column: int = authored_offset - (contents.rfind("\n", 0, authored_offset) + 1) + 1
     return CompileInputError(f"{path}:{line}:{column}: {message}", bridge_independent=True)
 
 
+def _authored_fault_offset(
+    *,
+    failure: SqlReferenceScanFailure,
+    sql: str,
+    origin: SqlReferenceOrigin,
+    syntax: SqlLexicalSyntax,
+) -> int | None:
+    """Map the fault to authored text unless expansion output produced or shaped the failure."""
+
+    source_map: SqlReferenceSourceMap | None = origin.source_map
+    start: int = failure[1]
+    body_start: int | None = source_map.body_start() if source_map is not None else None
+    if source_map is None or body_start is None or start >= len(sql):
+        return None
+    mapped: MappedOffset = map_through_passes(offset=start, passes=source_map.passes)
+    authored_offset: int = body_start + mapped.offset
+    if (
+        mapped.generated
+        or authored_offset >= len(origin.contents)
+        or origin.contents[authored_offset] != sql[start]
+    ):
+        return None
+    blanked: list[str] = list(sql)
+    for output_start, output_end in _expansion_output_ranges(source_map.passes):
+        blanked[output_start:output_end] = " " * (output_end - output_start)
+    if _reference_scan_outcome(sql="".join(blanked), syntax=syntax) != failure:
+        return None
+    return authored_offset
+
+
+def _expansion_output_ranges(
+    passes: tuple[tuple[ExpansionSpan, ...], ...],
+) -> list[tuple[int, int]]:
+    """Return every expansion's output range in the coordinates of the final pass's output."""
+
+    ranges: list[tuple[int, int]] = []
+    for index, spans in enumerate(passes):
+        for span in spans:
+            output_start: int = span.output_start
+            output_end: int = span.output_end
+            for later_spans in passes[index + 1 :]:
+                output_start = _later_pass_offset(offset=output_start, spans=later_spans, end=False)
+                output_end = _later_pass_offset(offset=output_end, spans=later_spans, end=True)
+            ranges.append((output_start, output_end))
+    return ranges
+
+
+def _later_pass_offset(*, offset: int, spans: tuple[ExpansionSpan, ...], end: bool) -> int:
+    """Carry a range boundary through one later pass, widening it over a span that covers it."""
+
+    shift: int = 0
+    for span in spans:
+        if offset < span.source_start or (end and offset == span.source_start):
+            break
+        if offset < span.source_end or (end and offset == span.source_end):
+            return span.output_end if end else span.output_start
+        shift += (span.output_end - span.output_start) - (span.source_end - span.source_start)
+    return offset + shift
+
+
 def report_invalid_reference_calls(
     *,
     invalid_calls: tuple[InvalidSqlReferenceCall, ...],
diff --git a/src/sqlbuild/compiler/compile/models.py b/src/sqlbuild/compiler/compile/models.py
index e7f4fd938c..1b3e17de44 100644
--- a/src/sqlbuild/compiler/compile/models.py
+++ b/src/sqlbuild/compiler/compile/models.py
@@ -555,6 +555,14 @@ class SqlReferenceScan:
     invalid_calls: tuple[InvalidSqlReferenceCall, ...] = ()
 
 
+@dataclass(frozen=True)
+class SqlReferenceSourceMap:
+    """Where the authored body starts in its file and the expansion passes from it to the SQL."""
+
+    body_start: Callable[[], int | None]
+    passes: tuple[tuple[ExpansionSpan, ...], ...]
+
+
 @dataclass(frozen=True)
 class SqlReferenceOrigin:
     """The authored file that SQL handed to reference extraction came from."""
@@ -564,6 +572,7 @@ class SqlReferenceOrigin:
     contents: str
     resource_type: CompiledResourceType | None = None
     resource_name: str | None = None
+    source_map: SqlReferenceSourceMap | None = None
 
 
 @dataclass(frozen=True)
diff --git a/src/sqlbuild/compiler/discovery/main/_model_query_start.py b/src/sqlbuild/compiler/discovery/main/_model_query_start.py
new file mode 100644
index 0000000000..c0411a93f8
--- /dev/null
+++ b/src/sqlbuild/compiler/discovery/main/_model_query_start.py
@@ -0,0 +1,17 @@
+"""Locate a model's discovered query SQL within its authored file."""
+
+from __future__ import annotations
+
+from sqlbuild.compiler.discovery._helpers.sql.model_files import match_model_header
+from sqlbuild.compiler.discovery.models import ModelHeaderMatch
+
+
+def get_model_query_start(*, contents: str, query_sql: str) -> int | None:
+    """Return where discovery's stripped query SQL starts after the MODEL header, if it does."""
+
+    header_match: ModelHeaderMatch | None = match_model_header(contents)
+    if header_match is None:
+        return None
+    sql: str = header_match.sql
+    start: int = header_match.sql_start + len(sql) - len(sql.lstrip())
+    return start if contents.startswith(query_sql, start) else None
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
index 9084c6c61e..378f5b79ac 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py
@@ -910,11 +910,12 @@ class NativeDeclarationErrorTestCase:
 
 @dataclass(frozen=True)
 class ReferenceScanErrorTestCase:
-    """A model whose reference scan error every engine must locate and type the same way."""
+    """A project whose reference scan error every engine must locate and type the same way."""
 
     description: str
-    mart_body: str
+    project_files: dict[str, str]
     expected_message: str
+    expected_macro_calls: tuple[int, int, int]
     expected_error_types: tuple[str, ...]
 
 
diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py
index 181623e368..48346c0cf5 100644
--- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py
+++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_native_reference_errors.py
@@ -7,7 +7,11 @@
 
 import pytest
 
-from scripts.compiler_differential.constants import FAILURE_BASE_FILES, FAILURE_MART_PATH
+from scripts.compiler_differential.constants import (
+    FAILURE_BASE_CONFIG,
+    FAILURE_BASE_FILES,
+    FAILURE_MART_PATH,
+)
 from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import (
     ReferenceScanErrorTestCase,
 )
@@ -21,7 +25,20 @@
 )
 
 _ENGINES: tuple[str, str, str] = ("python", "native", "native-preview")
+_CALL_LOG: str = "macro_calls.log"
+_EMIT_MACRO: str = (
+    "from pathlib import Path\n\n\n"
+    "def emit(expression: str) -> str:\n"
+    '    """Return the expression, logging each execution."""\n\n'
+    f"    with (Path(__file__).resolve().parents[1] / {_CALL_LOG!r}).open(\n"
+    '        "a", encoding="utf-8"\n'
+    "    ) as log:\n"
+    '        _ = log.write("call\\n")\n'
+    "    return expression\n"
+)
 _MART_HEADER: str = 'MODEL (\n  description "Order totals per customer",\n);\n\n'
+_QUOTED_NOTE: str = "over 12\\\" boxes, by customer's orders"
+_NO_SQL_ANALYSIS_CONFIG: str = FAILURE_BASE_CONFIG + "\n[settings]\nsql_analysis = false\n"
 
 
 @pytest.mark.parametrize(
@@ -29,32 +46,132 @@
     [
         ReferenceScanErrorTestCase(
             description="unclosed_reference_call",
-            mart_body='SELECT customer_id, 1 AS total_amount\nFROM __ref("stg_orders"\n',
+            project_files={
+                FAILURE_MART_PATH: _MART_HEADER
+                + 'SELECT customer_id, 1 AS total_amount\nFROM __ref("stg_orders"\n'
+            },
             expected_message=(
                 f"{FAILURE_MART_PATH}:6:6: SQL reference contains an unclosed parenthesis"
             ),
+            expected_macro_calls=(0, 0, 0),
             expected_error_types=("CompileInputError",) * 3,
         ),
         ReferenceScanErrorTestCase(
             description="unclosed_quote_after_a_rejected_call",
-            mart_body=(
-                "SELECT customer_id, 1 AS total_amount\nFROM __ref(stg_orders)\n"
+            project_files={
+                FAILURE_MART_PATH: _MART_HEADER
+                + "SELECT customer_id, 1 AS total_amount\nFROM __ref(stg_orders)\n"
                 "WHERE 'é' <> 'open\n"
-            ),
+            },
             expected_message=(
                 f"{FAILURE_MART_PATH}:7:14: SQL reference contains an unclosed quoted string"
             ),
+            expected_macro_calls=(0, 0, 0),
             expected_error_types=("CompileInputError",) * 3,
         ),
         ReferenceScanErrorTestCase(
             description="empty_table_function_argument",
-            mart_body=(
-                "SELECT customer_id, 1 AS total_amount\n"
+            project_files={
+                FAILURE_MART_PATH: _MART_HEADER + "SELECT customer_id, 1 AS total_amount\n"
                 'FROM __ref("stg_orders") JOIN __table_fn("orders_for")(1, , 2) USING (customer_id)\n'
-            ),
+            },
             expected_message=(
                 f"{FAILURE_MART_PATH}:6:31: SQL reference contains an empty argument"
             ),
+            expected_macro_calls=(0, 0, 0),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="header_string_with_an_escaped_quote_and_apostrophe",
+            project_files={
+                FAILURE_MART_PATH: f'MODEL (\n  description "{_QUOTED_NOTE}",\n);\n\n'
+                'SELECT customer_id, 1 AS total_amount\nFROM __ref("stg_orders")\nWHERE "open = 1\n'
+            },
+            expected_message=(
+                f"{FAILURE_MART_PATH}:7:7: SQL reference contains an unclosed quoted string"
+            ),
+            expected_macro_calls=(0, 0, 0),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="macro_output_fault_before_an_authored_fault_is_unlocated",
+            project_files={
+                "macros/emit.py": _EMIT_MACRO,
+                FAILURE_MART_PATH: _MART_HEADER
+                + "SELECT customer_id, @emit('__ref(1') AS total_amount\n"
+                'FROM __ref("stg_orders"\n',
+            },
+            expected_message=(
+                f"{FAILURE_MART_PATH}: SQL reference contains an unclosed parenthesis"
+            ),
+            expected_macro_calls=(1, 1, 2),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="authored_fault_after_a_macro_is_located_without_a_python_rerun",
+            project_files={
+                "macros/emit.py": _EMIT_MACRO,
+                FAILURE_MART_PATH: _MART_HEADER + "SELECT customer_id, @emit('1') AS total_amount\n"
+                'FROM __ref("stg_orders"\n',
+            },
+            expected_message=(
+                f"{FAILURE_MART_PATH}:6:6: SQL reference contains an unclosed parenthesis"
+            ),
+            expected_macro_calls=(1, 1, 1),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="test_file_header_names_the_file",
+            project_files={
+                "tests/unit/test_customer_totals.sql": (
+                    'TEST (\n  name "totals_by_note",\n  parameters (note string),\n'
+                    f'  cases (boxes (note "{_QUOTED_NOTE}")),\n);\n\nWITH\n'
+                    "__ref__stg_orders AS (\n"
+                    "  SELECT 10 AS customer_id, CAST(5 AS DOUBLE) AS amount\n),\n"
+                    "__expected__customer_totals AS (\n"
+                    '  SELECT 10 AS customer_id, 5 AS total_amount, @param("note") AS note\n),\n'
+                    "__assert__totals AS (\n"
+                    '  SELECT * FROM __table_fn("orders_for")(1, , 2)\n)\n'
+                )
+            },
+            expected_message=(
+                "tests/unit/test_customer_totals.sql: SQL reference contains an empty argument"
+            ),
+            expected_macro_calls=(0, 0, 0),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="audit_file_header_names_the_file",
+            project_files={
+                "audits/generic/positive_total.sql": (
+                    "AUDIT (\n  -- totals over 12\" boxes, by customer's orders\n"
+                    "  severity warn\n);\n\n"
+                    'SELECT * FROM __ref("@model") WHERE note = \'open\n'
+                ),
+                FAILURE_MART_PATH: FAILURE_BASE_FILES[FAILURE_MART_PATH].replace(
+                    "  description", "  audits [positive_total],\n  description"
+                ),
+            },
+            expected_message=(
+                "audits/generic/positive_total.sql: SQL reference contains an unclosed quoted string"
+            ),
+            expected_macro_calls=(0, 0, 0),
+            expected_error_types=("CompileInputError",) * 3,
+        ),
+        ReferenceScanErrorTestCase(
+            description="function_file_header_names_the_file",
+            project_files={
+                "sqlbuild_project.toml": _NO_SQL_ANALYSIS_CONFIG,
+                "functions/sql/orders_for.sql": (
+                    f'FUNCTION (\n  description "{_QUOTED_NOTE}",\n'
+                    "  arguments (p_customer_id INTEGER),\n  returns INTEGER\n);\n\n"
+                    'SELECT COUNT(*) FROM __ref("stg_orders" WHERE customer_id = p_customer_id\n'
+                ),
+            },
+            expected_message=(
+                "functions/sql/orders_for.sql: SQL reference contains an unclosed parenthesis"
+            ),
+            expected_macro_calls=(0, 0, 0),
             expected_error_types=("CompileInputError",) * 3,
         ),
     ],
@@ -65,17 +182,20 @@ def test_given_reference_scan_error_when_compiling_with_each_engine_then_located
 ) -> None:
     project_dir: Path = tmp_path / "orders"
     runs: list[CompileReuseRun] = []
+    macro_calls: list[int] = []
     error_types: list[str] = []
     for engine in _ENGINES:
         shutil.rmtree(project_dir, ignore_errors=True)
         for relative_path, contents in {
             **FAILURE_BASE_FILES,
-            FAILURE_MART_PATH: _MART_HEADER + test_case.mart_body,
+            _CALL_LOG: "",
+            **test_case.project_files,
         }.items():
             write_project_file(project_dir, relative_path, contents)
         runs.append(
             run_reuse_compile(project_dir=project_dir, global_args=("--compiler-engine", engine))
         )
+        macro_calls.append(len((project_dir / _CALL_LOG).read_text(encoding="utf-8").splitlines()))
         error_types.append(
             lifecycle_error_type(project_dir=project_dir, engine=engine, monkeypatch=monkeypatch)
         )
@@ -86,11 +206,16 @@ def test_given_reference_scan_error_when_compiling_with_each_engine_then_located
         {stderr_without_durations(stderr=run.stderr) for run in runs}
         == {stderr_without_durations(stderr=runs[0].stderr)},
         test_case.expected_message in runs[0].report,
+        tuple(macro_calls),
         tuple(error_types),
-    ) == ((1, 1, 1), True, True, True, test_case.expected_error_types), (
-        runs[0].report,
-        runs[2].report,
-    )
+    ) == (
+        (1, 1, 1),
+        True,
+        True,
+        True,
+        test_case.expected_macro_calls,
+        test_case.expected_error_types,
+    ), (runs[0].report, runs[2].report)
 
 
 if __name__ == "__main__":
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
index 5c6c735fd4..177744b936 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/_test_types.py
@@ -2,6 +2,7 @@
 
 from dataclasses import dataclass
 
+from sqlbuild.compiler.compile.models import SqlReferenceSourceMap
 from sqlbuild.compiler.frontier.types import CompilerEngine
 
 
@@ -44,11 +45,12 @@ class ReferenceEngineTestCase:
 
 @dataclass(frozen=True)
 class NativeReferenceErrorTestCase:
-    """Reference SQL whose native error must stand, located, without a Python re-scan."""
+    """Expanded SQL whose native error is located through its source map without Python."""
 
     description: str
     sql: str
     contents: str
+    source_map: SqlReferenceSourceMap | None
     expected_message: str
     expected_bridge_independent: bool
 
@@ -62,6 +64,7 @@ class ReferenceDiagnosticParityTestCase:
     seed: int
     count: int
     expected_minimum_located: int
+    expected_minimum_located_errors: int
 
 
 @dataclass(frozen=True)
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
index 4b8d376308..9091cd6beb 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/helpers.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import random
+import re
 from dataclasses import dataclass
 from itertools import chain, compress
 from operator import attrgetter, itemgetter
@@ -27,8 +28,10 @@
 from sqlbuild.compiler.compile.models import (
     CompilerDiagnostic,
     CompileSqlReference,
+    ExpansionSpan,
     SqlReferenceOrigin,
     SqlReferenceScan,
+    SqlReferenceSourceMap,
 )
 from sqlbuild.compiler.compile.types import CompiledResourceType, SqlReferenceScanFailure
 from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
@@ -239,10 +242,12 @@ def reference_parity(*, sqls: list[str], syntax: SqlLexicalSyntax) -> ReferenceP
     )
 
 
+_LOCATED_ERROR_PATTERN: re.Pattern[str] = re.compile(r"[^:]+:\d+:\d+: ")
 _FILE_HEADERS: tuple[str, ...] = (
     'MODEL (description "Orders");\n\n',
     'MODEL (\n  description "Orders",\n);\n-- notes\n\n',
     'MODEL (description "Ordres é");\n/* commentaire */\n',
+    'MODEL (description "Orders over 12\\" boxes, by customer\'s totals");\n\n',
 )
 
 
@@ -273,6 +278,7 @@ def reported_reference_outcomes(
                 contents=_FILE_HEADERS[index % len(_FILE_HEADERS)] + body,
                 resource_type=CompiledResourceType.MODEL,
                 resource_name=Path(relative_path).stem,
+                source_map=source_map_at(body_start=len(_FILE_HEADERS[index % len(_FILE_HEADERS)])),
             ),
             sql=body,
             syntax=syntax,
@@ -293,6 +299,13 @@ def _reported_reference_outcome(
     return (references, tuple(collected.diagnostics), "", None)
 
 
+def located_error_count(*, outcomes: list[object]) -> int:
+    """Count the scan errors that carry an authored line and column."""
+
+    errors: list[str] = list(map(itemgetter(2), cast(list[tuple[object, object, str]], outcomes)))
+    return sum(bool(_LOCATED_ERROR_PATTERN.match(error)) for error in errors)
+
+
 def located_diagnostic_count(*, outcomes: list[object]) -> int:
     """Count the reported diagnostics that carry an authored location."""
 
@@ -306,3 +319,11 @@ def python_scan_not_expected(**_: object) -> SqlReferenceScan | SqlReferenceScan
     """A replacement for the Python scanner that fails the test when it is called."""
 
     raise AssertionError("the Python reference scanner ran")
+
+
+def source_map_at(
+    *, body_start: int, passes: tuple[tuple[ExpansionSpan, ...], ...] = ()
+) -> SqlReferenceSourceMap:
+    """Return a source map whose authored body starts at `body_start` in its file."""
+
+    return SqlReferenceSourceMap(body_start=lambda: body_start, passes=passes)
diff --git a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
index feac088e67..7f7068041e 100644
--- a/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
+++ b/tests/integration/src/sqlbuild/compiler/compile/_helpers/refs/test_native_references.py
@@ -16,7 +16,7 @@
 from sqlbuild.compiler.compile._helpers.refs.native import extract_native_sql_references
 from sqlbuild.compiler.compile._helpers.refs.references import extract_sql_references
 from sqlbuild.compiler.compile.exceptions import CompileInputError
-from sqlbuild.compiler.compile.models import SqlReferenceOrigin, SqlReferenceScan
+from sqlbuild.compiler.compile.models import ExpansionSpan, SqlReferenceOrigin, SqlReferenceScan
 from sqlbuild.compiler.compile.types import SqlReferenceScanFailure
 from sqlbuild.compiler.frontier.constants import COMPILER_ENGINE_ENV_VAR
 from sqlbuild.compiler.frontier.types import CompilerEngine
@@ -35,9 +35,11 @@
     generated_reference_files,
     generated_reference_sqls,
     located_diagnostic_count,
+    located_error_count,
     python_scan_not_expected,
     reference_parity,
     reported_reference_outcomes,
+    source_map_at,
 )
 from tests.integration.src.sqlbuild.compiler.helpers import mismatches
 
@@ -307,18 +309,18 @@ def counting_extract(
     "test_case",
     [
         NativeReferenceErrorTestCase(
-            description="unclosed_reference_call_points_at_the_call",
+            description="unclosed_reference_call_points_at_the_call_after_the_header",
             sql='SELECT *\nFROM __ref("orders"',
             contents='MODEL (description "Orders");\n\nSELECT *\nFROM __ref("orders"',
-            expected_message=(
-                "models/orders.sql:4:6: SQL reference contains an unclosed parenthesis"
-            ),
+            source_map=source_map_at(body_start=31),
+            expected_message="models/orders.sql:4:6: SQL reference contains an unclosed parenthesis",
             expected_bridge_independent=True,
         ),
         NativeReferenceErrorTestCase(
             description="empty_table_function_argument_after_a_rejected_call",
             sql='SELECT * FROM __ref(orders)\nJOIN __table_fn("orders_for")(1,,2)',
             contents='SELECT * FROM __ref(orders)\nJOIN __table_fn("orders_for")(1,,2)',
+            source_map=source_map_at(body_start=0),
             expected_message="models/orders.sql:2:6: SQL reference contains an empty argument",
             expected_bridge_independent=True,
         ),
@@ -326,18 +328,66 @@ def counting_extract(
             description="unclosed_quote_points_at_the_quote_in_code_points",
             sql="SELECT 'é', __ref(\"orders\") WHERE note = 'open",
             contents="-- é\nSELECT 'é', __ref(\"orders\") WHERE note = 'open",
-            expected_message=(
-                "models/orders.sql:2:42: SQL reference contains an unclosed quoted string"
+            source_map=source_map_at(body_start=5),
+            expected_message="models/orders.sql:2:42: SQL reference contains an unclosed quoted string",
+            expected_bridge_independent=True,
+        ),
+        NativeReferenceErrorTestCase(
+            description="fault_after_neutral_expansions_in_two_passes_maps_through_both",
+            sql="SELECT 'open', 1 + 2\nFROM __ref(\"orders\"",
+            contents='SELECT @enum("s").OPEN, @m()\nFROM __ref("orders"',
+            source_map=source_map_at(
+                body_start=0,
+                passes=(
+                    (ExpansionSpan(source_start=7, source_end=22, output_start=7, output_end=13),),
+                    (
+                        ExpansionSpan(
+                            source_start=15, source_end=19, output_start=15, output_end=20
+                        ),
+                    ),
+                ),
             ),
+            expected_message="models/orders.sql:2:6: SQL reference contains an unclosed parenthesis",
             expected_bridge_independent=True,
         ),
         NativeReferenceErrorTestCase(
-            description="error_only_in_expanded_sql_names_the_file",
+            description="fault_inside_macro_output_is_unlocated",
             sql='SELECT * FROM __ref("orders") /* expanded',
             contents='SELECT * FROM __ref("orders") @expand()',
+            source_map=source_map_at(
+                body_start=0,
+                passes=(
+                    (
+                        ExpansionSpan(
+                            source_start=30, source_end=39, output_start=30, output_end=41
+                        ),
+                    ),
+                ),
+            ),
             expected_message="models/orders.sql: SQL reference contains an unclosed block comment",
             expected_bridge_independent=False,
         ),
+        NativeReferenceErrorTestCase(
+            description="authored_fault_shaped_by_an_unclosed_macro_quote_is_unlocated",
+            sql="SELECT 'x || ' || note || 'open",
+            contents="SELECT @q() || ' || note || 'open",
+            source_map=source_map_at(
+                body_start=0,
+                passes=(
+                    (ExpansionSpan(source_start=7, source_end=11, output_start=7, output_end=9),),
+                ),
+            ),
+            expected_message="models/orders.sql: SQL reference contains an unclosed quoted string",
+            expected_bridge_independent=False,
+        ),
+        NativeReferenceErrorTestCase(
+            description="without_a_source_map_the_file_alone_is_named",
+            sql='SELECT *\nFROM __ref("orders"',
+            contents='SELECT *\nFROM __ref("orders"',
+            source_map=None,
+            expected_message="models/orders.sql: SQL reference contains an unclosed parenthesis",
+            expected_bridge_independent=False,
+        ),
     ],
     ids=lambda case: case.description,
 )
@@ -355,6 +405,7 @@ def test_given_native_reference_error_when_extracting_then_raised_located_withou
                 file_path=Path("/project/models/orders.sql"),
                 relative_path=Path("models/orders.sql"),
                 contents=test_case.contents,
+                source_map=test_case.source_map,
             ),
         )
 
@@ -381,7 +432,8 @@ def test_given_native_reference_error_when_extracting_then_raised_located_withou
             syntax=syntax,
             seed=20261009 + offset,
             count=400,
-            expected_minimum_located=150,
+            expected_minimum_located=120,
+            expected_minimum_located_errors=90,
         )
         for offset, syntax in enumerate(LEXICAL_SYNTAXES)
     ],
@@ -405,7 +457,8 @@ def test_given_generated_files_when_reporting_rejected_calls_then_engines_report
     assert (
         mismatches(inputs=[body for _, body in files], expected=python, actual=preview),
         located_diagnostic_count(outcomes=python) >= test_case.expected_minimum_located,
-    ) == ([], True)
+        located_error_count(outcomes=python) >= test_case.expected_minimum_located_errors,
+    ) == ([], True, True)
 
 
 @pytest.mark.parametrize(