@@ -73,9 +73,13 @@ pub struct SchemaColumnFact {
     pub name: String,
     pub data_type: Option<String>,
     pub nullable: Option<bool>,
-    pub audit_count: u32,
-    /// Whether the column is declared outside the model file, in a named schema.
-    pub declared_outside_model: bool,
+    /// Whether the column declares a description.
+    pub described: bool,
+    pub migrate_from: Option<String>,
+    /// Each column audit's definition name, in declaration order.
+    pub audit_definitions: Vec<String>,
+    /// The file of a column declared outside the model file, in a named schema.
+    pub location_path: Option<String>,
 }
 
 /// One declared dynamic-column family; `key` is its name casefolded.
 
@@ -1,7 +1,7 @@
 //! The Python shape names of the natively hosted compiler shapes.
 
 /// The Python model type a natively hosted compiler shape is captured as.
-pub const HOSTED_SHAPE_TYPES: [(&str, &str); 492] = [
+pub const HOSTED_SHAPE_TYPES: [(&str, &str); 490] = [
     (
         "sqlbuild._native:AlignedInterval",
         "sqlbuild.cursor_algebra.models:AlignedInterval",
@@ -1104,14 +1104,6 @@ pub const HOSTED_SHAPE_TYPES: [(&str, &str); 492] = [
         "sqlbuild._native:NativeLintError",
         "sqlbuild.lint.exceptions:NativeLintError",
     ),
-    (
-        "sqlbuild._native:NativeModelAnalyses",
-        "sqlbuild.compiler.analysis_session.models:NativeModelAnalyses",
-    ),
-    (
-        "sqlbuild._native:NativeModelAnalysisRequest",
-        "sqlbuild.compiler.analysis_session.models:NativeModelAnalysisRequest",
-    ),
     (
         "sqlbuild._native:NativeModelConfigInputs",
         "sqlbuild.compiler.compile.models:NativeModelConfigInputs",
 
@@ -3,15 +3,14 @@
 use pyo3::prelude::{Bound, PyModule, PyModuleMethods, PyResult};
 
 use crate::bindings::_helpers::analysis_session::{
-    fixture_columns, fixture_inferences, model_analyses, oracle, outcome_objects, query_columns,
-    session, stage_functions,
+    fixture_columns, fixture_inferences, oracle, outcome_objects, query_columns, session,
+    stage_functions,
 };
 
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     session::register(module)?;
     fixture_columns::register(module)?;
     module.add_class::<fixture_inferences::FixtureColumnInferences>()?;
-    model_analyses::register(module)?;
     outcome_objects::register(module)?;
     query_columns::register(module)?;
     stage_functions::register(module)?;
 
@@ -2,7 +2,7 @@
 
 use pyo3::prelude::{Bound, Py, PyAny, PyErr, PyModule, PyModuleMethods, PyRef, PyResult, Python};
 use pyo3::types::PyDict;
-use pyo3::{pyclass, pyfunction, pymethods, wrap_pyfunction};
+use pyo3::{pyclass, pyfunction, pymethods};
 use sqlbuild_analysis::assembly::analysis_session::main::attach_analysis_cache::attach_analysis_cache;
 use sqlbuild_analysis::assembly::analysis_session::main::expression_shapes::expression_shapes;
 use sqlbuild_analysis::assembly::analysis_session::main::finish_analysis_session::finish_analysis_session;
@@ -58,7 +58,7 @@ type ModelRow = (
     Vec<FamilyRow>,
 );
 /// The profile settings, relation shapes and models one session analyses.
-type RequestRow = (
+pub(crate) type RequestRow = (
     String,
     bool,
     Pairs,
@@ -72,7 +72,7 @@ type RequestRow = (
     Vec<ModelRow>,
 );
 /// `(publications, failures)`.
-type StepRow = (Shapes, Vec<String>);
+pub(crate) type StepRow = (Shapes, Vec<String>);
 /// One model's analysis, lineage source and cleaned SQL.
 type OutcomeRow = (
     bool,
@@ -108,9 +108,9 @@ type PivotRequestRow = (
 /// `("absent" | "deferred" | "proof", proof)`.
 pub(crate) type ContractRow = (&'static str, Option<ProofRow>);
 /// `(hits, misses, stored, why the store was not read or saved)`.
-type CacheStatsRow = (usize, usize, usize, Option<String>);
+pub(crate) type CacheStatsRow = (usize, usize, usize, Option<String>);
 /// `(catalog schema additions, analysis-shape names, whether each model's analysis succeeded)`.
-type FinishRow = (Shapes, Vec<String>, Vec<bool>);
+pub(crate) type FinishRow = (Shapes, Vec<String>, Vec<bool>);
 
 /// One compile's native model analysis; an internal failure raises `NativeCompilerError`.
 #[pyclass(module = "sqlbuild._native")]
@@ -160,19 +160,9 @@ fn internal_error(reason: &str) -> PyErr {
     compiler_error(native_failure(SESSION_CONTEXT, reason))
 }
 
-#[pymethods]
 impl NativeModelAnalysisSession {
-    /// Models whose output names and lineage the finished session keeps for later stages.
-    #[getter]
-    pub(crate) fn fact_models(&self) -> Vec<String> {
-        self.finished
-            .as_ref()
-            .map(finished_fact_models)
-            .unwrap_or_default()
-    }
-
     /// Analyse every model.
-    fn run(&mut self, py: Python<'_>) -> PyResult<StepRow> {
+    pub(crate) fn run(&mut self, py: Python<'_>) -> PyResult<StepRow> {
         let mut session: AnalysisSession = self.running()?;
         let result: Result<(AnalysisSession, SessionStep), String> =
             py.compiler_detach(move || {
@@ -188,7 +178,7 @@ impl NativeModelAnalysisSession {
     }
 
     /// Every model's outcome; saves the cache best-effort.
-    fn finish(&mut self, py: Python<'_>) -> PyResult<FinishRow> {
+    pub(crate) fn finish(&mut self, py: Python<'_>) -> PyResult<FinishRow> {
         let mut session: AnalysisSession = self.running()?;
         let cache: Option<(NativeStore, AnalysisCacheStats)> = take_analysis_cache(&mut session);
         let (outcome, finished) = catch_compiler_panic(|| finish_analysis_session(session))
@@ -213,9 +203,21 @@ impl NativeModelAnalysisSession {
         self.outcomes = models.into_iter().zip(dynamic_contracts).collect();
         Ok((schema_additions, analysis_names, succeeded))
     }
+}
+
+#[pymethods]
+impl NativeModelAnalysisSession {
+    /// Models whose output names and lineage the finished session keeps for later stages.
+    #[getter]
+    pub(crate) fn fact_models(&self) -> Vec<String> {
+        self.finished
+            .as_ref()
+            .map(finished_fact_models)
+            .unwrap_or_default()
+    }
 
     /// Each requested model's raw binding diagnostic rows, in request order.
-    fn binding_rows(&self) -> Vec<Vec<DiagnosticRow>> {
+    pub(crate) fn binding_rows(&self) -> Vec<Vec<DiagnosticRow>> {
         self.outcomes
             .iter()
             .map(|(outcome, _)| outcome.analysis.binding_diagnostics.clone())
@@ -236,7 +238,7 @@ impl NativeModelAnalysisSession {
 
     /// The analysis cache's `(hits, misses, stored, failure)`; None without a cache.
     #[getter]
-    fn cache_stats(&self) -> Option<CacheStatsRow> {
+    pub(crate) fn cache_stats(&self) -> Option<CacheStatsRow> {
         self.cache_stats.clone()
     }
 
@@ -265,9 +267,7 @@ impl NativeModelAnalysisSession {
 }
 
 /// Start a session on `catalog`.
-#[pyfunction]
-#[pyo3(signature = (catalog, request, cache=None, adapter_rules=None))]
-fn start_model_analysis_session(
+pub(crate) fn start_model_analysis_session(
     catalog: PyRef<'_, ProjectCatalog>,
     request: RequestRow,
     cache: Option<(PathBuf, String)>,
@@ -514,6 +514,5 @@ fn pivot_model(row: PivotModelRow) -> PivotModel {
 
 pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_class::<NativeModelAnalysisSession>()?;
-    module.add_function(wrap_pyfunction!(start_model_analysis_session, module)?)?;
     Ok(())
 }
@@ -6,6 +6,9 @@ use pyo3::types::{PyBool, PyDict, PyDictMethods, PyFrozenSet, PyString, PyTuple}
 use sqlbuild_analysis::assembly::analysis_session::main::native_dialect::native_dialect;
 
 use crate::bindings::_helpers::analysis::binding_catalog::{BindingCatalog, ignores_quoted_case};
+use crate::bindings::_helpers::analysis_session::model_analyses::{
+    ModelAnalyses, ModelAnalysisRequest, analyze_models,
+};
 use crate::bindings::_helpers::compile_driver::analysis_cache::analysis_cache_context;
 use crate::bindings::_helpers::compile_driver::assembly::AssemblyRequest;
 use crate::bindings::_helpers::compile_driver::compile_timings::{
@@ -30,8 +33,8 @@ pub(crate) struct AnalysisContext<'py> {
     pub(crate) nullability: Bound<'py, PyAny>,
     pub(crate) complete_schemas: Bound<'py, PyAny>,
     pub(crate) dynamic_families: Bound<'py, PyAny>,
-    /// The `NativeModelAnalyses`, or `None` when no model was analysed.
-    pub(crate) analyses: Bound<'py, PyAny>,
+    /// The model analyses, or `None` when SQL analysis does not run.
+    pub(crate) analyses: Option<ModelAnalyses<'py>>,
 }
 
 /// Whether `model_input` is selected for analysis by `names` (`None` selects every model).
@@ -90,7 +93,7 @@ pub(crate) fn analysis_context<'py>(
         nullability,
         complete_schemas,
         dynamic_families,
-        analyses: py.None().into_bound(py),
+        analyses: None,
     };
     if context.sql_analysis_enabled.is_truthy()? {
         record_expression_shapes(inputs, &context)?;
@@ -324,10 +327,10 @@ fn record_expression_shapes<'py>(
 fn model_analyses<'py>(
     request: &AssemblyRequest,
     context: &AnalysisContext<'py>,
-) -> PyResult<Bound<'py, PyAny>> {
+) -> PyResult<Option<ModelAnalyses<'py>>> {
     let py: Python<'py> = context.profile.py();
     if !context.sql_analysis_enabled.is_truthy()? {
-        return Ok(py.None().into_bound(py));
+        return Ok(None);
     }
     let inputs: &Bound<'py, PyAny> = request.inputs.bind(py);
     let names: &Bound<'py, PyAny> = request.analysis_model_names.bind(py);
@@ -346,8 +349,6 @@ fn model_analyses<'py>(
     };
     record_compile_timing("model_analysis_ms", || {
         record_compile_cpu_timing(py, "model_analysis_cpu_ms", || {
-            let (known_functions, known_types): (Bound<'py, PyAny>, Bound<'py, PyAny>) =
-                declared_names(inputs, &context.column_types)?;
             let mut model_inputs: Vec<Bound<'py, PyAny>> = Vec::new();
             for model_input in inputs.getattr("model_inputs")?.try_iter()? {
                 let model_input: Bound<'py, PyAny> = model_input?;
@@ -357,29 +358,17 @@ fn model_analyses<'py>(
                     model_inputs.push(model_input);
                 }
             }
-            let analysis_request: Bound<'py, PyAny> = call_with(
-                &registry::class(py, RecordId::NativeModelAnalysisRequest)?,
-                &[
-                    ("known_functions", known_functions),
-                    ("known_types", known_types),
-                    ("model_inputs", PyTuple::new(py, model_inputs)?.into_any()),
-                    ("column_nullability_by_table", context.nullability.clone()),
-                    ("column_types_by_table", context.column_types.clone()),
-                    ("inference_profile", context.profile.clone()),
-                    ("allow_compact_analysis", flags.0.clone()),
-                    ("rich_type_inference", flags.1.clone()),
-                    ("analysis_cache", cache.clone()),
-                    ("complete_binding_schemas", context.complete_schemas.clone()),
-                    (
-                        "dynamic_families_by_table",
-                        context.dynamic_families.clone(),
-                    ),
-                ],
-            )?;
-            call_with(
-                &registry::native(py, NativeAttr::AnalyzeNativeModelSql)?,
-                &[("request", analysis_request)],
-            )
+            analyze_models(&ModelAnalysisRequest {
+                model_inputs,
+                profile: context.profile.clone(),
+                column_types: context.column_types.clone(),
+                nullability: context.nullability.clone(),
+                complete_schemas: context.complete_schemas.clone(),
+                dynamic_families: context.dynamic_families.clone(),
+                rich_type_inference: rich,
+                cache,
+            })
+            .map(Some)
         })
     })
 }
 
@@ -24,16 +24,15 @@ pub(crate) fn native_project_facts<'py>(
     let py: Python<'py> = inputs.py();
     let model_inputs: Bound<'py, PyAny> = inputs.getattr("model_inputs")?;
     let dialect: Bound<'py, PyAny> = context.profile.getattr("sql_analysis_dialect")?;
-    let analysed: bool = !context.analyses.is_none();
-    let succeeded: Bound<'py, PyAny> = if analysed {
-        context.analyses.getattr("succeeded")?
-    } else {
-        PyFrozenSet::empty(py)?.into_any()
-    };
-    let positions: Bound<'py, PyAny> = if analysed {
-        context.analyses.getattr("positions")?
-    } else {
-        PyDict::new(py).into_any()
+    let (succeeded, positions): (Bound<'py, PyAny>, Bound<'py, PyAny>) = match &context.analyses {
+        Some(analyses) => (
+            analyses.succeeded.clone().into_any(),
+            analyses.positions.clone().into_any(),
+        ),
+        None => (
+            PyFrozenSet::empty(py)?.into_any(),
+            PyDict::new(py).into_any(),
+        ),
     };
     let syntax_checks: Bound<'py, PyAny> =
         model_syntax_checks(&model_inputs.clone(), &names.clone(), &succeeded)