@@ -552,6 +552,7 @@ pub(crate) fn decode_outcome(
             binding_validated,
         },
         cleaned_sql: String::new(),
+        deferred_cleaning: None,
         validated_schema: schema.clone(),
         fused_binding_validated: true,
     };
 
@@ -31,9 +31,9 @@ use crate::assembly::analysis_session::_helpers::mappings::{
 use crate::assembly::analysis_session::_helpers::mappings::{producers, waves};
 use crate::assembly::analysis_session::constants::{UNKNOWN_NULLABILITY, UNKNOWN_TYPE};
 use crate::assembly::analysis_session::models::{
-    AnalysisSession, ColumnFact, ContractProof, FinishedSession, LineageFacts, LineageRow,
-    ModelAnalysis, ModelOutcome, ModelRequest, Phase, PivotOutcome, PivotTables, SessionModelFacts,
-    SessionOutcome, SessionRequest, SessionStep,
+    AnalysisSession, ColumnFact, ContractProof, DeferredCleaning, FinishedSession, LineageFacts,
+    LineageRow, ModelAnalysis, ModelOutcome, ModelRequest, Phase, PivotOutcome, PivotTables,
+    SessionModelFacts, SessionOutcome, SessionRequest, SessionStep,
 };
 use crate::assembly::analysis_session::types::{Pairs, Shapes};
 use crate::semantic_validation::models::ProjectCatalog;
@@ -63,6 +63,7 @@ impl AnalysisSession {
             available_nullability: ShapeTable::from_shapes(&request.column_nullability),
             complete_shapes,
             outcomes: vec![None; request.models.len()],
+            pending_cleaned: vec![false; request.models.len()],
             waves: if dependency_ordered || request.models.is_empty() {
                 waves
             } else {
@@ -112,13 +113,26 @@ impl AnalysisSession {
         }
         let dynamic_contracts: Vec<PivotOutcome> = self.dynamic_contracts()?;
         let pool: Result<Arc<ThreadPool>, String> = self.catalog.native.analysis_pool();
-        let models: Vec<ModelOutcome> = self
+        let mut models: Vec<ModelOutcome> = self
             .outcomes
             .into_iter()
             .collect::<Option<_>>()
             .ok_or("the session left a model unanalysed")?;
         let (schema_additions, analysis_names) = self.catalog.into_changes();
-        let request: SessionRequest = self.request;
+        let mut request: SessionRequest = self.request;
+        for ((outcome, model), pending) in models
+            .iter_mut()
+            .zip(request.models.iter_mut())
+            .zip(self.pending_cleaned)
+        {
+            if pending {
+                outcome.deferred_cleaning = Some(DeferredCleaning {
+                    dialect: request.dialect.clone(),
+                    query_sql: std::mem::take(&mut model.query_sql),
+                    placeholders: std::mem::take(&mut model.placeholders),
+                });
+            }
+        }
         let mut model_facts: HashMap<String, SessionModelFacts> = HashMap::new();
         let mut dynamic_proofs: HashMap<String, ContractProof> = HashMap::new();
         for ((model, outcome), proof) in request.models.iter().zip(&models).zip(&dynamic_contracts)
@@ -247,6 +261,7 @@ impl AnalysisSession {
             self.outcomes[*model] = Some(ModelOutcome {
                 analysis,
                 cleaned_sql: result.cleaned_sql,
+                deferred_cleaning: None,
                 validated_schema: schema,
                 fused_binding_validated: true,
             });
@@ -394,8 +409,8 @@ impl AnalysisSession {
     /// Python's deferred binding validation, then the dataflow's shape publication.
     fn finish_wave(&mut self, wave: usize) -> Result<(), String> {
         let models: Vec<usize> = self.waves[wave].clone();
-        let mut validations: Vec<(usize, String, Shapes)> = Vec::new();
         let validated: &[usize] = if self.cyclic { &[] } else { &models };
+        let mut revalidated: Vec<(usize, Shapes)> = Vec::new();
         for model in validated {
             let schema: Shapes = self.binding_schema(&self.request.models[*model]);
             let outcome: &ModelOutcome = self.outcomes[*model]
@@ -404,9 +419,18 @@ impl AnalysisSession {
             if !outcome.fused_binding_validated
                 || !same_relations(&outcome.validated_schema, &schema)
             {
-                validations.push((*model, outcome.cleaned_sql.clone(), schema));
+                revalidated.push((*model, schema));
             }
         }
+        let pending: Vec<usize> = revalidated.iter().map(|(model, _)| *model).collect();
+        self.ensure_cleaned(&pending)?;
+        let mut validations: Vec<(usize, String, Shapes)> = Vec::new();
+        for (model, schema) in revalidated {
+            let outcome: &ModelOutcome = self.outcomes[model]
+                .as_ref()
+                .ok_or("a wave model was not analysed")?;
+            validations.push((model, outcome.cleaned_sql.clone(), schema));
+        }
         if !validations.is_empty() {
             let requests: Vec<(&str, &Shapes)> = validations
                 .iter()
@@ -421,6 +445,9 @@ impl AnalysisSession {
             }
         }
         if self.dependency_ordered {
+            if self.request.case_sensitive_shapes {
+                self.ensure_cleaned(&models)?;
+            }
             for model in &models {
                 self.publish(*model)?;
             }
@@ -500,26 +527,11 @@ impl AnalysisSession {
                 None => misses.push(position),
             }
         }
-        let members: Vec<BatchMember<'_>> = hits
-            .iter()
-            .map(|(position, _)| {
-                batch_member(&self.request.models[models[*position]], &schemas[*position])
-            })
-            .collect();
-        let cleaned: Vec<Result<String, String>> = self
-            .catalog
-            .normalize_each(&self.request.dialect, &members)
-            .unwrap_or_else(|error| members.iter().map(|_| Err(error.clone())).collect());
-        for ((position, mut outcome), sql) in hits.into_iter().zip(cleaned) {
+        for (position, outcome) in hits {
             let model: usize = models[position];
-            match sql {
-                Ok(sql) => {
-                    outcome.cleaned_sql = sql;
-                    self.outcomes[model] = Some(outcome);
-                    cache.hits[model] = true;
-                }
-                Err(_) => misses.push(position),
-            }
+            self.outcomes[model] = Some(outcome);
+            self.pending_cleaned[model] = true;
+            cache.hits[model] = true;
         }
         misses.sort_unstable();
         cache.stats.hits += models.len() - misses.len();
@@ -528,6 +540,38 @@ impl AnalysisSession {
         Ok(misses)
     }
 
+    /// Normalize the cleaned SQL of the cache hits among `models` that still lack it.
+    fn ensure_cleaned(&mut self, models: &[usize]) -> Result<(), String> {
+        let pending: Vec<usize> = models
+            .iter()
+            .copied()
+            .filter(|model| self.pending_cleaned[*model])
+            .collect();
+        if pending.is_empty() {
+            return Ok(());
+        }
+        let schemas: Vec<Shapes> = pending
+            .iter()
+            .map(|model| self.binding_schema(&self.request.models[*model]))
+            .collect();
+        let members: Vec<BatchMember<'_>> = pending
+            .iter()
+            .zip(&schemas)
+            .map(|(model, schema)| batch_member(&self.request.models[*model], schema))
+            .collect();
+        let cleaned: Vec<Result<String, String>> = self
+            .catalog
+            .normalize_each(&self.request.dialect, &members)?;
+        for (model, sql) in pending.into_iter().zip(cleaned) {
+            let sql: String = sql.map_err(|reason| {
+                native_failure(&model_context(&self.request.models[model]), &reason)
+            })?;
+            self.outcome_mut(model)?.cleaned_sql = sql;
+            self.pending_cleaned[model] = false;
+        }
+        Ok(())
+    }
+
     /// Store each finished native outcome of `models` the cache did not answer.
     fn store_cached(&mut self, models: &[usize]) {
         let Some(cache) = self.cache.as_mut() else {
 
@@ -10,6 +10,7 @@ pub mod prove_dynamic_contracts;
 pub mod prove_finished_dynamic_contracts;
 pub mod query_columns;
 pub mod recover_cte_facts;
+pub mod resolved_cleaned_sql;
 pub mod run_analysis_session;
 pub mod session_sharing;
 pub mod sql_test_helper_reads;
 
@@ -0,0 +1,29 @@
+//! A finished model outcome's cleaned analysis SQL, normalized on first read for cache hits.
+
+use std::borrow::Cow;
+use std::collections::HashMap;
+
+use sqlbuild_core::panics::main::catch_compiler_panic::catch_compiler_panic;
+
+use crate::assembly::analysis_session::models::{DeferredCleaning, ModelOutcome};
+use crate::semantic_validation::main::normalize::normalize_analysis_sql;
+use crate::semantic_validation::models::NormalizationInput;
+
+/// The outcome's cleaned SQL, normalizing a cache hit's deferred input when it has one.
+pub fn resolved_cleaned_sql(outcome: &ModelOutcome) -> Result<Cow<'_, str>, String> {
+    let Some(DeferredCleaning {
+        dialect,
+        query_sql,
+        placeholders,
+    }) = outcome.deferred_cleaning.as_ref()
+    else {
+        return Ok(Cow::Borrowed(&outcome.cleaned_sql));
+    };
+    let request: NormalizationInput = NormalizationInput {
+        sql: query_sql.clone(),
+        dialect: dialect.clone(),
+        stubs: HashMap::new(),
+        placeholders: placeholders.iter().cloned().collect(),
+    };
+    catch_compiler_panic(|| normalize_analysis_sql(request)).map(Cow::Owned)
+}
@@ -213,11 +213,21 @@ pub struct ModelAnalysis {
     pub binding_validated: bool,
 }
 
+/// The inputs a cache hit normalizes into its cleaned SQL only when a reader needs it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DeferredCleaning {
+    pub(crate) dialect: String,
+    pub(crate) query_sql: String,
+    pub(crate) placeholders: Pairs,
+}
+
 /// Python's `ModelSqlAnalysis` for one model, without the placeholders Python keeps.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ModelOutcome {
     pub analysis: ModelAnalysis,
+    /// Empty while `deferred_cleaning` is set; read through `resolved_cleaned_sql`.
     pub cleaned_sql: String,
+    pub deferred_cleaning: Option<DeferredCleaning>,
     pub validated_schema: Shapes,
     pub fused_binding_validated: bool,
 }
@@ -307,6 +317,8 @@ pub struct AnalysisSession {
     pub(crate) available_nullability: ShapeTable,
     pub(crate) complete_shapes: ShapeTable,
     pub(crate) outcomes: Vec<Option<ModelOutcome>>,
+    /// Cache hits whose cleaned SQL is still unnormalized.
+    pub(crate) pending_cleaned: Vec<bool>,
     pub(crate) phase: Phase,
     pub(crate) publications: Shapes,
     pub(crate) failures: Vec<String>,
 
@@ -83,11 +83,17 @@ pub(crate) fn rename_ctes(
     preserves_tokens((sql, &tokens), (&renamed, &renamed_tokens), &replaced).then_some(renamed)
 }
 
-/// Case-folded names of every CTE a statement defines at any depth; `None` when unscannable.
-pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Option<Vec<String>> {
-    let tokens = match tokens(sql, dialect) {
-        Ok(tokens) => tokens,
-        Err(_) => return None,
+/// Case-folded names of every CTE a statement defines at any depth; empty without `with`.
+pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Vec<String> {
+    if !sql
+        .as_bytes()
+        .windows(4)
+        .any(|window| window.eq_ignore_ascii_case(b"with"))
+    {
+        return Vec::new();
+    }
+    let Ok(tokens) = tokens(sql, dialect) else {
+        return Vec::new();
     };
     let code: Vec<&Token> = tokens.iter().filter(|token| !token.comment).collect();
     let text = |index: usize| code.get(index).map(|token| &sql[token.start..token.end]);
@@ -138,7 +144,7 @@ pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Option<Vec<S
             keys.push(key.to_string());
         }
     }
-    Some(keys)
+    keys
 }
 
 /// Whether each token sits directly in a WITH list, so a comma there separates CTEs.
 
@@ -228,7 +228,7 @@ impl ScopeGraph {
         request: &ScopeRequest<'_>,
     ) -> Result<(), String> {
         for (label, sql) in readers {
-            for key in defined_cte_keys(sql, request.slice_dialect).unwrap_or_default() {
+            for key in defined_cte_keys(sql, request.slice_dialect) {
                 if let Some(index) = self.helper_indexes.get(&key) {
                     return Err(compile_error(&format!(
                         "SQL test '{}' defines CTE '{}' inside CTE '{label}', which redefines \
 
@@ -1,7 +1,7 @@
 //! Coarse project-level planning for SQL-native test artifacts.
 
 use std::collections::{BTreeMap, HashMap, HashSet};
-use std::sync::{Arc, Mutex, OnceLock};
+use std::sync::{Arc, Mutex, OnceLock, PoisonError};
 use std::time::Instant;
 
 use polyglot_sql::{Dialect, Expression};
@@ -91,6 +91,8 @@ struct ProjectContext {
     patterns: SqlTestPatterns,
     render_dialect: Arc<Dialect>,
     rejects_nested_with: bool,
+    /// Each model query's reserved CTE name, since every test reading the model checks it.
+    reserved_ctes: Arc<Mutex<HashMap<String, Option<String>>>>,
 }
 
 #[derive(Clone)]
@@ -542,6 +544,7 @@ pub(crate) fn plan_batch(request: SqlTestPlanBatch) -> Result<SqlTestPlanBatchOu
         patterns: SqlTestPatterns::new(request.lexical_syntax)?,
         render_dialect,
         rejects_nested_with,
+        reserved_ctes: Arc::new(Mutex::new(HashMap::new())),
     };
     let workers = request.workers.clamp(1, MAX_WORKERS);
     let pool = rayon::ThreadPoolBuilder::new()
@@ -695,6 +698,27 @@ fn plan_direct_test(
     })
 }
 
+impl ProjectContext {
+    fn reserved_cte_name(&self, query_sql: &str) -> Option<String> {
+        let cached: Option<Option<String>> = self
+            .reserved_ctes
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .get(query_sql)
+            .cloned();
+        if let Some(name) = cached {
+            return name;
+        }
+        let name: Option<String> =
+            reserved_cte_name(query_sql, SliceDialect::new(Some(&self.dialect)));
+        self.reserved_ctes
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .insert(query_sql.to_owned(), name.clone());
+        name
+    }
+}
+
 /// The first CTE a model defines under a name reserved for generated test-query CTEs.
 fn reserved_cte_name(query_sql: &str, dialect: SliceDialect) -> Option<String> {
     let reserved = MOCK_CTE_PREFIXES.iter().copied().chain([
@@ -705,7 +729,6 @@ fn reserved_cte_name(query_sql: &str, dialect: SliceDialect) -> Option<String> {
     ]);
     let prefixes: Vec<&str> = reserved.collect();
     defined_cte_keys(query_sql, dialect)
-        .unwrap_or_default()
         .into_iter()
         .find(|key| is_reserved_cte_key(key, &prefixes))
 }
@@ -769,8 +792,7 @@ fn plan_model_test(
             .model_query_overrides
             .get(model_name)
             .unwrap_or(&model.query_sql);
-        if let Some(name) = reserved_cte_name(query_sql, SliceDialect::new(Some(&context.dialect)))
-        {
+        if let Some(name) = context.reserved_cte_name(query_sql) {
             return Err(compile_error(&format!(
                 "SQL test '{}' reads model '{model_name}', which defines CTE '{name}'; names \
                  starting with a SQLBuild test prefix such as __helper__ or __ref__ are reserved \
 
@@ -449,7 +449,7 @@ impl RenderCteState {
             .collect();
         let keys: HashSet<String> = top_level.iter().map(|(name, _, _)| cte_key(name)).collect();
         for (_, body, origin) in &top_level {
-            let nested = defined_cte_keys(body, self.dialect).unwrap_or_default();
+            let nested = defined_cte_keys(body, self.dialect);
             if let Some(name) = nested.iter().find(|name| keys.contains(*name)) {
                 return Err(format!(
                     "{origin} defines CTE '{name}', which is also a CTE SQLBuild generates at the \
 
@@ -1793,7 +1793,7 @@ fn assert_names_isolated(sql: &str, dialect: &str, helper_name: &str, label: &st
     });
     assert!(!helper_at_top_level, "{label}: {sql}");
     for cte in &split.ctes {
-        let nested = defined_cte_keys(cte.body, slice_dialect).expect("body scans");
+        let nested = defined_cte_keys(cte.body, slice_dialect);
         assert!(
             nested.iter().all(|name| !top_level.contains(name)),
             "{label}: {} nests {nested:?}: {sql}",