From 4a84f8d6bbe5fe9b3fa50970f297afa7b92583a6 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 13:18:04 +0100
Subject: [PATCH 1/7] fix: report unsafe formatting and preserve commented
 implicit aliases

---
 .../src/sql_lint/_helpers/formatter.rs        | 91 +++++++++++++------
 .../src/sql_lint/models.rs                    |  2 +-
 src/sqlbuild/lint/_helpers/native_format.py   | 61 +++++++++++--
 src/sqlbuild/lint/constants.py                |  1 +
 src/sqlbuild/lint/main/run_format.py          | 12 ++-
 .../sqlbuild/cli/commands/main/test_format.py | 57 ++++++++++--
 6 files changed, 174 insertions(+), 50 deletions(-)

diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
index ecc51bb577..1209f2de0f 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
@@ -1,7 +1,7 @@
 use std::collections::BTreeMap;
 use std::str::FromStr;
 
-use polyglot_sql::tokens::Token;
+use polyglot_sql::tokens::{Token, TokenType};
 use polyglot_sql::{Dialect, DialectType, format_by_name};
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
 
@@ -13,7 +13,8 @@ use crate::sql_lint::models::{FormatRequest, FormatResponse};
 
 const COMMENT_ATTACHMENT_FAILURE: &str =
     "native formatter could not preserve comment token attachments";
-const UNSUPPORTED_SQL_FAILURE: &str = "native formatter could not safely format unsupported SQL";
+const UNSUPPORTED_SQL_FAILURE: &str =
+    "native formatter could not safely restore literal or cast spelling";
 
 pub(crate) fn format_json_impl(request_json: &str) -> Result<String, String> {
     let request: FormatRequest =
@@ -31,6 +32,20 @@ pub(crate) fn format_batch_json_impl(request_json: &str) -> Result<String, Strin
 }
 
 fn format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
+    let original = request.sql.clone();
+    match try_format_sql(request) {
+        Ok(value) => Ok(value),
+        Err(reason) => Ok(FormatResponse {
+            version: LINT_API_VERSION,
+            sql: original,
+            changed: false,
+            formatted: false,
+            reason: Some(reason),
+        }),
+    }
+}
+
+fn try_format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
     if request.version != LINT_API_VERSION {
         return Err(format!(
             "unsupported native format request version {}; expected {LINT_API_VERSION}",
@@ -41,15 +56,14 @@ fn format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
     let dialect_type =
         DialectType::from_str(&request.dialect).map_err(|error| error.to_string())?;
     let dialect = Dialect::get(dialect_type);
-    let tokens = match dialect.tokenize(&original) {
-        Ok(value) => value,
-        Err(_) => return response(original, false, false, Some(UNSUPPORTED_SQL_FAILURE)),
-    };
+    let tokens = dialect
+        .tokenize(&original)
+        .map_err(|error| format!("native formatter tokenization failed: {error}"))?;
     let comments = comments_in(&original, &tokens);
     let neutral = neutralize_comments(&original, &comments);
-    if dialect.parse(&neutral).is_err() {
-        return response(original, false, false, Some(UNSUPPORTED_SQL_FAILURE));
-    }
+    dialect
+        .parse(&neutral)
+        .map_err(|error| format!("native formatter could not parse SQL: {error}"))?;
     let semantic_tokens = without_statement_terminators(&tokens);
     let formatted_context = FormatOnceContext {
         dialect_name: &request.dialect,
@@ -58,13 +72,7 @@ fn format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
         original_tokens: &semantic_tokens,
         comments: &comments,
     };
-    let formatted = match format_once(&neutral, &formatted_context) {
-        Ok(value) => value,
-        Err(error) if error == COMMENT_ATTACHMENT_FAILURE => {
-            return response(original, false, false, Some(COMMENT_ATTACHMENT_FAILURE));
-        }
-        Err(_) => return response(original, false, false, Some(UNSUPPORTED_SQL_FAILURE)),
-    };
+    let formatted = format_once(&neutral, &formatted_context)?;
     let formatted_tokens = dialect
         .tokenize(&formatted)
         .map_err(|error| error.to_string())?;
@@ -86,13 +94,7 @@ fn format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
         original_tokens: &second_semantic_tokens,
         comments: &second_comments,
     };
-    let second_pass = match format_once(&second_neutral, &second_context) {
-        Ok(value) => value,
-        Err(error) if error == COMMENT_ATTACHMENT_FAILURE => {
-            return response(original, false, false, Some(COMMENT_ATTACHMENT_FAILURE));
-        }
-        Err(error) => return Err(error),
-    };
+    let second_pass = format_once(&second_neutral, &second_context)?;
     if second_pass != formatted {
         return Err("native formatter output is not idempotent".to_string());
     }
@@ -127,6 +129,7 @@ fn format_once(neutral_sql: &str, context: &FormatOnceContext<'_>) -> Result<Str
     if context.comments.is_empty() {
         return Ok(formatted);
     }
+    formatted = restore_implicit_aliases(formatted, context)?;
     let formatted_tokens = context
         .dialect
         .tokenize(&formatted)
@@ -170,6 +173,42 @@ fn format_once(neutral_sql: &str, context: &FormatOnceContext<'_>) -> Result<Str
     Ok(formatted)
 }
 
+fn restore_implicit_aliases(
+    mut formatted: String,
+    context: &FormatOnceContext<'_>,
+) -> Result<String, String> {
+    let tokens = context
+        .dialect
+        .tokenize(&formatted)
+        .map_err(|error| error.to_string())?;
+    let mut original = context.original_tokens.iter().peekable();
+    let mut removals: Vec<(usize, usize)> = Vec::new();
+    for token in &tokens {
+        if original
+            .peek()
+            .is_some_and(|before| before.token_type == token.token_type)
+        {
+            original.next();
+        } else if token.token_type == TokenType::As {
+            removals.push((token.span.start, token.span.end));
+        } else {
+            return Err(COMMENT_ATTACHMENT_FAILURE.to_string());
+        }
+    }
+    if original.next().is_some() {
+        return Err(COMMENT_ATTACHMENT_FAILURE.to_string());
+    }
+    for (start, end) in removals.into_iter().rev() {
+        let start = char_to_byte(&formatted, start)?;
+        let end = char_to_byte(&formatted, end)?;
+        formatted.replace_range(
+            start..end + usize::from(formatted[end..].starts_with(' ')),
+            "",
+        );
+    }
+    Ok(formatted)
+}
+
 fn restore_unparenthesized_from_values(
     mut formatted: String,
     context: &FormatOnceContext<'_>,
@@ -200,7 +239,7 @@ fn restore_unparenthesized_from_values(
         })
         .collect();
     if authored_parenthesized.len() != formatted_values.len() {
-        return Ok(formatted);
+        return Err("native formatter could not preserve VALUES relation count".to_string());
     }
     let mut removals: Vec<(usize, usize)> = Vec::new();
     for (was_parenthesized, values_index) in
@@ -234,7 +273,7 @@ fn restore_unparenthesized_from_values(
             }
         }
         let Some(close_index) = close_index else {
-            return Ok(formatted);
+            return Err("native formatter could not locate the VALUES wrapper end".to_string());
         };
         let open = &formatted_tokens[wrapper_index];
         let close = &formatted_tokens[close_index];
@@ -544,6 +583,6 @@ fn response(
         changed,
         sql,
         formatted,
-        reason,
+        reason: reason.map(str::to_owned),
     })
 }
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/models.rs b/crates/sqlbuild-rules-native/src/sql_lint/models.rs
index 9d5229419a..90e875457b 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/models.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/models.rs
@@ -132,5 +132,5 @@ pub(crate) struct FormatResponse {
     pub changed: bool,
     pub formatted: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub reason: Option<&'static str>,
+    pub reason: Option<String>,
 }
diff --git a/src/sqlbuild/lint/_helpers/native_format.py b/src/sqlbuild/lint/_helpers/native_format.py
index 00a9d13dda..893d736ab6 100644
--- a/src/sqlbuild/lint/_helpers/native_format.py
+++ b/src/sqlbuild/lint/_helpers/native_format.py
@@ -13,9 +13,12 @@
 from sqlbuild.lint.constants import (
     CARRIAGE_RETURN_LINE_FEED,
     LINE_FEED,
+    LINT_ENGINE_NATIVE,
+    RULE_FORMAT_UNSAFE,
+    VIOLATION_SEVERITY_FAULT,
 )
-from sqlbuild.lint.exceptions import NativeLintError
-from sqlbuild.lint.models import HeaderSpan, InterpolationSite, LintConfig
+from sqlbuild.lint.exceptions import InterpolationRestorationError, NativeLintError
+from sqlbuild.lint.models import HeaderSpan, InterpolationSite, LintConfig, LintViolation
 
 _NATIVE_FORMAT_API_VERSION: int = 1
 
@@ -41,11 +44,15 @@ def with_newline_style(*, contents: str, newline: str) -> str:
 
 
 def format_native_sql_bodies(
-    *, files: dict[Path, str], config: LintConfig, project_dir: Path
-) -> dict[Path, str]:
-    """Format supported SQL bodies while leaving declined bodies unchanged."""
+    *,
+    files: dict[Path, str],
+    config: LintConfig,
+    project_dir: Path,
+) -> tuple[dict[Path, str], list[LintViolation]]:
+    """Format SQL bodies and report every declined body as a named fault."""
 
     prepared_by_path: dict[Path, tuple[_PreparedBody, ...]] = {}
+    faults: list[LintViolation] = []
     requests_by_key: dict[tuple[str, str], dict[str, object]] = {}
     for file_path, contents in sorted(files.items()):
         headers: tuple[HeaderSpan, ...] = scan_headers(contents=contents)
@@ -98,19 +105,53 @@ def format_native_sql_bodies(
             ):
                 raise NativeLintError("native formatter returned invalid SQL or changed state")
             if not formatted:
+                faults.append(
+                    _format_fault(
+                        file_path=file_path,
+                        contents=contents,
+                        start=prepared.start,
+                        reason=str(
+                            response.get("reason") or "native formatter declined without a reason"
+                        ),
+                    )
+                )
                 continue
             if not changed:
                 continue
-            restored: str = restore_interpolation(
-                fixed=raw_sql,
-                sites=prepared.interpolation_sites,
-            )
+            try:
+                restored: str = restore_interpolation(
+                    fixed=raw_sql,
+                    sites=prepared.interpolation_sites,
+                )
+            except InterpolationRestorationError as error:
+                faults.append(
+                    _format_fault(
+                        file_path=file_path,
+                        contents=contents,
+                        start=prepared.start,
+                        reason=str(error),
+                    )
+                )
+                continue
             updated = (
                 f"{updated[: prepared.start]}{restored}{prepared.trailing}{updated[prepared.end :]}"
             )
         if updated != contents:
             formatted_files[file_path] = updated
-    return formatted_files
+    return formatted_files, faults
+
+
+def _format_fault(*, file_path: Path, contents: str, start: int, reason: str) -> LintViolation:
+    return LintViolation(
+        file_path=file_path,
+        line=contents.count(LINE_FEED, 0, start) + 1,
+        column=1,
+        code=RULE_FORMAT_UNSAFE,
+        message=f"SQL body was not formatted: {reason}",
+        severity=VIOLATION_SEVERITY_FAULT,
+        engine=LINT_ENGINE_NATIVE,
+        remediation="Rewrite the unsupported SQL shape or report a minimal formatter reproduction.",
+    )
 
 
 def _format_responses(
diff --git a/src/sqlbuild/lint/constants.py b/src/sqlbuild/lint/constants.py
index 6d517783bd..acd6f8ac60 100644
--- a/src/sqlbuild/lint/constants.py
+++ b/src/sqlbuild/lint/constants.py
@@ -60,6 +60,7 @@
 RULE_LEADING_COMMENT_DESCRIPTION: str = "leading-comment-description"
 RULE_HEADER_WHITESPACE: str = "header-whitespace"
 RULE_HEADER_PARSE: str = "header-parse"
+RULE_FORMAT_UNSAFE: str = "format-unsafe"
 
 PROJECT_CONFIG_FILENAME_KEY: str = "sqlbuild_project.toml"
 FORMAT_SECTION_KEY: str = "format"
diff --git a/src/sqlbuild/lint/main/run_format.py b/src/sqlbuild/lint/main/run_format.py
index 0693b86fa8..47d4320324 100644
--- a/src/sqlbuild/lint/main/run_format.py
+++ b/src/sqlbuild/lint/main/run_format.py
@@ -133,6 +133,7 @@ def _apply_fixes(
     current_files: dict[Path, str] = {
         file_path: updated.get(file_path, contents) for file_path, contents in files.items()
     }
+    faults: list[LintViolation] = []
     _ = prepare_native_header_cache(files=current_files)
     for file_path, contents in sorted(current_files.items()):
         native_result: tuple[str, tuple[LintViolation, ...]] = format_native_headers(
@@ -140,22 +141,24 @@ def _apply_fixes(
             file_path=file_path,
             config=config,
         )
+        faults.extend(native_result[1])
         if native_result[0] != contents:
             updated[file_path] = native_result[0]
     if not config.native_enabled:
         return reject_unparseable_header_rewrites(
             updated=updated,
             config=config,
-            faults=[],
+            faults=faults,
         )
     current_files = {
         file_path: updated.get(file_path, contents) for file_path, contents in files.items()
     }
-    native_formatted: dict[Path, str] = format_native_sql_bodies(
+    native_formatted, native_faults = format_native_sql_bodies(
         files=current_files,
         config=config,
         project_dir=project_dir,
     )
+    faults.extend(native_faults)
     updated.update(native_formatted)
     post_native_files: dict[Path, str] = {
         file_path: updated.get(file_path, contents) for file_path, contents in files.items()
@@ -170,17 +173,18 @@ def _apply_fixes(
         else {}
     )
     if fixture_formatted:
-        final_native: dict[Path, str] = format_native_sql_bodies(
+        final_native, final_faults = format_native_sql_bodies(
             files=fixture_formatted,
             config=config,
             project_dir=project_dir,
         )
+        faults.extend(final_faults)
         updated.update(fixture_formatted)
         updated.update(final_native)
     return reject_unparseable_header_rewrites(
         updated=updated,
         config=config,
-        faults=[],
+        faults=faults,
     )
 
 
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
index 961dcb0c07..b79b22f7d6 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
@@ -32,6 +32,44 @@
 _UNFORMATTED_ORDERS_SQL: str = "MODEL (materialized table);\nselect   1 as order_id\n"
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatCompileIntegrationTestCase(
+            "commented lateral alias", "LATERAL FLATTEN(input => parsed.items, outer => TRUE) f"
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_commented_lateral_alias_when_formatting_then_canonical_sql_is_written(
+    test_case: FormatCompileIntegrationTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A comment plus generated AS used to decline the entire larger query silently."""
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "snowflake"\n')
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    columns: str = ", ".join(f"parsed.order_attribute_{index}" for index in range(47))
+    model.write_text(
+        'MODEL (description "Order items");\n'
+        "WITH parsed AS (SELECT SPLIT(REGEXP_REPLACE(order_text, '[ ]+', ''), ',') AS items, order_id FROM orders "
+        "UNION ALL BY NAME SELECT items, order_id FROM customers), "
+        "distinct_orders AS (SELECT DISTINCT items, order_id FROM parsed), final AS (\n"
+        "-- Expand the order items.\n"
+        f"select f.index, f.value, {columns} FROM parsed, LATERAL FLATTEN(input => parsed.items, outer => TRUE) f) "
+        "SELECT * FROM final\n"
+    )
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 1
+    capsys.readouterr()
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 0
+    capsys.readouterr()
+    formatted: str = model.read_text()
+    assert "    f.index,\n    f.value,\n    parsed.order_attribute_0," in formatted
+    assert "-- Expand the order items." in formatted
+    assert test_case.expected_literal in formatted
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 0
+    assert model.read_text() == formatted
+
+
 @pytest.mark.parametrize(
     "test_case",
     [
@@ -404,14 +442,9 @@ def test_given_unparseable_file_when_formatting_then_file_is_untouched_and_fault
     "test_case",
     [
         FormatterDeclineIntegrationTestCase(
-            description="unsupported SQL body is skipped while header formats",
+            description="unsupported SQL body reports a fault while header formats",
             authored_body="select from\n",
-            expected_exit_code=0,
-        ),
-        FormatterDeclineIntegrationTestCase(
-            description="comment attachment decline is skipped while header formats",
-            authored_body="select order_id order_key /* keep */ from orders\n",
-            expected_exit_code=0,
+            expected_exit_code=1,
         ),
     ],
     ids=lambda case: case.description,
@@ -445,8 +478,14 @@ def test_given_native_formatter_decline_when_formatting_then_header_formats_and_
     assert second_exit == test_case.expected_exit_code
     assert '  description "Orders.",\n' in formatted_once
     assert test_case.authored_body in formatted_once
-    assert "FAULT=0  WARN=0" in first_output
-    assert "FAULT=0  WARN=0" in second_output
+    assert "format-unsafe" in first_output
+    assert "format-unsafe" in second_output
+    assert "orders.sql" in second_output
+    json_exit: int = main(["--project-dir", str(tmp_path), "format", "--check", "--json"])
+    json_output: str = capsys.readouterr().out
+    assert json_exit == 1
+    assert "format-unsafe" in json_output
+    assert json.loads(json_output)
     assert formatted_once == model.read_text(encoding="utf-8")
 
 

From ae780e2b954066ce0190a5be29081424e160faa0 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 13:18:33 +0100
Subject: [PATCH 2/7] feat: add plain join predicate and final CTE naming rules

---
 README.md                                     |  16 +-
 .../src/sql_lint/_helpers/engine.rs           |  33 +++-
 .../src/sql_lint/_helpers/join_predicates.rs  | 128 ++++++++++++++
 .../src/sql_lint/_helpers/mod.rs              |   1 +
 .../src/sql_lint/_helpers/terminal_shape.rs   |  74 ++++++++
 docs/native-lint-rule-matrix.md               |  39 +++-
 .../hourly_activity_with_daily_context.sql    |  29 ++-
 .../rule_engine/_helpers/guidance/skills.py   |   8 +
 .../sqlbuild/cli/commands/main/_test_types.py |   9 +
 .../cli/commands/main/test_sql_readability.py | 167 ++++++++++++++++++
 .../_helpers/engine/test_catalogue.py         |   2 +-
 11 files changed, 491 insertions(+), 15 deletions(-)
 create mode 100644 crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
 create mode 100644 tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py

diff --git a/README.md b/README.md
index 35bff33ae7..a8ade3890d 100644
--- a/README.md
+++ b/README.md
@@ -182,11 +182,11 @@ physical line width in `sqlbuild_project.toml` (the default is `100`):
 line_width = 100
 ```
 
-Select rules in `sqlbuild_project.toml` by exact code or derived family prefix:
+Prefer family selection in `sqlbuild_project.toml` over hand-maintained lists of exact codes:
 
 ```toml
 [rules]
-select = ["SQBRSQL", "XSQBRARCH"]
+select = ["SQBRSQL", "SQBRMODEL", "SQBRGRAPH", "XSQBRARCH"]
 ignore = ["SQBRSQL004"]
 ```
 
@@ -194,6 +194,18 @@ Built-in codes use `SQBR<FAMILY><three digits>`, such as `SQBRSQL001` and
 `SQBRGRAPH101`. Custom codes use `XSQBR<optional family><three digits>`, such as
 `XSQBRARCH001`. A family is always the code with its final three digits removed.
 
+Family selectors automatically include new built-in rules on upgrade. Exact-code lists retain
+their current membership and must be updated manually. In particular, selecting `SQBRSQL` enables
+`SQBRSQL040` (plain JOIN predicates) and `SQBRSQL041` (terminal CTE naming); existing projects may
+need to extract computed join keys into input CTEs and rename their last CTE to `final`.
+See the [SQL rule conventions](docs/native-lint-rule-matrix.md#join-keys-and-final-cte-names)
+for their exact scope and relationship to `SQBRSQL035`.
+
+`sqb format` reports a file-specific `format-unsafe` fault whenever a SQL body cannot be safely
+formatted, including parser, comment-attachment, interpolation-restoration, and idempotence
+failures. The original body is retained, the reason appears in human and JSON output, and both
+formatting and `sqb format --check` exit nonzero. A declined body is never counted as canonical.
+
 Custom rules are ordinary Python beneath `rules/**/*.py`. Only `@rule` functions register; helper
 functions, constants, dataclasses, classes, and nested packages remain ordinary Python. Typed,
 keyword-only parameters determine whether a rule runs once per model or once per project:
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
index e505296172..51a3e300ea 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
@@ -229,6 +229,16 @@ const INLINE_QUERY_RELATION: LintRuleMetadata = LintRuleMetadata {
     remediation: "Extract the derived relation into a named top-level CTE; use explicit LATERAL only when the relation must remain correlated.",
 };
 const TRIVIAL_EQUALITY_TOKEN_COUNT: usize = 3;
+const JOIN_EXPRESSION: LintRuleMetadata = LintRuleMetadata {
+    code: "SQBRSQL040",
+    message: "JOIN ON contains a computed expression",
+    remediation: "Compute join keys in input CTEs and join on plain columns.",
+};
+const FINAL_CTE_NAME: LintRuleMetadata = LintRuleMetadata {
+    code: "SQBRSQL041",
+    message: "Only the last top-level CTE may be named final, and it must use that name",
+    remediation: "Name the last top-level CTE final and rename any other final CTE; SQBRSQL035 checks the terminal SELECT.",
+};
 
 const DEFAULT_RULES: [&str; 13] = [
     NULL_COMPARISON.code,
@@ -246,7 +256,7 @@ const DEFAULT_RULES: [&str; 13] = [
     INLINE_QUERY_RELATION.code,
 ];
 
-const ALL_RULE_METADATA: [&LintRuleMetadata; 39] = [
+const ALL_RULE_METADATA: [&LintRuleMetadata; 41] = [
     &NULL_COMPARISON,
     &IMPLICIT_CARTESIAN_JOIN,
     &JOIN_WITHOUT_CONDITION,
@@ -286,6 +296,8 @@ const ALL_RULE_METADATA: [&LintRuleMetadata; 39] = [
     &RECURSIVE_CTE,
     &CROSS_JOIN,
     &INLINE_QUERY_RELATION,
+    &JOIN_EXPRESSION,
+    &FINAL_CTE_NAME,
 ];
 
 fn is_ceremonial_cte_name(name: &str) -> bool {
@@ -395,6 +407,25 @@ pub(crate) fn lint(request: LintRequest) -> Result<LintResponse, String> {
         enabled: &enabled,
     };
     let mut diagnostics = diagnostics(&context);
+    let depths = token_depths(&tokens);
+    if enabled.contains(JOIN_EXPRESSION.code) {
+        diagnostics.extend(diagnostics_for_spans(
+            &JOIN_EXPRESSION,
+            &crate::sql_lint::_helpers::join_predicates::join_expression_spans(
+                &tokens,
+                &depths,
+                dialect_type,
+            ),
+            Some("computing a join key requires an authored CTE rewrite"),
+        ));
+    }
+    if enabled.contains(FINAL_CTE_NAME.code) && !request.allows_ceremonial_select {
+        diagnostics.extend(diagnostics_for_spans(
+            &FINAL_CTE_NAME,
+            &crate::sql_lint::_helpers::terminal_shape::final_cte_name_spans(&tokens, &depths),
+            Some("renaming a CTE requires updating its references"),
+        ));
+    }
     diagnostics.sort_by_key(|diagnostic| (diagnostic.start, diagnostic.end, diagnostic.code));
     Ok(LintResponse {
         version: LINT_API_VERSION,
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
new file mode 100644
index 0000000000..960e76a070
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
@@ -0,0 +1,128 @@
+use polyglot_sql::DialectType;
+use polyglot_sql::expressions::Expression;
+use polyglot_sql::parser::{Parser, ParserConfig};
+use polyglot_sql::tokens::{Span, Token, TokenType};
+
+pub(super) fn join_expression_spans(
+    tokens: &[Token],
+    depths: &[usize],
+    dialect: DialectType,
+) -> Vec<Span> {
+    let mut spans: Vec<Span> = Vec::new();
+    for (index, token) in tokens.iter().enumerate() {
+        if token.token_type != TokenType::On {
+            continue;
+        }
+        let clause = (0..index).rev().find(|&previous| {
+            depths[previous] == depths[index]
+                && matches!(
+                    tokens[previous].token_type,
+                    TokenType::Join
+                        | TokenType::Select
+                        | TokenType::Where
+                        | TokenType::From
+                        | TokenType::On
+                )
+        });
+        if clause.is_none_or(|previous| tokens[previous].token_type != TokenType::Join) {
+            continue;
+        }
+        let end = (index + 1..tokens.len())
+            .find(|&next| {
+                depths[next] < depths[index]
+                    || (depths[next] == depths[index]
+                        && matches!(
+                            tokens[next].token_type,
+                            TokenType::Join
+                                | TokenType::Where
+                                | TokenType::Group
+                                | TokenType::Having
+                                | TokenType::Qualify
+                                | TokenType::Order
+                                | TokenType::Limit
+                                | TokenType::Union
+                                | TokenType::Intersect
+                                | TokenType::Except
+                                | TokenType::Semicolon
+                        ))
+            })
+            .unwrap_or(tokens.len());
+        let mut parser = Parser::with_config(
+            tokens[index + 1..end].to_vec(),
+            ParserConfig {
+                dialect: Some(dialect),
+                ..Default::default()
+            },
+        );
+        if !matches!(parser.parse_expressions(), Ok(Some(ref predicate)) if plain_predicate(predicate))
+        {
+            spans.push(token.span);
+        }
+    }
+    spans
+}
+
+fn plain_column(expression: &Expression) -> bool {
+    matches!(
+        expression,
+        Expression::Column(_) | Expression::Identifier(_)
+    ) || matches!(expression, Expression::Paren(value) if plain_column(&value.this))
+}
+
+fn literal(expression: &Expression) -> bool {
+    matches!(
+        expression,
+        Expression::Literal(_) | Expression::Boolean(_) | Expression::Null(_)
+    ) || matches!(expression, Expression::Neg(value) if matches!(&value.this, Expression::Literal(_)))
+}
+
+fn operand(expression: &Expression) -> bool {
+    plain_column(expression) || literal(expression)
+}
+
+fn comparison(left: &Expression, right: &Expression) -> bool {
+    operand(left) && operand(right) && (plain_column(left) || plain_column(right))
+}
+
+fn plain_predicate(expression: &Expression) -> bool {
+    match expression {
+        Expression::Paren(value) => plain_predicate(&value.this),
+        Expression::And(value) | Expression::Or(value) => {
+            plain_predicate(&value.left) && plain_predicate(&value.right)
+        }
+        Expression::Eq(value)
+        | Expression::Neq(value)
+        | Expression::Lt(value)
+        | Expression::Lte(value)
+        | Expression::Gt(value)
+        | Expression::Gte(value)
+        | Expression::NullSafeEq(value)
+        | Expression::NullSafeNeq(value)
+        | Expression::Adjacent(value)
+        | Expression::ArrayContainsAll(value)
+        | Expression::ArrayContainedBy(value)
+        | Expression::ArrayOverlaps(value)
+        | Expression::ExtendsLeft(value)
+        | Expression::ExtendsRight(value)
+        | Expression::Match(value)
+        | Expression::TsMatch(value) => comparison(&value.left, &value.right),
+        Expression::Like(value) | Expression::ILike(value) => {
+            comparison(&value.left, &value.right)
+                && value.escape.as_ref().is_none_or(literal)
+                && value.quantifier.is_none()
+        }
+        Expression::Between(value) => {
+            plain_column(&value.this) && operand(&value.low) && operand(&value.high)
+        }
+        Expression::In(value) => {
+            plain_column(&value.this)
+                && value.query.is_none()
+                && value.unnest.is_none()
+                && !value.is_field
+                && !value.expressions.is_empty()
+                && value.expressions.iter().all(literal)
+        }
+        Expression::IsNull(value) => plain_column(&value.this),
+        _ => false,
+    }
+}
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
index 3641ba2ff6..6b9914339b 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
@@ -2,5 +2,6 @@ pub(crate) mod additional;
 pub(crate) mod engine;
 pub(crate) mod formatter;
 pub(crate) mod inline_relations;
+pub(crate) mod join_predicates;
 pub(crate) mod preparation;
 pub(crate) mod terminal_shape;
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
index d897e4e7cd..103e6c5384 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
@@ -2,6 +2,80 @@ use polyglot_sql::tokens::{Span, Token, TokenType};
 
 const CEREMONIAL_SELECT_LITERAL: &str = "1";
 
+pub(super) fn final_cte_name_spans(tokens: &[Token], depths: &[usize]) -> Vec<Span> {
+    let mut names: Vec<(usize, bool)> = Vec::new();
+    for (index, token) in tokens.iter().enumerate() {
+        if token.token_type != TokenType::With {
+            continue;
+        }
+        let depth = depths[index];
+        let first = names.len();
+        let mut next = index + 1;
+        if tokens
+            .get(next)
+            .is_some_and(|token| token.text.eq_ignore_ascii_case("recursive"))
+        {
+            next += 1;
+        }
+        loop {
+            let name = next;
+            let Some(open) = cte_body_open(tokens, depths, name) else {
+                break;
+            };
+            let Some(close) = (open + 1..tokens.len()).find(|&position| {
+                depths[position] == depth + 1 && tokens[position].token_type == TokenType::RParen
+            }) else {
+                break;
+            };
+            names.push((name, false));
+            if tokens
+                .get(close + 1)
+                .is_none_or(|token| token.token_type != TokenType::Comma)
+            {
+                break;
+            }
+            next = close + 2;
+        }
+        if depth == 0
+            && names.len() > first
+            && let Some((_, terminal)) = names.last_mut()
+        {
+            *terminal = true;
+        }
+    }
+    names
+        .into_iter()
+        .filter_map(|(index, terminal)| {
+            let is_final = tokens[index].text.eq_ignore_ascii_case("final");
+            (terminal != is_final).then_some(tokens[index].span)
+        })
+        .collect()
+}
+
+fn cte_body_open(tokens: &[Token], depths: &[usize], name: usize) -> Option<usize> {
+    let mut position = name + 1;
+    if tokens.get(position)?.token_type == TokenType::LParen {
+        position = (position + 1..tokens.len()).find(|&index| {
+            depths[index] == depths[name] + 1 && tokens[index].token_type == TokenType::RParen
+        })? + 1;
+    }
+    if tokens.get(position)?.token_type != TokenType::As {
+        return None;
+    }
+    position += 1;
+    if tokens.get(position)?.token_type == TokenType::Not {
+        position += 1;
+    }
+    if tokens
+        .get(position)?
+        .text
+        .eq_ignore_ascii_case("materialized")
+    {
+        position += 1;
+    }
+    (tokens.get(position)?.token_type == TokenType::LParen).then_some(position)
+}
+
 pub(super) fn collect_terminal_shape_facts(
     tokens: &[Token],
     depths: &[usize],
diff --git a/docs/native-lint-rule-matrix.md b/docs/native-lint-rule-matrix.md
index a31532421b..455b2cf0fb 100644
--- a/docs/native-lint-rule-matrix.md
+++ b/docs/native-lint-rule-matrix.md
@@ -139,6 +139,41 @@ SQLBuild has one canonical formatter and does not expose capitalization rule con
 | `SQBRSQL037` | Optional rule | never | Recursive CTEs require an explicit alternative design. |
 | `SQBRSQL038` | Optional rule | never | Explicit cartesian products require a reasoned suppression. |
 | `SQBRSQL039` | Core compiler | never | Non-lateral derived relations are named as top-level CTEs instead of being hidden inside `FROM` or `JOIN`. |
+| `SQBRSQL040` | Optional rule | never | JOIN predicates compare plain columns and literals; computed keys belong in input CTEs. |
+| `SQBRSQL041` | Optional rule | never | The last top-level CTE is named `final`; no other CTE at any nesting level uses that name. |
+
+## Join keys and final CTE names
+
+`SQBRSQL040` checks every `JOIN ... ON` in the authored SQL resources covered by SQL Rules,
+including joins inside CTEs and subqueries. `USING (...)` is accepted. Allowed predicates are:
+
+- Plain column comparisons, including equality, inequality, range and null-safe comparisons.
+- A plain column compared with a literal, including a signed numeric literal.
+- `BETWEEN` / `NOT BETWEEN` on a plain column with plain-column or literal bounds.
+- `IN` / `NOT IN` on a plain column with a literal list.
+- `IS NULL` / `IS NOT NULL` on a plain column.
+- Parenthesized predicates and `AND` or `OR` combinations whose **every branch** is allowed.
+
+Functions, casts (including `::`), arithmetic, concatenation, CASE, subqueries and other computed
+operands are findings. For example, replace `ON LOWER(o.category) = c.category_key` with a CTE
+that projects `LOWER(category) AS category_key`, then `ON o.category_key = c.category_key`.
+An `OR` with only plain predicates passes; an `OR` containing a computed branch fails.
+
+`SQBRSQL041` checks naming, while `SQBRSQL035` owns the terminal SELECT's plain projection and
+reference to the last top-level CTE. Select both (normally via `SQBRSQL`) for the complete
+convention. The last top-level CTE must be named `final`, compared case-insensitively as in the
+existing SQL rules, including quoted identifiers. Every other CTE, including a nested one, must
+use a different name. Queries without CTEs are unaffected. SQL-test fixture CTEs and their
+ceremonial `SELECT 1` remain exempt from this model naming convention.
+
+Prefer `[rules] select = ["SQBRSQL", "SQBRMODEL", "SQBRGRAPH"]` to exact-code lists. New rules
+in those families apply automatically after an upgrade, so adopting this release can expose
+computed join keys or noncanonical CTE names previously accepted by a family selection.
+
+The formatter preserves authored implicit aliases when comments require token-aligned attachment.
+This avoids a silent decline when canonical generation inserts an optional `AS`, including for
+`LATERAL FLATTEN(...) f`. Every safety decline is surfaced as a `format-unsafe` fault rather
+than being treated as already formatted.
 
 Schema-dependent nullable `NOT IN`, inferred join cardinality, denominator safety, integer division,
 timezone comparison, persisted-output rules, and incremental cursor/replay checks remain compiler or
@@ -147,9 +182,9 @@ generic guesses.
 
 ## Current native totals
 
-- 39 SQLBuild-owned generic native rule codes (`SQBRSQL001`–`SQBRSQL039`)
+- 41 SQLBuild-owned generic native rule codes (`SQBRSQL001`–`SQBRSQL041`)
 - 13 core defaults
-- 26 optional rules selectable by exact code or prefix under `[rules].select`
+- 28 optional rules selectable by exact code or family under `[rules].select`
 - 14 always-or-conditionally formatter-rewritable native analysis types
 - `SQBRSQL000` suppression validation, with stale standalone directives formatter-rewritable
 - five SQLBuild header diagnostics, with whitespace and leading-comment promotion handled when safe
diff --git a/src/sqlbuild/cli/commands/_helpers/playground/templates/waffle_shop/models/marts/hourly_activity_with_daily_context.sql b/src/sqlbuild/cli/commands/_helpers/playground/templates/waffle_shop/models/marts/hourly_activity_with_daily_context.sql
index 1233517a10..e759853fc5 100644
--- a/src/sqlbuild/cli/commands/_helpers/playground/templates/waffle_shop/models/marts/hourly_activity_with_daily_context.sql
+++ b/src/sqlbuild/cli/commands/_helpers/playground/templates/waffle_shop/models/marts/hourly_activity_with_daily_context.sql
@@ -26,12 +26,23 @@ MODEL (
   ],
 );
 
-SELECT
-  h.activity_hour,
-  h.orders_placed,
-  d.orders_placed AS day_orders_placed,
-  h.waffles_ordered,
-  h.revenue_cents
-FROM __ref("hourly_order_activity") h
-INNER JOIN __ref("daily_activity_rollup") d
-  ON @timestamp_trunc('day', 'h.activity_hour') = d.activity_day
+WITH hourly_activity AS (
+  SELECT
+    activity_hour,
+    @timestamp_trunc('day', 'activity_hour') AS activity_day,
+    orders_placed,
+    waffles_ordered,
+    revenue_cents
+  FROM __ref("hourly_order_activity")
+), final AS (
+  SELECT
+    h.activity_hour,
+    h.orders_placed,
+    d.orders_placed AS day_orders_placed,
+    h.waffles_ordered,
+    h.revenue_cents
+  FROM hourly_activity AS h
+  INNER JOIN __ref("daily_activity_rollup") AS d
+    ON h.activity_day = d.activity_day
+)
+SELECT * FROM final
diff --git a/src/sqlbuild/rule_engine/_helpers/guidance/skills.py b/src/sqlbuild/rule_engine/_helpers/guidance/skills.py
index 7b29d67f74..4f2fddc868 100644
--- a/src/sqlbuild/rule_engine/_helpers/guidance/skills.py
+++ b/src/sqlbuild/rule_engine/_helpers/guidance/skills.py
@@ -111,6 +111,14 @@ def render_skills(*, config: RulesConfig, project_dir: Path) -> tuple[str, str]:
 
 def _rule_example(*, rule: Rule) -> str:
     examples: dict[str, str] = {
+        "SQBRSQL040": (
+            "Compute category_key in an input CTE, then JOIN categories AS c "
+            "ON o.category_key = c.category_key."
+        ),
+        "SQBRSQL041": (
+            "WITH orders AS (...), final AS (SELECT order_id FROM orders) "
+            "SELECT order_id FROM final."
+        ),
         "SQBRMODEL101": 'WITH upstream AS (SELECT * FROM __ref("domain__stg__entity")), ...',
         "SQBRMODEL102": "SELECT id, status FROM final; use a lone SELECT * only when allowed.",
         "SQBRMODEL103": "A view uses stg_v, int_v, or mart_v; a table uses a non-v layer.",
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
index 3c8f564b10..b462569bad 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
@@ -677,3 +677,12 @@ class AbsentMicrobatchFullRefreshTestCase:
     setup_build_flags: tuple[str, ...]
     setup_drop_sql: str
     expected_rows: tuple[tuple[object, ...], ...]
+
+
+@dataclass(frozen=True)
+class SqlReadabilityTestCase:
+    """One authored SQL readability contract case."""
+
+    description: str
+    sql: str
+    expected_fault: bool
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py b/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py
new file mode 100644
index 0000000000..b861041aa9
--- /dev/null
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py
@@ -0,0 +1,167 @@
+"""Real compiler and CLI coverage for join-key and final-CTE conventions."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from _pytest.capture import CaptureResult
+
+from sqlbuild.cli.commands.main.entrypoint.entry import main
+from tests.integration.src.sqlbuild.cli.commands.main._test_types import SqlReadabilityTestCase
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SqlReadabilityTestCase("equality", "ON o.order_id = c.order_id", False),
+        SqlReadabilityTestCase("parenthesized column", "ON (o.order_id) = c.order_id", False),
+        SqlReadabilityTestCase("null safe", "ON o.order_id IS NOT DISTINCT FROM c.order_id", False),
+        SqlReadabilityTestCase("signed literal", "ON o.order_id >= -1", False),
+        SqlReadabilityTestCase(
+            "range", "ON o.order_id >= c.order_id AND o.order_id <> c.order_id", False
+        ),
+        SqlReadabilityTestCase("between", "ON o.order_id BETWEEN c.order_id AND c.order_id", False),
+        SqlReadabilityTestCase("literal", "ON o.status = 'success'", False),
+        SqlReadabilityTestCase("literal list", "ON o.status IN ('new', 'success')", False),
+        SqlReadabilityTestCase("null", "ON o.order_id IS NOT NULL", False),
+        SqlReadabilityTestCase("using", "USING (order_id)", False),
+        SqlReadabilityTestCase(
+            "multiple joins",
+            "ON o.order_id = c.order_id LEFT JOIN customers AS p ON p.order_id = c.order_id",
+            False,
+        ),
+        SqlReadabilityTestCase(
+            "second computed join",
+            "ON o.order_id = c.order_id LEFT JOIN customers AS p ON p.order_id + 1 = c.order_id",
+            True,
+        ),
+        SqlReadabilityTestCase(
+            "plain disjunction", "ON (o.order_id = c.order_id OR o.status = 'new')", False
+        ),
+        SqlReadabilityTestCase("function", "ON LOWER(o.status) = c.status", True),
+        SqlReadabilityTestCase("cast", "ON CAST(o.order_id AS BIGINT) = c.order_id", True),
+        SqlReadabilityTestCase("colon cast", "ON o.order_id = c.order_id::BIGINT", True),
+        SqlReadabilityTestCase("arithmetic", "ON o.order_id + 1 = c.order_id", True),
+        SqlReadabilityTestCase("concatenation", "ON o.status || 'x' = c.status", True),
+        SqlReadabilityTestCase(
+            "case", "ON CASE WHEN o.order_id = 1 THEN 1 ELSE 2 END = c.order_id", True
+        ),
+        SqlReadabilityTestCase(
+            "subquery", "ON o.order_id IN (SELECT order_id FROM customers)", True
+        ),
+        SqlReadabilityTestCase(
+            "computed disjunction", "ON o.order_id = c.order_id OR LOWER(o.status) = c.status", True
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_join_predicate_when_compiling_and_running_rules_then_expression_policy_applies(
+    test_case: SqlReadabilityTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "duckdb"\n')
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    model.write_text(
+        'MODEL (description "Orders");\n'
+        "WITH customers AS (SELECT 1 AS order_id, 'success' AS status),\n"
+        "final AS (SELECT o.order_id FROM customers AS o LEFT JOIN customers AS c "
+        f"{test_case.sql}) SELECT * FROM final\n"
+    )
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+    capsys.readouterr()
+    result: int = main(["--project-dir", str(tmp_path), "rules", "--json", "run", "SQBRSQL040"])
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert result == int(test_case.expected_fault)
+    assert bool(payload["findings"]) == test_case.expected_fault
+    assert ("SQBRSQL040" in json.dumps(payload["findings"])) == test_case.expected_fault
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SqlReadabilityTestCase("no CTE", "SELECT 1 AS order_id", False),
+        SqlReadabilityTestCase(
+            "final", "WITH final AS (SELECT 1 AS order_id) SELECT * FROM final", False
+        ),
+        SqlReadabilityTestCase(
+            "quoted mixed case",
+            'WITH "FiNaL" AS (SELECT 1 AS order_id) SELECT * FROM "FiNaL"',
+            False,
+        ),
+        SqlReadabilityTestCase(
+            "column list", "WITH final(order_id) AS (SELECT 1) SELECT * FROM final", False
+        ),
+        SqlReadabilityTestCase(
+            "other name", "WITH orders AS (SELECT 1 AS order_id) SELECT * FROM orders", True
+        ),
+        SqlReadabilityTestCase(
+            "early final",
+            "WITH final AS (SELECT 1 AS order_id), orders AS (SELECT * FROM final) SELECT * FROM orders",
+            True,
+        ),
+        SqlReadabilityTestCase(
+            "nested final",
+            "WITH orders AS (WITH final AS (SELECT 1 AS order_id) SELECT * FROM final), final AS (SELECT * FROM orders) SELECT * FROM final",
+            True,
+        ),
+        SqlReadabilityTestCase(
+            "terminal shape belongs to 035",
+            "WITH final AS (SELECT 1 AS order_id) SELECT order_id + 1 AS order_id FROM final",
+            False,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_cte_names_when_compiling_and_running_rules_then_final_name_policy_applies(
+    test_case: SqlReadabilityTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "duckdb"\n')
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    model.write_text(f'MODEL (description "Orders");\n{test_case.sql}\n')
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+    capsys.readouterr()
+    result: int = main(["--project-dir", str(tmp_path), "rules", "--json", "run", "SQBRSQL041"])
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert result == int(test_case.expected_fault)
+    assert bool(payload["findings"]) == test_case.expected_fault
+    assert ("SQBRSQL041" in json.dumps(payload["findings"])) == test_case.expected_fault
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [SqlReadabilityTestCase("family enables both rules", "SQBRSQL", True)],
+    ids=lambda case: case.description,
+)
+def test_given_family_selection_when_compiling_then_new_rules_are_enforced(
+    test_case: SqlReadabilityTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(
+        f'name = "orders"\nadapter = "duckdb"\n[rules]\nselect = ["{test_case.sql}"]\n'
+    )
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    model.write_text(
+        'MODEL (description "Orders");\n'
+        "WITH customers AS (SELECT 1 AS order_id), "
+        "orders AS (SELECT o.order_id FROM customers AS o INNER JOIN customers AS c "
+        "ON o.order_id + 1 = c.order_id) SELECT order_id FROM orders\n"
+    )
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == int(
+        test_case.expected_fault
+    )
+    output: CaptureResult[str] = capsys.readouterr()
+    assert "SQBRSQL040" in output.out + output.err
+    assert "SQBRSQL041" in output.out + output.err
+    assert main(["--project-dir", str(tmp_path), "rules", "--json", "run", test_case.sql]) == int(
+        test_case.expected_fault
+    )
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert "SQBRSQL040" in json.dumps(payload["findings"])
+    assert "SQBRSQL041" in json.dumps(payload["findings"])
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])
diff --git a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_catalogue.py b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_catalogue.py
index 900a6762ce..8b63690055 100644
--- a/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_catalogue.py
+++ b/tests/unit/src/sqlbuild/rule_engine/_helpers/engine/test_catalogue.py
@@ -66,7 +66,7 @@
     "SQBRTEST203",
     "SQBRTEST301",
 )
-SQL_CODES: tuple[str, ...] = tuple(f"SQBRSQL{number:03d}" for number in range(1, 40))
+SQL_CODES: tuple[str, ...] = tuple(f"SQBRSQL{number:03d}" for number in range(1, 42))
 BUILT_IN_CODES: tuple[str, ...] = tuple(sorted((*RULES_BUILT_IN_CODES, *SQL_CODES)))
 STRUCTURE_CODES: tuple[str, ...] = (
     "SQBRMODEL101",

From 93b331ca63f77f281b696c7764df0b0f7a50dea2 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 14:21:34 +0100
Subject: [PATCH 3/7] fix: preserve authored SQL syntax through native
 formatting

---
 README.md                                     |   9 +-
 .../src/sql_lint/_helpers/formatter.rs        | 177 ++++----
 .../src/sql_lint/_helpers/formatter_syntax.rs | 401 ++++++++++++++++++
 .../src/sql_lint/_helpers/mod.rs              |   1 +
 .../src/sql_lint/constants.rs                 |   3 +
 .../src/sql_lint/models.rs                    |   2 +
 .../src/sql_lint/tests/test_sql_lint.rs       |   2 +-
 src/sqlbuild/lint/_helpers/native_format.py   |   2 +
 src/sqlbuild/lint/main/run_format.py          |   4 +
 .../sqlbuild/cli/commands/main/_test_types.py |  10 +
 .../sqlbuild/cli/commands/main/test_format.py |   9 +-
 .../commands/main/test_format_preservation.py | 133 ++++++
 12 files changed, 666 insertions(+), 87 deletions(-)
 create mode 100644 crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
 create mode 100644 tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py

diff --git a/README.md b/README.md
index a8ade3890d..cee15a6433 100644
--- a/README.md
+++ b/README.md
@@ -203,9 +203,16 @@ for their exact scope and relationship to `SQBRSQL035`.
 
 `sqb format` reports a file-specific `format-unsafe` fault whenever a SQL body cannot be safely
 formatted, including parser, comment-attachment, interpolation-restoration, and idempotence
-failures. The original body is retained, the reason appears in human and JSON output, and both
+failures. The entire original file is retained (including its headers and fixtures), the reason
+appears in human and JSON output, and both
 formatting and `sqb format --check` exit nonzero. A declined body is never counted as canonical.
 
+Formatting preserves authored cast types, postfix casts, quoted literals, variant paths, typed
+lambda parameters, and supported SQL function spellings while applying canonical layout. It uses
+the compiler's trusted-SQL function-depth budget and does not impose the separate browser-oriented
+UNION-chain limit from Polyglot's convenience formatting API. CTE-producing macros are formatted as
+authored macro calls rather than expanded project SQL.
+
 Custom rules are ordinary Python beneath `rules/**/*.py`. Only `@rule` functions register; helper
 functions, constants, dataclasses, classes, and nested packages remain ordinary Python. Typed,
 keyword-only parameters determine whether a rule runs once per model or once per project:
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
index 1209f2de0f..5f93cdf520 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
@@ -2,9 +2,10 @@ use std::collections::BTreeMap;
 use std::str::FromStr;
 
 use polyglot_sql::tokens::{Token, TokenType};
-use polyglot_sql::{Dialect, DialectType, format_by_name};
+use polyglot_sql::{ComplexityGuardOptions, Dialect, DialectType, ParseOptions};
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
 
+use crate::sql_lint::_helpers::formatter_syntax::protect_syntax;
 use crate::sql_lint::constants::{
     CAST_TYPE_SEPARATOR_KEYWORD, CLOSE_PARENTHESIS, LINT_API_VERSION, OPEN_PARENTHESIS,
     VALUES_RELATION_PREFIX_TOKEN_COUNT,
@@ -56,17 +57,24 @@ fn try_format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
     let dialect_type =
         DialectType::from_str(&request.dialect).map_err(|error| error.to_string())?;
     let dialect = Dialect::get(dialect_type);
+    let parse_options = ParseOptions {
+        complexity_guard: Some(ComplexityGuardOptions {
+            max_function_call_depth: request.max_function_call_depth,
+            ..Default::default()
+        }),
+    };
     let tokens = dialect
         .tokenize(&original)
         .map_err(|error| format!("native formatter tokenization failed: {error}"))?;
     let comments = comments_in(&original, &tokens);
     let neutral = neutralize_comments(&original, &comments);
+    let validation_sql = protect_syntax(&neutral, &tokens, &[], false)?;
     dialect
-        .parse(&neutral)
+        .parse_with_options(&validation_sql.sql, &parse_options)
         .map_err(|error| format!("native formatter could not parse SQL: {error}"))?;
     let semantic_tokens = without_statement_terminators(&tokens);
     let formatted_context = FormatOnceContext {
-        dialect_name: &request.dialect,
+        parse_options: &parse_options,
         dialect: &dialect,
         original_sql: &neutral,
         original_tokens: &semantic_tokens,
@@ -78,8 +86,9 @@ fn try_format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
         .map_err(|error| error.to_string())?;
     let formatted_neutral =
         neutralize_comments(&formatted, &comments_in(&formatted, &formatted_tokens));
+    let validation_sql = protect_syntax(&formatted_neutral, &formatted_tokens, &[], false)?;
     let _ = dialect
-        .parse(&formatted_neutral)
+        .parse_with_options(&validation_sql.sql, &parse_options)
         .map_err(|error| error.to_string())?;
     let second_tokens = dialect
         .tokenize(&formatted)
@@ -88,7 +97,7 @@ fn try_format_sql(request: FormatRequest) -> Result<FormatResponse, String> {
     let second_neutral = neutralize_comments(&formatted, &second_comments);
     let second_semantic_tokens = without_statement_terminators(&second_tokens);
     let second_context = FormatOnceContext {
-        dialect_name: &request.dialect,
+        parse_options: &parse_options,
         dialect: &dialect,
         original_sql: &second_neutral,
         original_tokens: &second_semantic_tokens,
@@ -112,7 +121,7 @@ struct Comment {
 }
 
 struct FormatOnceContext<'a> {
-    dialect_name: &'a str,
+    parse_options: &'a ParseOptions,
     dialect: &'a Dialect,
     original_sql: &'a str,
     original_tokens: &'a [Token],
@@ -120,12 +129,28 @@ struct FormatOnceContext<'a> {
 }
 
 fn format_once(neutral_sql: &str, context: &FormatOnceContext<'_>) -> Result<String, String> {
-    let mut formatted = format_by_name(neutral_sql, context.dialect_name)
-        .map_err(|error| error.to_string())?
-        .join(";\n");
+    let protected = protect_syntax(
+        neutral_sql,
+        context.original_tokens,
+        &cast_type_token_ranges(context)?,
+        true,
+    )?;
+    let expressions = context
+        .dialect
+        .parse_with_options(&protected.sql, context.parse_options)
+        .map_err(|error| error.to_string())?;
+    let formatted_statements: Result<Vec<String>, String> = expressions
+        .iter()
+        .map(|expression| {
+            context
+                .dialect
+                .generate_pretty(expression)
+                .map_err(|error| error.to_string())
+        })
+        .collect();
+    let mut formatted = protected.restore(formatted_statements?.join(";\n"), context.dialect)?;
     formatted = restore_unparenthesized_from_values(formatted, context)?;
-    formatted = restore_string_literals(formatted, context)?;
-    formatted = restore_cast_type_spellings(formatted, context)?;
+    formatted = restore_null_treatment(formatted, context)?;
     if context.comments.is_empty() {
         return Ok(formatted);
     }
@@ -192,7 +217,11 @@ fn restore_implicit_aliases(
         } else if token.token_type == TokenType::As {
             removals.push((token.span.start, token.span.end));
         } else {
-            return Err(COMMENT_ATTACHMENT_FAILURE.to_string());
+            return Err(format!(
+                "{COMMENT_ATTACHMENT_FAILURE}: expected {:?}, generated {:?}",
+                original.peek().map(|token| token.token_type),
+                token.token_type
+            ));
         }
     }
     if original.next().is_some() {
@@ -209,6 +238,52 @@ fn restore_implicit_aliases(
     Ok(formatted)
 }
 
+fn restore_null_treatment(
+    mut formatted: String,
+    context: &FormatOnceContext<'_>,
+) -> Result<String, String> {
+    let authored: Vec<bool> = context
+        .original_tokens
+        .windows(3)
+        .filter(|tokens| {
+            matches!(
+                tokens[0].text.to_ascii_uppercase().as_str(),
+                "IGNORE" | "RESPECT"
+            ) && tokens[1].text.eq_ignore_ascii_case("NULLS")
+        })
+        .map(|tokens| tokens[2].token_type == TokenType::RParen)
+        .collect();
+    if authored.is_empty() {
+        return Ok(formatted);
+    }
+    let tokens = context
+        .dialect
+        .tokenize(&formatted)
+        .map_err(|error| error.to_string())?;
+    let generated: Vec<usize> = (1..tokens.len().saturating_sub(1))
+        .filter(|&index| {
+            matches!(
+                tokens[index].text.to_ascii_uppercase().as_str(),
+                "IGNORE" | "RESPECT"
+            ) && tokens[index + 1].text.eq_ignore_ascii_case("NULLS")
+        })
+        .collect();
+    if authored.len() != generated.len() {
+        return Err("native formatter changed null-treatment clauses".to_string());
+    }
+    for (inside, index) in authored.into_iter().zip(generated).rev() {
+        if inside && tokens[index - 1].token_type == TokenType::RParen {
+            let start = char_to_byte(&formatted, tokens[index - 1].span.start)?;
+            let end = char_to_byte(&formatted, tokens[index + 1].span.end)?;
+            formatted.replace_range(
+                start..end,
+                &format!(" {} NULLS)", tokens[index].text.to_ascii_uppercase()),
+            );
+        }
+    }
+    Ok(formatted)
+}
+
 fn restore_unparenthesized_from_values(
     mut formatted: String,
     context: &FormatOnceContext<'_>,
@@ -386,73 +461,6 @@ fn trailing_comment_insertion(
     }
 }
 
-fn restore_string_literals(
-    mut formatted: String,
-    context: &FormatOnceContext<'_>,
-) -> Result<String, String> {
-    let originals: Vec<String> = context
-        .original_tokens
-        .iter()
-        .filter(|token| token.token_type == polyglot_sql::tokens::TokenType::String)
-        .map(|token| char_slice(context.original_sql, token.span.start, token.span.end))
-        .collect::<Option<Vec<String>>>()
-        .ok_or_else(|| UNSUPPORTED_SQL_FAILURE.to_string())?;
-    let formatted_tokens = context
-        .dialect
-        .tokenize(&formatted)
-        .map_err(|error| error.to_string())?;
-    let formatted_strings: Vec<&Token> = formatted_tokens
-        .iter()
-        .filter(|token| token.token_type == polyglot_sql::tokens::TokenType::String)
-        .collect();
-    if originals.len() != formatted_strings.len() {
-        return Err(UNSUPPORTED_SQL_FAILURE.to_string());
-    }
-    for (token, original) in formatted_strings.into_iter().zip(originals).rev() {
-        let start = char_to_byte(&formatted, token.span.start)?;
-        let end = char_to_byte(&formatted, token.span.end)?;
-        formatted.replace_range(start..end, &original);
-    }
-    Ok(formatted)
-}
-
-fn restore_cast_type_spellings(
-    mut formatted: String,
-    context: &FormatOnceContext<'_>,
-) -> Result<String, String> {
-    let type_ranges = cast_type_token_ranges(context)?;
-    if type_ranges.is_empty() {
-        return Ok(formatted);
-    }
-    let formatted_tokens = context
-        .dialect
-        .tokenize(&formatted)
-        .map_err(|error| error.to_string())?;
-    if context.original_tokens.len() != formatted_tokens.len() {
-        return Err(UNSUPPORTED_SQL_FAILURE.to_string());
-    }
-    let mut type_indices: Vec<usize> = type_ranges
-        .into_iter()
-        .flat_map(|(start, end)| start..=end)
-        .collect();
-    type_indices.sort_unstable();
-    type_indices.dedup();
-    for index in type_indices.into_iter().rev() {
-        let original_token = &context.original_tokens[index];
-        let formatted_token = &formatted_tokens[index];
-        let original = char_slice(
-            context.original_sql,
-            original_token.span.start,
-            original_token.span.end,
-        )
-        .ok_or_else(|| UNSUPPORTED_SQL_FAILURE.to_string())?;
-        let start = char_to_byte(&formatted, formatted_token.span.start)?;
-        let end = char_to_byte(&formatted, formatted_token.span.end)?;
-        formatted.replace_range(start..end, &original);
-    }
-    Ok(formatted)
-}
-
 fn cast_type_token_ranges(context: &FormatOnceContext<'_>) -> Result<Vec<(usize, usize)>, String> {
     let mut ranges: Vec<(usize, usize)> = Vec::new();
     let mut cast_type_starts: Vec<Option<usize>> = Vec::new();
@@ -492,8 +500,15 @@ fn char_slice(value: &str, start: usize, end: usize) -> Option<String> {
 fn without_statement_terminators(tokens: &[Token]) -> Vec<Token> {
     tokens
         .iter()
-        .filter(|token| token.token_type != polyglot_sql::tokens::TokenType::Semicolon)
-        .cloned()
+        .enumerate()
+        .filter(|(index, token)| {
+            token.token_type != TokenType::Semicolon
+                && !(token.token_type == TokenType::Comma
+                    && tokens
+                        .get(index + 1)
+                        .is_some_and(|next| next.token_type == TokenType::From))
+        })
+        .map(|(_, token)| token.clone())
         .collect()
 }
 
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
new file mode 100644
index 0000000000..91c3266f24
--- /dev/null
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
@@ -0,0 +1,401 @@
+use crate::sql_lint::_helpers::engine::token_depths;
+use crate::sql_lint::constants::{
+    CTE_MACRO_PLACEHOLDER_LITERAL, POSITION_ARGUMENT_SEPARATOR, POSTFIX_CAST_PREFIX_TOKEN_COUNT,
+};
+use polyglot_sql::Dialect;
+use polyglot_sql::tokens::{Token, TokenType};
+use std::collections::{HashMap, HashSet};
+
+pub(super) struct ProtectedSql {
+    pub sql: String,
+    fragments: Vec<(String, String)>,
+    in_functions: HashSet<String>,
+}
+
+impl ProtectedSql {
+    pub(super) fn restore(&self, formatted: String, dialect: &Dialect) -> Result<String, String> {
+        let tokens = dialect
+            .tokenize(&formatted)
+            .map_err(|error| error.to_string())?;
+        let mut positions: HashMap<String, Vec<usize>> = HashMap::new();
+        for (index, token) in tokens.iter().enumerate() {
+            positions
+                .entry(token.text.to_ascii_lowercase())
+                .or_default()
+                .push(index);
+        }
+        let mut edits: Vec<(usize, usize, String)> = Vec::new();
+        for (placeholder, original) in &self.fragments {
+            let key = placeholder
+                .strip_suffix(" AS (SELECT 1)")
+                .unwrap_or(placeholder)
+                .trim_start_matches(':')
+                .trim_start_matches('.')
+                .trim_matches('\'')
+                .to_ascii_lowercase();
+            let found = positions
+                .get(&key)
+                .filter(|found| found.len() == 1)
+                .ok_or_else(|| {
+                    "native formatter could not restore a protected SQL fragment".to_string()
+                })?;
+            let index = found[0];
+            if self.in_functions.contains(&key) {
+                let mut depth = 0;
+                let separator = (index + 1..tokens.len())
+                    .find(|&position| {
+                        match tokens[position].token_type {
+                            TokenType::LParen => depth += 1,
+                            TokenType::RParen => depth -= 1,
+                            _ => {}
+                        }
+                        depth == 1 && tokens[position].token_type == TokenType::Comma
+                    })
+                    .ok_or_else(|| "native formatter lost a POSITION separator".to_string())?;
+                edits.push((
+                    tokens[separator].span.start,
+                    tokens[separator].span.end,
+                    " IN".to_string(),
+                ));
+            }
+            if let Some(sentinel) = placeholder.strip_suffix(" AS (SELECT 1)") {
+                let _ = sentinel;
+                edits.push(restore_cte_macro(&tokens, index, original)?);
+                continue;
+            }
+            if placeholder.starts_with("::") {
+                edits.extend(restore_postfix_cast(&tokens, index, original)?);
+                continue;
+            }
+            let start = if placeholder.starts_with('.') {
+                if index == 0 || tokens[index - 1].token_type != TokenType::Dot {
+                    return Err("native formatter changed a protected path accessor".to_string());
+                }
+                tokens[index - 1].span.start
+            } else {
+                tokens[index].span.start
+            };
+            edits.push((start, tokens[index].span.end, original.clone()));
+        }
+        edits.sort_by_key(|edit| edit.0);
+        let chars: Vec<char> = formatted.chars().collect();
+        let mut result = String::new();
+        let mut cursor = 0;
+        for (start, end, replacement) in edits {
+            if start < cursor {
+                return Err("native formatter produced overlapping syntax restorations".to_string());
+            }
+            result.extend(&chars[cursor..start]);
+            result.push_str(&replacement);
+            cursor = end;
+        }
+        result.extend(&chars[cursor..]);
+        Ok(result)
+    }
+}
+
+pub(super) fn protect_syntax(
+    sql: &str,
+    tokens: &[Token],
+    cast_types: &[(usize, usize)],
+    expressions: bool,
+) -> Result<ProtectedSql, String> {
+    let mut prefix = "__sqlbuild_format_".to_string();
+    while sql.to_ascii_lowercase().contains(&prefix) {
+        prefix.push('_');
+    }
+    let mut ranges: Vec<(usize, usize, String)> = Vec::new();
+    let cte_macros = cte_macro_indices(tokens);
+    let depths = token_depths(tokens);
+    let mut separators: HashSet<usize> = HashSet::new();
+    let mut in_functions: HashSet<String> = HashSet::new();
+    let mut index = 0;
+    while index < tokens.len() {
+        let token = &tokens[index];
+        let sentinel = format!("{prefix}{}__", ranges.len());
+        if separators.contains(&index) {
+            ranges.push((token.span.start, token.span.end, ",".to_string()));
+            index += 1;
+            continue;
+        }
+        let (end, replacement) = if cte_macros.contains(&index) {
+            (index, format!("{sentinel} AS (SELECT 1)"))
+        } else if !expressions {
+            index += 1;
+            continue;
+        } else if let Some((_, end)) = cast_types.iter().find(|(start, _)| *start == index) {
+            (*end, sentinel)
+        } else if token.token_type == TokenType::DColon {
+            (
+                postfix_type_end(tokens, index + 1)?,
+                format!("::{sentinel}"),
+            )
+        } else if token.token_type == TokenType::Colon
+            && index > 0
+            && tokens[index - 1].token_type != TokenType::String
+            && tokens.get(index + 1).is_some_and(|next| {
+                next.text
+                    .chars()
+                    .next()
+                    .is_some_and(|character| character.is_alphanumeric() || character == '_')
+                    || matches!(
+                        next.token_type,
+                        TokenType::QuotedIdentifier | TokenType::String
+                    )
+            })
+        {
+            (index + 1, format!(".{sentinel}"))
+        } else if is_typed_literal(tokens, index) {
+            (index + 1, sentinel)
+        } else if matches!(token.token_type, TokenType::Var | TokenType::Identifier)
+            && tokens.get(index + 1).is_some_and(|next| {
+                matches!(next.token_type, TokenType::Var | TokenType::Identifier)
+            })
+            && tokens
+                .get(index + 2)
+                .is_some_and(|next| next.token_type == TokenType::Arrow)
+        {
+            (index + 1, sentinel)
+        } else if matches!(token.text.to_ascii_uppercase().as_str(), "MOD" | "POSITION")
+            && tokens
+                .get(index + 1)
+                .is_some_and(|next| next.token_type == TokenType::LParen)
+        {
+            if token.text.eq_ignore_ascii_case("POSITION") {
+                let separator = (index + 2..tokens.len())
+                    .take_while(|&position| {
+                        !(tokens[position].token_type == TokenType::RParen
+                            && depths[position] == depths[index] + 1)
+                    })
+                    .find(|&position| {
+                        tokens[position].token_type == TokenType::In
+                            && depths[position] == depths[index] + 1
+                    });
+                if let Some(separator) = separator {
+                    separators.insert(separator);
+                    in_functions.insert(sentinel.clone());
+                }
+            }
+            (index, sentinel)
+        } else if matches!(
+            token.token_type,
+            TokenType::String | TokenType::DollarString
+        ) {
+            (index, format!("'{sentinel}'"))
+        } else {
+            index += 1;
+            continue;
+        };
+        ranges.push((token.span.start, tokens[end].span.end, replacement));
+        index = end + 1;
+    }
+    let chars: Vec<char> = sql.chars().collect();
+    let token_indices: HashMap<usize, usize> = tokens
+        .iter()
+        .enumerate()
+        .map(|(index, token)| (token.span.start, index))
+        .collect();
+    let mut prepared = String::new();
+    let mut fragments: Vec<(String, String)> = Vec::new();
+    let mut cursor = 0;
+    for (start, end, replacement) in ranges {
+        prepared.extend(&chars[cursor..start]);
+        prepared.push_str(&replacement);
+        if replacement != POSITION_ARGUMENT_SEPARATOR {
+            let first = token_indices[&start];
+            let last = (first..tokens.len())
+                .find(|&index| tokens[index].span.end == end)
+                .ok_or_else(|| "native formatter lost a source fragment boundary".to_string())?;
+            fragments.push((
+                replacement,
+                canonical_fragment(&chars, &tokens[first..=last]),
+            ));
+        }
+        cursor = end;
+    }
+    prepared.extend(&chars[cursor..]);
+    Ok(ProtectedSql {
+        sql: prepared,
+        fragments,
+        in_functions,
+    })
+}
+
+fn canonical_fragment(chars: &[char], tokens: &[Token]) -> String {
+    let mut result = String::new();
+    for (index, token) in tokens.iter().enumerate() {
+        if index > 0
+            && !matches!(
+                token.token_type,
+                TokenType::LParen
+                    | TokenType::RParen
+                    | TokenType::Comma
+                    | TokenType::Dot
+                    | TokenType::Colon
+                    | TokenType::DColon
+                    | TokenType::LBracket
+                    | TokenType::RBracket
+                    | TokenType::Lt
+                    | TokenType::Gt
+            )
+            && !matches!(
+                tokens[index - 1].token_type,
+                TokenType::LParen
+                    | TokenType::Dot
+                    | TokenType::Colon
+                    | TokenType::DColon
+                    | TokenType::LBracket
+                    | TokenType::Lt
+            )
+        {
+            result.push(' ');
+        }
+        result.extend(&chars[token.span.start..token.span.end]);
+    }
+    result
+}
+
+fn cte_macro_indices(tokens: &[Token]) -> HashSet<usize> {
+    let depths = token_depths(tokens);
+    let mut scopes: HashSet<usize> = HashSet::new();
+    let mut indices: HashSet<usize> = HashSet::new();
+    for (index, token) in tokens.iter().enumerate() {
+        if token.token_type == TokenType::With {
+            scopes.insert(depths[index]);
+        } else if token.token_type == TokenType::Select {
+            scopes.remove(&depths[index]);
+        } else if token.text.starts_with("__sqb_lint_")
+            && scopes.contains(&depths[index])
+            && index > 0
+            && matches!(
+                tokens[index - 1].token_type,
+                TokenType::With | TokenType::Comma
+            )
+            && tokens
+                .get(index + 1)
+                .is_some_and(|next| matches!(next.token_type, TokenType::Comma | TokenType::Select))
+        {
+            indices.insert(index);
+        }
+    }
+    indices
+}
+
+fn restore_cte_macro(
+    tokens: &[Token],
+    start: usize,
+    original: &str,
+) -> Result<(usize, usize, String), String> {
+    let end = start + 5;
+    if tokens
+        .get(end)
+        .is_none_or(|token| token.token_type != TokenType::RParen)
+        || tokens[start + 1].token_type != TokenType::As
+        || tokens[start + 2].token_type != TokenType::LParen
+        || tokens[start + 3].token_type != TokenType::Select
+        || tokens[start + 4].text != CTE_MACRO_PLACEHOLDER_LITERAL
+    {
+        return Err("native formatter changed a CTE macro placeholder".to_string());
+    }
+    Ok((
+        tokens[start].span.start,
+        tokens[end].span.end,
+        original.to_string(),
+    ))
+}
+
+fn restore_postfix_cast(
+    tokens: &[Token],
+    index: usize,
+    original: &str,
+) -> Result<Vec<(usize, usize, String)>, String> {
+    if index < POSTFIX_CAST_PREFIX_TOKEN_COUNT
+        || tokens[index - 1].token_type != TokenType::As
+        || tokens
+            .get(index + 1)
+            .is_none_or(|token| token.token_type != TokenType::RParen)
+    {
+        return Err("native formatter changed a postfix cast wrapper".to_string());
+    }
+    let mut depth = 0;
+    let mut open = None;
+    for position in (0..index - 1).rev() {
+        match tokens[position].token_type {
+            TokenType::RParen => depth += 1,
+            TokenType::LParen if depth == 0 => {
+                open = Some(position);
+                break;
+            }
+            TokenType::LParen => depth -= 1,
+            _ => {}
+        }
+    }
+    let open = open
+        .filter(|&position| position > 0 && tokens[position - 1].token_type == TokenType::Cast)
+        .ok_or_else(|| "native formatter could not identify a postfix cast wrapper".to_string())?;
+    Ok(vec![
+        (
+            tokens[open - 1].span.start,
+            tokens[open + 1].span.start,
+            String::new(),
+        ),
+        (
+            tokens[index - 2].span.end,
+            tokens[index + 1].span.end,
+            original.to_string(),
+        ),
+    ])
+}
+
+fn is_typed_literal(tokens: &[Token], index: usize) -> bool {
+    matches!(
+        tokens[index].text.to_ascii_uppercase().as_str(),
+        "DATE"
+            | "TIME"
+            | "TIMESTAMP"
+            | "TIMESTAMPTZ"
+            | "TIMESTAMP_NTZ"
+            | "TIMESTAMP_LTZ"
+            | "TIMESTAMP_TZ"
+    ) && tokens.get(index + 1).is_some_and(|token| {
+        matches!(
+            token.token_type,
+            TokenType::String | TokenType::DollarString
+        )
+    })
+}
+
+fn postfix_type_end(tokens: &[Token], start: usize) -> Result<usize, String> {
+    let Some(_) = tokens.get(start) else {
+        return Err("native formatter could not locate a postfix cast type".to_string());
+    };
+    let mut end = start;
+    while tokens.get(end + 1).is_some_and(|token| {
+        matches!(
+            token.text.to_ascii_uppercase().as_str(),
+            "PRECISION" | "VARYING" | "WITH" | "WITHOUT" | "TIME" | "ZONE"
+        )
+    }) {
+        end += 1;
+    }
+    if tokens
+        .get(end + 1)
+        .is_some_and(|token| token.token_type == TokenType::LParen)
+    {
+        let mut depth = 0;
+        for (index, token) in tokens.iter().enumerate().skip(end + 1) {
+            match token.token_type {
+                TokenType::LParen => depth += 1,
+                TokenType::RParen => {
+                    depth -= 1;
+                    if depth == 0 {
+                        return Ok(index);
+                    }
+                }
+                _ => {}
+            }
+        }
+        return Err("native formatter could not locate postfix cast parameters".to_string());
+    }
+    Ok(end)
+}
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
index 6b9914339b..18467af6db 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/mod.rs
@@ -1,6 +1,7 @@
 pub(crate) mod additional;
 pub(crate) mod engine;
 pub(crate) mod formatter;
+pub(crate) mod formatter_syntax;
 pub(crate) mod inline_relations;
 pub(crate) mod join_predicates;
 pub(crate) mod preparation;
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/constants.rs b/crates/sqlbuild-rules-native/src/sql_lint/constants.rs
index 539928f25d..873ef0ab42 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/constants.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/constants.rs
@@ -3,3 +3,6 @@ pub(crate) const CLOSE_PARENTHESIS: &str = ")";
 pub(crate) const LINT_API_VERSION: u32 = 1;
 pub(crate) const OPEN_PARENTHESIS: &str = "(";
 pub(crate) const VALUES_RELATION_PREFIX_TOKEN_COUNT: usize = 2;
+pub(crate) const POSITION_ARGUMENT_SEPARATOR: &str = ",";
+pub(crate) const CTE_MACRO_PLACEHOLDER_LITERAL: &str = "1";
+pub(crate) const POSTFIX_CAST_PREFIX_TOKEN_COUNT: usize = 2;
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/models.rs b/crates/sqlbuild-rules-native/src/sql_lint/models.rs
index 90e875457b..b2848a5f73 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/models.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/models.rs
@@ -123,6 +123,8 @@ pub(crate) struct FormatRequest {
     pub version: u32,
     pub sql: String,
     pub dialect: String,
+    #[serde(default)]
+    pub max_function_call_depth: Option<usize>,
 }
 
 #[derive(Debug, Serialize)]
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
index 71155cae67..5501eeb3c9 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs
@@ -1896,7 +1896,7 @@ fn given_format_cases_when_formatting_then_output_matches() -> Result<(), String
         test_types::FormatTestCase {
             description: "unparenthesized values relation remains unparenthesized",
             sql: "select column1::varchar as customer_key, column2::integer as order_count from values ('c1', 1), ('c2', 2)",
-            expected_sql: "SELECT\n  CAST(column1 AS VARCHAR) AS customer_key,\n  CAST(column2 AS INT) AS order_count\nFROM VALUES ('c1', 1), ('c2', 2)",
+            expected_sql: "SELECT\n  column1::varchar AS customer_key,\n  column2::integer AS order_count\nFROM VALUES ('c1', 1), ('c2', 2)",
             expected_changed: true,
         },
     ];
diff --git a/src/sqlbuild/lint/_helpers/native_format.py b/src/sqlbuild/lint/_helpers/native_format.py
index 893d736ab6..c6b046c277 100644
--- a/src/sqlbuild/lint/_helpers/native_format.py
+++ b/src/sqlbuild/lint/_helpers/native_format.py
@@ -8,6 +8,7 @@
 from typing import Any
 
 import sqlbuild._native as _native
+from sqlbuild.compiler.sql_analysis.constants import POLYGLOT_MAX_FUNCTION_CALL_DEPTH
 from sqlbuild.lint._helpers.headers import lint_body_ranges, scan_headers
 from sqlbuild.lint._helpers.sqlbuild_tokens import neutralize_interpolation, restore_interpolation
 from sqlbuild.lint.constants import (
@@ -75,6 +76,7 @@ def format_native_sql_bodies(
                     "version": _NATIVE_FORMAT_API_VERSION,
                     "sql": neutralized,
                     "dialect": config.dialect,
+                    "max_function_call_depth": POLYGLOT_MAX_FUNCTION_CALL_DEPTH,
                 },
             )
             prepared_bodies.append(
diff --git a/src/sqlbuild/lint/main/run_format.py b/src/sqlbuild/lint/main/run_format.py
index 47d4320324..145fbae72f 100644
--- a/src/sqlbuild/lint/main/run_format.py
+++ b/src/sqlbuild/lint/main/run_format.py
@@ -54,6 +54,10 @@ def run_format(
         discovered_inputs=discovered_inputs,
         fixtures_only=fixtures_only,
     )
+    declined_paths: set[Path] = {fault.file_path for fault in format_faults}
+    updated_contents: dict[Path, str] = {
+        path: contents for path, contents in updated_contents.items() if path not in declined_paths
+    }
     formatted: list[Path] = []
     changes: list[FormatChange] = []
     file_path: Path
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
index b462569bad..ab1521260c 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
@@ -686,3 +686,13 @@ class SqlReadabilityTestCase:
     description: str
     sql: str
     expected_fault: bool
+
+
+@dataclass(frozen=True)
+class FormatterSyntaxTestCase:
+    """Authored constructs that must survive canonical layout formatting."""
+
+    description: str
+    sql: str
+    expected_fragments: tuple[str, ...]
+    macro_source: str = ""
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
index b79b22f7d6..32ec863f5f 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
@@ -442,14 +442,14 @@ def test_given_unparseable_file_when_formatting_then_file_is_untouched_and_fault
     "test_case",
     [
         FormatterDeclineIntegrationTestCase(
-            description="unsupported SQL body reports a fault while header formats",
+            description="unsupported SQL body leaves the whole file unchanged",
             authored_body="select from\n",
             expected_exit_code=1,
         ),
     ],
     ids=lambda case: case.description,
 )
-def test_given_native_formatter_decline_when_formatting_then_header_formats_and_body_is_unchanged(
+def test_given_native_formatter_decline_when_formatting_then_whole_file_is_unchanged(
     test_case: FormatterDeclineIntegrationTestCase,
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],
@@ -468,6 +468,7 @@ def test_given_native_formatter_decline_when_formatting_then_header_formats_and_
         encoding="utf-8",
     )
 
+    original: str = model.read_text(encoding="utf-8")
     first_exit: int = main(["--project-dir", str(tmp_path), "format"])
     first_output: str = capsys.readouterr().out
     formatted_once: str = model.read_text(encoding="utf-8")
@@ -476,7 +477,7 @@ def test_given_native_formatter_decline_when_formatting_then_header_formats_and_
 
     assert first_exit == test_case.expected_exit_code
     assert second_exit == test_case.expected_exit_code
-    assert '  description "Orders.",\n' in formatted_once
+    assert formatted_once == original
     assert test_case.authored_body in formatted_once
     assert "format-unsafe" in first_output
     assert "format-unsafe" in second_output
@@ -553,7 +554,7 @@ def test_given_inline_typed_null_fixture_when_formatting_then_one_pass_is_idempo
     [
         CanonicalFixtureFormatIntegrationTestCase(
             description="post-native fixture simplification reaches fixed point in one pass",
-            expected_retained_literal="CAST(COLUMN2 AS TEXT) AS CLUSTER_ID",
+            expected_retained_literal="COLUMN2::VARCHAR AS CLUSTER_ID",
             expected_removed_literal="is_archived",
             expected_exit_code=0,
         )
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py
new file mode 100644
index 0000000000..376004b781
--- /dev/null
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py
@@ -0,0 +1,133 @@
+"""Canonical formatting retains authored syntax and compiler-scale query budgets."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from sqlbuild.cli.commands.main.entrypoint.entry import main
+from tests.integration.src.sqlbuild.cli.commands.main._test_types import FormatterSyntaxTestCase
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatterSyntaxTestCase(
+            "casts and qualified aliases",
+            "SELECT CAST(o.amount AS NUMERIC(10, 2)) AS amount, o.amount::DOUBLE AS total FROM orders o",
+            ("CAST(o.amount AS NUMERIC(10, 2))", "o.amount::DOUBLE"),
+        ),
+        FormatterSyntaxTestCase(
+            "variant paths and keyword keys",
+            "SELECT o.payload:customer.name::VARCHAR AS customer_name, o.payload:index::INTEGER AS item_index FROM orders o",
+            ("o.payload:customer.name::VARCHAR", "o.payload:index::INTEGER"),
+        ),
+        FormatterSyntaxTestCase(
+            "literal spelling",
+            r"SELECT $$Order summary$$ AS label, 'Customer''s order' AS other_label, '\\s+' AS pattern, 1.25e-5 AS ratio, TIMESTAMP '2025-01-01 00:00:00' AS created_at",
+            (
+                "$$Order summary$$",
+                "'Customer''s order'",
+                r"'\\s+'",
+                "1.25e-5",
+                "TIMESTAMP '2025-01-01 00:00:00'",
+            ),
+        ),
+        FormatterSyntaxTestCase(
+            "null treatment and function syntax",
+            "SELECT FIRST_VALUE(o.id IGNORE NULLS) OVER (ORDER BY o.id) AS first_id, POSITION('new' IN o.status) AS status_position, MOD(o.id, 2) AS remainder FROM orders o",
+            ("FIRST_VALUE(o.id IGNORE NULLS)", "POSITION('new' IN o.status)", "MOD(o.id, 2)"),
+        ),
+        FormatterSyntaxTestCase(
+            "typed lambda",
+            "SELECT TRANSFORM(o.items, item VARIANT -> item:order_id::INTEGER) AS order_ids FROM orders o",
+            ("item VARIANT ->", "item:order_id::INTEGER"),
+        ),
+        FormatterSyntaxTestCase(
+            "commented cast parameters",
+            "SELECT CAST(amount AS NUMERIC(/* precision */ 10, 2)) AS amount FROM orders",
+            ("/* precision */", "NUMERIC("),
+        ),
+        FormatterSyntaxTestCase(
+            "trailing projection comma",
+            "SELECT o.order_id, -- Keep the order identifier.\nFROM orders o",
+            ("-- Keep the order identifier.", "o.order_id"),
+        ),
+        FormatterSyntaxTestCase(
+            "compiler function depth",
+            "SELECT " + "LOWER(" * 80 + "'NEW'" + ")" * 80 + " AS status",
+            ("'NEW'", "AS status"),
+        ),
+        FormatterSyntaxTestCase(
+            "compiler set operation budget",
+            " UNION ALL ".join("SELECT 1 AS order_id" for _ in range(300)),
+            ("UNION ALL", "AS order_id"),
+        ),
+        FormatterSyntaxTestCase(
+            "CTE-producing macro",
+            "WITH @order_input(), final AS (SELECT order_id FROM orders) SELECT * FROM final",
+            ("@order_input()", "final AS ("),
+            'def order_input() -> str:\n    """Return an orders input CTE."""\n'
+            '    return "orders AS (SELECT 1 AS order_id)"\n',
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_authored_syntax_when_formatting_then_spelling_and_idempotence_are_preserved(
+    test_case: FormatterSyntaxTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "snowflake"\n')
+    macros: Path = tmp_path / "models" / "_macros"
+    macros.mkdir(parents=True)
+    (macros / "orders.py").write_text(test_case.macro_source)
+    model: Path = tmp_path / "models" / "order_summary.sql"
+    model.write_text(
+        'MODEL (description "Order summary", database warehouse, schema analytics);\n-- Order query.\n'
+        + test_case.sql
+        + "\n"
+    )
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 0
+    capsys.readouterr()
+    formatted: str = model.read_text()
+    assert all(fragment in formatted for fragment in test_case.expected_fragments)
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 0
+    assert model.read_text() == formatted
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatterSyntaxTestCase(
+            "computed variant key needs source-preserving parser support",
+            "SELECT payload:customers[TO_VARCHAR(order_id)] AS customer FROM orders",
+            ("format-unsafe", "Expected RBracket"),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_unsupported_variant_key_when_formatting_then_compilable_file_stays_untouched(
+    test_case: FormatterSyntaxTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "snowflake"\n')
+    model: Path = tmp_path / "models" / "order_summary.sql"
+    model.parent.mkdir()
+    original: str = (
+        'MODEL (description "Order summary", database warehouse, schema analytics);   \n'
+        + test_case.sql
+        + "\n"
+    )
+    model.write_text(original)
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+    capsys.readouterr()
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 1
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert all(fragment in json.dumps(payload) for fragment in test_case.expected_fragments)
+    assert payload["formatted_files"] == []
+    assert model.read_text() == original
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 1
+    assert model.read_text() == original
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])

From 57f5d5d8675642bef40178548b4b071677f32665 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 14:31:24 +0100
Subject: [PATCH 4/7] fix: recognize wrapped root CTEs and literal-only join
 predicates

---
 .../src/sql_lint/_helpers/engine.rs           |  8 +-
 .../src/sql_lint/_helpers/join_predicates.rs  |  3 +-
 .../src/sql_lint/_helpers/terminal_shape.rs   | 75 +++++++++++---
 docs/native-lint-rule-matrix.md               |  9 ++
 src/sqlbuild/lint/_helpers/native_format.py   |  7 +-
 .../sqlbuild/cli/commands/main/_test_types.py |  1 +
 .../sqlbuild/cli/commands/main/test_format.py |  2 +
 .../cli/commands/main/test_sql_readability.py | 98 ++++++++++++++++++-
 8 files changed, 184 insertions(+), 19 deletions(-)

diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
index 51a3e300ea..904148eebb 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs
@@ -232,7 +232,7 @@ const TRIVIAL_EQUALITY_TOKEN_COUNT: usize = 3;
 const JOIN_EXPRESSION: LintRuleMetadata = LintRuleMetadata {
     code: "SQBRSQL040",
     message: "JOIN ON contains a computed expression",
-    remediation: "Compute join keys in input CTEs and join on plain columns.",
+    remediation: "Compute keys, including variant path access, in input CTEs and join on plain columns. Literal-only predicates such as ON TRUE are allowed.",
 };
 const FINAL_CTE_NAME: LintRuleMetadata = LintRuleMetadata {
     code: "SQBRSQL041",
@@ -539,6 +539,8 @@ fn collect_rule_migration_facts(
         .filter(|&index| !is_layout(&tokens[index]) && !is_comment(&tokens[index]))
         .collect();
 
+    let root_with = crate::sql_lint::_helpers::terminal_shape::root_with_indices(tokens, depths);
+
     for (position, &index) in significant.iter().enumerate() {
         let text = tokens[index].text.as_str();
         if text.eq_ignore_ascii_case("cross")
@@ -548,11 +550,11 @@ fn collect_rule_migration_facts(
         {
             facts.cross_joins.push(tokens[index].span);
         }
-        if text.eq_ignore_ascii_case("with") {
+        if tokens[index].token_type == TokenType::With {
             let starts_hierarchy = position.checked_sub(1).is_some_and(|previous| {
                 tokens[significant[previous]].token_type == TokenType::Start
             });
-            if depths[index] > 0 && !starts_hierarchy {
+            if !root_with.contains(&index) && !starts_hierarchy {
                 facts.nested_ctes.push(tokens[index].span);
             }
             if !starts_hierarchy
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
index 960e76a070..48e7d2d56e 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/join_predicates.rs
@@ -81,11 +81,12 @@ fn operand(expression: &Expression) -> bool {
 }
 
 fn comparison(left: &Expression, right: &Expression) -> bool {
-    operand(left) && operand(right) && (plain_column(left) || plain_column(right))
+    operand(left) && operand(right)
 }
 
 fn plain_predicate(expression: &Expression) -> bool {
     match expression {
+        Expression::Boolean(_) => true,
         Expression::Paren(value) => plain_predicate(&value.this),
         Expression::And(value) | Expression::Or(value) => {
             plain_predicate(&value.left) && plain_predicate(&value.right)
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
index 103e6c5384..73637fa9e9 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/terminal_shape.rs
@@ -3,6 +3,7 @@ use polyglot_sql::tokens::{Span, Token, TokenType};
 const CEREMONIAL_SELECT_LITERAL: &str = "1";
 
 pub(super) fn final_cte_name_spans(tokens: &[Token], depths: &[usize]) -> Vec<Span> {
+    let roots = root_with_indices(tokens, depths);
     let mut names: Vec<(usize, bool)> = Vec::new();
     for (index, token) in tokens.iter().enumerate() {
         if token.token_type != TokenType::With {
@@ -36,7 +37,7 @@ pub(super) fn final_cte_name_spans(tokens: &[Token], depths: &[usize]) -> Vec<Sp
             }
             next = close + 2;
         }
-        if depth == 0
+        if roots.contains(&index)
             && names.len() > first
             && let Some((_, terminal)) = names.last_mut()
         {
@@ -52,6 +53,54 @@ pub(super) fn final_cte_name_spans(tokens: &[Token], depths: &[usize]) -> Vec<Sp
         .collect()
 }
 
+pub(super) fn root_with_indices(tokens: &[Token], depths: &[usize]) -> Vec<usize> {
+    let significant: Vec<usize> = (0..tokens.len())
+        .filter(|&index| {
+            !matches!(
+                tokens[index].token_type,
+                TokenType::Space
+                    | TokenType::Break
+                    | TokenType::LineComment
+                    | TokenType::BlockComment
+            )
+        })
+        .collect();
+    significant
+        .split(|&index| tokens[index].token_type == TokenType::Semicolon)
+        .filter_map(|statement| {
+            root_query_tokens(tokens, depths, statement)
+                .first()
+                .copied()
+        })
+        .filter(|&index| tokens[index].token_type == TokenType::With)
+        .collect()
+}
+
+fn root_query_tokens<'a>(
+    tokens: &[Token],
+    depths: &[usize],
+    mut significant: &'a [usize],
+) -> &'a [usize] {
+    while significant
+        .last()
+        .is_some_and(|&index| tokens[index].token_type == TokenType::Semicolon)
+    {
+        significant = &significant[..significant.len() - 1];
+    }
+    while let Some((&first, rest)) = significant.split_first() {
+        if tokens[first].token_type != TokenType::LParen
+            || rest
+                .last()
+                .is_none_or(|&index| tokens[index].token_type != TokenType::RParen)
+            || rest.iter().any(|&index| depths[index] <= depths[first])
+        {
+            break;
+        }
+        significant = &rest[..rest.len() - 1];
+    }
+    significant
+}
+
 fn cte_body_open(tokens: &[Token], depths: &[usize], name: usize) -> Option<usize> {
     let mut position = name + 1;
     if tokens.get(position)?.token_type == TokenType::LParen {
@@ -84,11 +133,16 @@ pub(super) fn collect_terminal_shape_facts(
 ) -> (Vec<Span>, Vec<Span>) {
     let mut cte_only_bodies: Vec<Span> = Vec::new();
     let mut terminal_selects: Vec<Span> = Vec::new();
-    let has_top_level_with = significant
-        .iter()
-        .any(|&index| depths[index] == 0 && tokens[index].text.eq_ignore_ascii_case("with"));
+    let significant = root_query_tokens(tokens, depths, significant);
+    let Some(&first) = significant.first() else {
+        return (cte_only_bodies, terminal_selects);
+    };
+    let root_depth = depths[first];
+    let has_top_level_with = significant.iter().any(|&index| {
+        depths[index] == root_depth && tokens[index].text.eq_ignore_ascii_case("with")
+    });
     let terminal_set_operator = significant.iter().find(|&&index| {
-        depths[index] == 0
+        depths[index] == root_depth
             && matches!(
                 tokens[index].token_type,
                 TokenType::Union | TokenType::Intersect | TokenType::Except
@@ -101,10 +155,9 @@ pub(super) fn collect_terminal_shape_facts(
         }
         return (cte_only_bodies, terminal_selects);
     }
-    let Some(root_select_position) = significant
-        .iter()
-        .position(|&index| depths[index] == 0 && tokens[index].token_type == TokenType::Select)
-    else {
+    let Some(root_select_position) = significant.iter().position(|&index| {
+        depths[index] == root_depth && tokens[index].token_type == TokenType::Select
+    }) else {
         return (cte_only_bodies, terminal_selects);
     };
     let root_select = significant[root_select_position];
@@ -112,7 +165,7 @@ pub(super) fn collect_terminal_shape_facts(
         .iter()
         .copied()
         .take_while(|&index| tokens[index].token_type != TokenType::Semicolon)
-        .filter(|&index| depths[index] == 0)
+        .filter(|&index| depths[index] == root_depth)
         .collect();
     let has_terminal_logic = tail.iter().any(|&index| {
         matches!(
@@ -140,7 +193,7 @@ pub(super) fn collect_terminal_shape_facts(
     let final_cte = (0..root_select_position)
         .filter(|&position| {
             let index = significant[position];
-            depths[index] == 0
+            depths[index] == root_depth
                 && tokens[index].token_type == TokenType::As
                 && significant
                     .get(position + 1)
diff --git a/docs/native-lint-rule-matrix.md b/docs/native-lint-rule-matrix.md
index 455b2cf0fb..dc06de8fcd 100644
--- a/docs/native-lint-rule-matrix.md
+++ b/docs/native-lint-rule-matrix.md
@@ -149,6 +149,10 @@ including joins inside CTEs and subqueries. `USING (...)` is accepted. Allowed p
 
 - Plain column comparisons, including equality, inequality, range and null-safe comparisons.
 - A plain column compared with a literal, including a signed numeric literal.
+- Literal-only predicates such as `ON TRUE`, `ON FALSE`, and `ON 1 = 1`, including
+  `LEFT JOIN LATERAL FLATTEN(...) AS f ON TRUE`. Functions or casts applied to literals are
+  still computed expressions. Other rules, such as `SQBRSQL003`, independently govern
+  placeholder join conditions; this allowance is specific to `SQBRSQL040`.
 - `BETWEEN` / `NOT BETWEEN` on a plain column with plain-column or literal bounds.
 - `IN` / `NOT IN` on a plain column with a literal list.
 - `IS NULL` / `IS NOT NULL` on a plain column.
@@ -158,6 +162,8 @@ Functions, casts (including `::`), arithmetic, concatenation, CASE, subqueries a
 operands are findings. For example, replace `ON LOWER(o.category) = c.category_key` with a CTE
 that projects `LOWER(category) AS category_key`, then `ON o.category_key = c.category_key`.
 An `OR` with only plain predicates passes; an `OR` containing a computed branch fails.
+Snowflake variant path access is a computed operand: `ON a.data:key = b.id` is flagged.
+Project that value as a named input column before joining.
 
 `SQBRSQL041` checks naming, while `SQBRSQL035` owns the terminal SELECT's plain projection and
 reference to the last top-level CTE. Select both (normally via `SQBRSQL`) for the complete
@@ -165,6 +171,9 @@ convention. The last top-level CTE must be named `final`, compared case-insensit
 existing SQL rules, including quoted identifiers. Every other CTE, including a nested one, must
 use a different name. Queries without CTEs are unaffected. SQL-test fixture CTEs and their
 ceremonial `SELECT 1` remain exempt from this model naming convention.
+Parentheses enclosing the entire query, including multiple wrapper pairs, do not create a nested
+CTE scope. CTEs inside scalar subqueries or CTE bodies remain nested. `SQBRSQL035` and
+`SQBRSQL036` use the same root-wrapper distinction.
 
 Prefer `[rules] select = ["SQBRSQL", "SQBRMODEL", "SQBRGRAPH"]` to exact-code lists. New rules
 in those families apply automatically after an upgrade, so adopting this release can expose
diff --git a/src/sqlbuild/lint/_helpers/native_format.py b/src/sqlbuild/lint/_helpers/native_format.py
index c6b046c277..4c3a5820ac 100644
--- a/src/sqlbuild/lint/_helpers/native_format.py
+++ b/src/sqlbuild/lint/_helpers/native_format.py
@@ -149,10 +149,13 @@ def _format_fault(*, file_path: Path, contents: str, start: int, reason: str) ->
         line=contents.count(LINE_FEED, 0, start) + 1,
         column=1,
         code=RULE_FORMAT_UNSAFE,
-        message=f"SQL body was not formatted: {reason}",
+        message=f"File left unchanged because its SQL body could not be formatted: {reason}",
         severity=VIOLATION_SEVERITY_FAULT,
         engine=LINT_ENGINE_NATIVE,
-        remediation="Rewrite the unsupported SQL shape or report a minimal formatter reproduction.",
+        remediation=(
+            "Rewrite the unsupported SQL shape or report a minimal formatter reproduction. "
+            "No header, SQL body, or fixture changes were written to this file."
+        ),
     )
 
 
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
index ab1521260c..2f4593e01b 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py
@@ -686,6 +686,7 @@ class SqlReadabilityTestCase:
     description: str
     sql: str
     expected_fault: bool
+    rule_code: str = "SQBRSQL041"
 
 
 @dataclass(frozen=True)
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
index 32ec863f5f..41c77c9aee 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
@@ -481,11 +481,13 @@ def test_given_native_formatter_decline_when_formatting_then_whole_file_is_uncha
     assert test_case.authored_body in formatted_once
     assert "format-unsafe" in first_output
     assert "format-unsafe" in second_output
+    assert "File left unchanged" in first_output
     assert "orders.sql" in second_output
     json_exit: int = main(["--project-dir", str(tmp_path), "format", "--check", "--json"])
     json_output: str = capsys.readouterr().out
     assert json_exit == 1
     assert "format-unsafe" in json_output
+    assert "File left unchanged" in json_output
     assert json.loads(json_output)
     assert formatted_once == model.read_text(encoding="utf-8")
 
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py b/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py
index b861041aa9..801da1dec7 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_sql_readability.py
@@ -82,6 +82,70 @@ def test_given_join_predicate_when_compiling_and_running_rules_then_expression_p
     "test_case",
     [
         SqlReadabilityTestCase("no CTE", "SELECT 1 AS order_id", False),
+        SqlReadabilityTestCase(
+            "wrapped root", "(WITH final AS (SELECT 1 AS order_id) SELECT * FROM final)", False
+        ),
+        SqlReadabilityTestCase(
+            "double wrapped root",
+            "((WITH final AS (SELECT 1 AS order_id) SELECT * FROM final))",
+            False,
+        ),
+        SqlReadabilityTestCase(
+            "wrapped wrong name",
+            "((WITH orders AS (SELECT 1 AS order_id) SELECT * FROM orders))",
+            True,
+        ),
+        SqlReadabilityTestCase(
+            "wrapped CTE union",
+            "((WITH final AS (SELECT 1 AS order_id UNION ALL SELECT 2 AS order_id) SELECT * FROM final))",
+            False,
+        ),
+        SqlReadabilityTestCase(
+            "wrapped terminal union naming",
+            "(WITH final AS (SELECT 1 AS order_id) SELECT * FROM final UNION ALL SELECT * FROM final)",
+            False,
+        ),
+        SqlReadabilityTestCase(
+            "nested scalar final",
+            "SELECT (WITH final AS (SELECT 1 AS order_id) SELECT order_id FROM final) AS order_id",
+            True,
+        ),
+        SqlReadabilityTestCase(
+            "wrapped plain terminal",
+            "((WITH final AS (SELECT 1 AS order_id) SELECT * FROM final))",
+            False,
+            "SQBRSQL035",
+        ),
+        SqlReadabilityTestCase(
+            "wrapped computed terminal",
+            "((WITH final AS (SELECT 1 AS order_id) SELECT order_id + 1 AS order_id FROM final))",
+            True,
+            "SQBRSQL035",
+        ),
+        SqlReadabilityTestCase(
+            "wrapped terminal union shape",
+            "(WITH final AS (SELECT 1 AS order_id) SELECT * FROM final UNION ALL SELECT * FROM final)",
+            True,
+            "SQBRSQL035",
+        ),
+        SqlReadabilityTestCase(
+            "wrapped CTE union terminal shape",
+            "((WITH final AS (SELECT 1 AS order_id UNION ALL SELECT 2 AS order_id) SELECT * FROM final))",
+            False,
+            "SQBRSQL035",
+        ),
+        SqlReadabilityTestCase(
+            "wrapped root is not nested",
+            "((WITH final AS (SELECT 1 AS order_id) SELECT * FROM final))",
+            False,
+            "SQBRSQL036",
+        ),
+        SqlReadabilityTestCase(
+            "scalar CTE is nested",
+            "SELECT (WITH orders AS (SELECT 1 AS order_id) SELECT order_id FROM orders) AS order_id",
+            True,
+            "SQBRSQL036",
+        ),
         SqlReadabilityTestCase(
             "final", "WITH final AS (SELECT 1 AS order_id) SELECT * FROM final", False
         ),
@@ -123,11 +187,13 @@ def test_given_cte_names_when_compiling_and_running_rules_then_final_name_policy
     model.write_text(f'MODEL (description "Orders");\n{test_case.sql}\n')
     assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
     capsys.readouterr()
-    result: int = main(["--project-dir", str(tmp_path), "rules", "--json", "run", "SQBRSQL041"])
+    result: int = main(
+        ["--project-dir", str(tmp_path), "rules", "--json", "run", test_case.rule_code]
+    )
     payload: dict[str, object] = json.loads(capsys.readouterr().out)
     assert result == int(test_case.expected_fault)
     assert bool(payload["findings"]) == test_case.expected_fault
-    assert ("SQBRSQL041" in json.dumps(payload["findings"])) == test_case.expected_fault
+    assert (test_case.rule_code in json.dumps(payload["findings"])) == test_case.expected_fault
 
 
 @pytest.mark.parametrize(
@@ -163,5 +229,33 @@ def test_given_family_selection_when_compiling_then_new_rules_are_enforced(
     assert "SQBRSQL041" in json.dumps(payload["findings"])
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SqlReadabilityTestCase("lateral true", "TRUE", False),
+        SqlReadabilityTestCase("lateral false", "FALSE", False),
+        SqlReadabilityTestCase("literal equality", "1 = 1", False),
+        SqlReadabilityTestCase("computed constant", "CAST(TRUE AS BOOLEAN)", True),
+        SqlReadabilityTestCase("variant key", "o.data:key = f.value", True),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_lateral_join_when_running_expression_rule_then_literals_pass_and_variant_keys_fail(
+    test_case: SqlReadabilityTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "snowflake"\n')
+    model: Path = tmp_path / "models" / "orders.sql"
+    model.parent.mkdir()
+    model.write_text(
+        'MODEL (description "Orders", database warehouse, schema analytics);\n'
+        "SELECT o.order_id FROM raw_orders AS o LEFT JOIN LATERAL FLATTEN(input => o.items) AS f "
+        f"ON {test_case.sql}\n"
+    )
+    result: int = main(["--project-dir", str(tmp_path), "rules", "--json", "run", "SQBRSQL040"])
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert result == int(test_case.expected_fault)
+    assert ("SQBRSQL040" in json.dumps(payload["findings"])) == test_case.expected_fault
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])

From 974654072d31a85d19cbb91876de567beabd7d5a Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 14:47:13 +0100
Subject: [PATCH 5/7] fix: preserve interpolation spelling and separate macro
 CTE items

---
 README.md                                     |   6 +-
 .../src/sql_lint/_helpers/formatter.rs        |  47 +++++++-
 .../src/sql_lint/_helpers/formatter_syntax.rs |   2 +-
 src/sqlbuild/lint/_helpers/sqlbuild_tokens.py |  17 ++-
 .../sqlbuild/cli/commands/main/test_format.py |   2 +-
 .../commands/main/test_format_preservation.py | 110 ++++++++++++++++++
 6 files changed, 175 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index cee15a6433..42c8441733 100644
--- a/README.md
+++ b/README.md
@@ -210,8 +210,10 @@ formatting and `sqb format --check` exit nonzero. A declined body is never count
 Formatting preserves authored cast types, postfix casts, quoted literals, variant paths, typed
 lambda parameters, and supported SQL function spellings while applying canonical layout. It uses
 the compiler's trusted-SQL function-depth budget and does not impose the separate browser-oriented
-UNION-chain limit from Polyglot's convenience formatting API. CTE-producing macros are formatted as
-authored macro calls rather than expanded project SQL.
+UNION-chain limit from Polyglot's convenience formatting API. SQLBuild calls retain their authored
+spelling, including zero-argument cursor and empty-fixture intrinsics. CTE-producing macros remain
+authored calls rather than expanded project SQL, with each call on its own CTE-list line and
+leading comments attached to the node they describe.
 
 Custom rules are ordinary Python beneath `rules/**/*.py`. Only `@rule` functions register; helper
 functions, constants, dataclasses, classes, and nested packages remain ordinary Python. Typed,
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
index 5f93cdf520..eec18b2813 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter.rs
@@ -5,7 +5,7 @@ use polyglot_sql::tokens::{Token, TokenType};
 use polyglot_sql::{ComplexityGuardOptions, Dialect, DialectType, ParseOptions};
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
 
-use crate::sql_lint::_helpers::formatter_syntax::protect_syntax;
+use crate::sql_lint::_helpers::formatter_syntax::{cte_macro_indices, protect_syntax};
 use crate::sql_lint::constants::{
     CAST_TYPE_SEPARATOR_KEYWORD, CLOSE_PARENTHESIS, LINT_API_VERSION, OPEN_PARENTHESIS,
     VALUES_RELATION_PREFIX_TOKEN_COUNT,
@@ -151,6 +151,7 @@ fn format_once(neutral_sql: &str, context: &FormatOnceContext<'_>) -> Result<Str
     let mut formatted = protected.restore(formatted_statements?.join(";\n"), context.dialect)?;
     formatted = restore_unparenthesized_from_values(formatted, context)?;
     formatted = restore_null_treatment(formatted, context)?;
+    formatted = layout_cte_macros(formatted, context)?;
     if context.comments.is_empty() {
         return Ok(formatted);
     }
@@ -238,6 +239,50 @@ fn restore_implicit_aliases(
     Ok(formatted)
 }
 
+fn layout_cte_macros(
+    mut formatted: String,
+    context: &FormatOnceContext<'_>,
+) -> Result<String, String> {
+    let tokens = context
+        .dialect
+        .tokenize(&formatted)
+        .map_err(|error| error.to_string())?;
+    let macros = cte_macro_indices(&tokens);
+    if macros.is_empty() {
+        return Ok(formatted);
+    }
+    let depths = crate::sql_lint::_helpers::engine::token_depths(&tokens);
+    let chars: Vec<char> = formatted.chars().collect();
+    let mut gaps: BTreeMap<(usize, usize), String> = BTreeMap::new();
+    for index in macros {
+        let with = (0..index)
+            .rev()
+            .find(|&previous| {
+                tokens[previous].token_type == TokenType::With && depths[previous] == depths[index]
+            })
+            .ok_or_else(|| "native formatter lost a CTE macro scope".to_string())?;
+        let indentation = line_indentation(&chars, tokens[with].span.start);
+        let newline = format!("\n{}", " ".repeat(indentation));
+        gaps.insert(
+            (tokens[index - 1].span.end, tokens[index].span.start),
+            newline.clone(),
+        );
+        if tokens
+            .get(index + 1)
+            .is_some_and(|token| token.token_type == TokenType::Comma)
+            && let Some(next) = tokens.get(index + 2)
+        {
+            gaps.insert((tokens[index + 1].span.end, next.span.start), newline);
+        }
+    }
+    for ((start, end), replacement) in gaps.into_iter().rev() {
+        let start = char_to_byte(&formatted, start)?;
+        let end = char_to_byte(&formatted, end)?;
+        formatted.replace_range(start..end, &replacement);
+    }
+    Ok(formatted)
+}
+
 fn restore_null_treatment(
     mut formatted: String,
     context: &FormatOnceContext<'_>,
diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
index 91c3266f24..4c56ca9fca 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
@@ -255,7 +255,7 @@ fn canonical_fragment(chars: &[char], tokens: &[Token]) -> String {
     result
 }
 
-fn cte_macro_indices(tokens: &[Token]) -> HashSet<usize> {
+pub(super) fn cte_macro_indices(tokens: &[Token]) -> HashSet<usize> {
     let depths = token_depths(tokens);
     let mut scopes: HashSet<usize> = HashSet::new();
     let mut indices: HashSet<usize> = HashSet::new();
diff --git a/src/sqlbuild/lint/_helpers/sqlbuild_tokens.py b/src/sqlbuild/lint/_helpers/sqlbuild_tokens.py
index 81adcbbc2b..4cbe45af07 100644
--- a/src/sqlbuild/lint/_helpers/sqlbuild_tokens.py
+++ b/src/sqlbuild/lint/_helpers/sqlbuild_tokens.py
@@ -31,6 +31,9 @@
 from sqlbuild.lint.models import InterpolationSite
 
 _SQLBUILD_FUNCTION_NAMES: tuple[str, ...] = (
+    "__cursor_start",
+    "__cursor_end",
+    "__empty_fixture",
     "__dbt_ref",
     "__table_fn",
     "__source",
@@ -55,10 +58,10 @@
 _BACKTICK_SCAN_PATTERN: str = r"|`(?:``|[^`])*(?:`|\Z)"
 _SITE_SCAN_PATTERN: str = rf"|(?P<site>@@|\$\{{|@|(?:{_SQLBUILD_FUNCTION_PATTERN})\s*\()"
 _INTERPOLATION_SCAN_PATTERN: re.Pattern[str] = re.compile(
-    _NON_CODE_SCAN_PATTERN + _SITE_SCAN_PATTERN
+    _NON_CODE_SCAN_PATTERN + _SITE_SCAN_PATTERN, re.IGNORECASE
 )
 _BACKTICK_INTERPOLATION_SCAN_PATTERN: re.Pattern[str] = re.compile(
-    _NON_CODE_SCAN_PATTERN + _BACKTICK_SCAN_PATTERN + _SITE_SCAN_PATTERN
+    _NON_CODE_SCAN_PATTERN + _BACKTICK_SCAN_PATTERN + _SITE_SCAN_PATTERN, re.IGNORECASE
 )
 
 
@@ -190,7 +193,7 @@ def _contains_interpolation_candidate(*, body: str) -> bool:
     return (
         MACRO_TOKEN in body
         or TEMPLATE_INTERPOLATION_START in body
-        or any(name in body for name in _SQLBUILD_FUNCTION_NAMES)
+        or any(name in body.lower() for name in _SQLBUILD_FUNCTION_NAMES)
     )
 
 
@@ -251,7 +254,11 @@ def _interpolation_site_end(*, body: str, start: int, backtick_identifiers: bool
         return _macro_site_end(body=body, start=start, backtick_identifiers=backtick_identifiers)
     if character == TEMPLATE_INTERPOLATION_START[0]:
         return _template_site_end(body=body, start=start)
-    if character == IDENTIFIER_EXTRA_CHARACTER and body.startswith(_SQLBUILD_FUNCTION_NAMES, start):
+    if (
+        character == IDENTIFIER_EXTRA_CHARACTER
+        and body[start : _identifier_end(body=body, start=start)].lower()
+        in _SQLBUILD_FUNCTION_NAMES
+    ):
         return _sqlbuild_function_site_end(
             body=body, start=start, backtick_identifiers=backtick_identifiers
         )
@@ -260,6 +267,8 @@ def _interpolation_site_end(*, body: str, start: int, backtick_identifiers: bool
 
 def _sqlbuild_function_site_end(*, body: str, start: int, backtick_identifiers: bool) -> int | None:
     name_end: int = _identifier_end(body=body, start=start)
+    while name_end < len(body) and body[name_end].isspace():
+        name_end += 1
     if name_end >= len(body) or body[name_end] != OPENING_PAREN_CHARACTER:
         return None
     call_end: int | None = _matching_paren_end(
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
index 41c77c9aee..bd64526e01 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format.py
@@ -498,7 +498,7 @@ def test_given_native_formatter_decline_when_formatting_then_whole_file_is_uncha
         TypedNullFormatIntegrationTestCase(
             description="inline colon casts reach empty fixture fixed point in one pass",
             fixture_projection=("NULL::VARCHAR AS customer_key, NULL::BIGINT AS order_count"),
-            expected_literal="__EMPTY_FIXTURE()",
+            expected_literal="__empty_fixture()",
             expected_exit_code=0,
         )
     ],
diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py b/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py
index 376004b781..f6b1d7e8ac 100644
--- a/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py
+++ b/tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py
@@ -129,5 +129,115 @@ def test_given_unsupported_variant_key_when_formatting_then_compilable_file_stay
     assert model.read_text() == original
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatterSyntaxTestCase(
+            "model reference", 'select order_id from __ref("orders")', ('__ref("orders")',)
+        ),
+        FormatterSyntaxTestCase(
+            "source reference",
+            'select order_id from __source("raw.orders")',
+            ('__source("raw.orders")',),
+        ),
+        FormatterSyntaxTestCase(
+            "seed reference", 'select order_id from __seed("orders")', ('__seed("orders")',)
+        ),
+        FormatterSyntaxTestCase(
+            "dbt reference", 'select order_id from __dbt_ref("orders")', ('__dbt_ref("orders")',)
+        ),
+        FormatterSyntaxTestCase(
+            "table function",
+            'select * from __table_fn("order_items")(1)',
+            ('__table_fn("order_items")(1)',),
+        ),
+        FormatterSyntaxTestCase(
+            "scalar function",
+            'select __udf("order_label")(1) as label',
+            ('__udf("order_label")(1)',),
+        ),
+        FormatterSyntaxTestCase(
+            "cursor start", "select __cursor_start() as start_at", ("__cursor_start()",)
+        ),
+        FormatterSyntaxTestCase(
+            "cursor end", "select __cursor_end() as end_at", ("__cursor_end()",)
+        ),
+        FormatterSyntaxTestCase(
+            "empty fixture", "select * from __empty_fixture()", ("__empty_fixture()",)
+        ),
+        FormatterSyntaxTestCase(
+            "authored case and spacing",
+            "select __CuRsOr_StArT () as start_at",
+            ("__CuRsOr_StArT ()",),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_sqlbuild_call_when_formatting_then_authored_spelling_is_preserved(
+    test_case: FormatterSyntaxTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "duckdb"\n')
+    model: Path = tmp_path / "models" / "order_summary.sql"
+    model.parent.mkdir()
+    model.write_text('MODEL (description "Order summary");\n' + test_case.sql + "\n")
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 0
+    capsys.readouterr()
+    formatted: str = model.read_text()
+    assert all(fragment in formatted for fragment in test_case.expected_fragments)
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 0
+    assert model.read_text() == formatted
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatterSyntaxTestCase(
+            "several CTE macros with attached comments",
+            "WITH base AS (SELECT 1 AS order_id),\n"
+            '-- First order stage.\n@make_ctes("first_orders", "base"),\n'
+            '-- Second order stage.\n@make_ctes("second_orders", "first_orders"),\n'
+            '@other_ctes("third_orders", "second_orders"),\n'
+            "-- Publish order identifiers.\nfinal AS (SELECT order_id FROM third_orders)\nSELECT * FROM final",
+            (
+                '\n@make_ctes("first_orders", "base"),\n',
+                '\n@make_ctes("second_orders", "first_orders"),\n',
+                '\n@other_ctes("third_orders", "second_orders"),\n',
+                "\n-- Publish order identifiers.\nfinal AS (",
+            ),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_cte_macros_and_comments_when_formatting_then_layout_and_rules_stay_valid(
+    test_case: FormatterSyntaxTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(
+        'name = "orders"\nadapter = "duckdb"\n[rules]\nselect = ["SQBRSQL033"]\n'
+    )
+    macros: Path = tmp_path / "models" / "_macros"
+    macros.mkdir(parents=True)
+    (macros / "ctes.py").write_text(
+        "def make_ctes(name: str, source: str) -> str:\n"
+        '    return f"{name} AS (SELECT order_id FROM {source})"\n\n'
+        "def other_ctes(name: str, source: str) -> str:\n"
+        "    return make_ctes(name, source)\n"
+    )
+    model: Path = tmp_path / "models" / "order_summary.sql"
+    model.write_text('MODEL (description "Order summary");\n' + test_case.sql + "\n")
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+    capsys.readouterr()
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 0
+    capsys.readouterr()
+    formatted: str = model.read_text()
+    assert all(fragment in formatted for fragment in test_case.expected_fragments)
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 0
+    assert model.read_text() == formatted
+    assert main(["--project-dir", str(tmp_path), "compile", "--no-cache"]) == 0
+    capsys.readouterr()
+    assert main(["--project-dir", str(tmp_path), "rules", "--json", "run", "SQBRSQL033"]) == 0
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert payload["findings"] == []
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])

From 954350fd6b080b0b3d0a6a465207d24041ff9ab7 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 15:27:02 +0100
Subject: [PATCH 6/7] fix: merge identical formatter syntax branches

---
 .../src/sql_lint/_helpers/formatter_syntax.rs   | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
index 4c56ca9fca..a6863ca386 100644
--- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
+++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/formatter_syntax.rs
@@ -145,15 +145,14 @@ pub(super) fn protect_syntax(
             })
         {
             (index + 1, format!(".{sentinel}"))
-        } else if is_typed_literal(tokens, index) {
-            (index + 1, sentinel)
-        } else if matches!(token.token_type, TokenType::Var | TokenType::Identifier)
-            && tokens.get(index + 1).is_some_and(|next| {
-                matches!(next.token_type, TokenType::Var | TokenType::Identifier)
-            })
-            && tokens
-                .get(index + 2)
-                .is_some_and(|next| next.token_type == TokenType::Arrow)
+        } else if is_typed_literal(tokens, index)
+            || (matches!(token.token_type, TokenType::Var | TokenType::Identifier)
+                && tokens.get(index + 1).is_some_and(|next| {
+                    matches!(next.token_type, TokenType::Var | TokenType::Identifier)
+                })
+                && tokens
+                    .get(index + 2)
+                    .is_some_and(|next| next.token_type == TokenType::Arrow))
         {
             (index + 1, sentinel)
         } else if matches!(token.text.to_ascii_uppercase().as_str(), "MOD" | "POSITION")

From e25effdb05a1ebe3b845e4f06a4edc6e0618d452 Mon Sep 17 00:00:00 2001
From: Kevin Longe <273747162+kvlonge@users.noreply.github.com>
Date: Fri, 25 Sep 2026 15:43:50 +0100
Subject: [PATCH 7/7] test: expect authored empty-fixture spelling after
 formatting

---
 tests/unit/src/sqlbuild/lint/test_cli_format.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/unit/src/sqlbuild/lint/test_cli_format.py b/tests/unit/src/sqlbuild/lint/test_cli_format.py
index 7f6c0b9d4a..782d03465f 100644
--- a/tests/unit/src/sqlbuild/lint/test_cli_format.py
+++ b/tests/unit/src/sqlbuild/lint/test_cli_format.py
@@ -174,7 +174,7 @@ def test_given_redundant_typed_fixture_null_when_formatting_then_safe_fix_uses_c
 
 @pytest.mark.parametrize(
     "test_case",
-    [FormatCliTestCase("all-null empty fixture", (), 0, "__EMPTY_FIXTURE()")],
+    [FormatCliTestCase("all-null empty fixture", (), 0, "__empty_fixture()")],
     ids=lambda case: case.description,
 )
 def test_given_all_null_empty_fixture_when_formatting_then_uses_empty_fixture_marker(
@@ -219,7 +219,7 @@ def test_given_all_null_empty_fixture_when_formatting_then_uses_empty_fixture_ma
     assert "Formatted files:" in capsys.readouterr().out
     formatted: str = test_file.read_text(encoding="utf-8")
     assert test_case.expected_fragment in formatted
-    assert "__ref__orders AS (\n  SELECT\n    *\n  FROM __EMPTY_FIXTURE()\n)" in formatted
+    assert "__ref__orders AS (\n  SELECT\n    *\n  FROM __empty_fixture()\n)" in formatted
     assert "CAST(NULL" not in formatted.split("__expected__orders", maxsplit=1)[0]