feat: contract diffs + breaking-change gates — structured contract analysis, declared renames, persisted contracts

Bundle 7 on top of shared state/cache. Turns the opaque contract_hash
change into a structured, machine-readable contract diff and gates
promotion on breaking contract changes, computed live.

- contracts.rs: contract_diff classifies SAFE/REVIEW/BREAKING per
  change — contract added/dropped, enforcement on/off, column
  added/removed/renamed, type changes (numeric widening review,
  incompatible breaking), nullability tighten/relax.
- Declared renames: [model.<name>.renames] new = "old" resolves a
  removed+added pair into a reviewable rename in both contract and
  physical schema diffs. Stale or ambiguous declarations resolve
  nothing — they cannot hide a real removal or addition.
- MaterializedRecord persists the full contract (contract_json,
  SQLite + Postgres with migration), so base/target comparisons work
  against what the environment actually recorded.
- BranchDiffReport gains contract_changes and lineage-aware impacts
  naming the downstream models/tests a breaking change affects.
- Promotion: contract breaks join physical schema breaks in the schema
  gate, computed live against the target's recorded contracts — a
  contract edited after `diff` cannot slip past on a stale artifact.
  --allow-breaking-schema waives both.
- plan/explain reasons name the changed columns and breaking count
  instead of "contract or assertions changed".

Author: iamgp

SPEC.md

@@ -811,6 +811,14 @@ type = "DOUBLE"
 nullable = false
 ```
 
+Column renames are declared alongside the contract so diffs can classify a
+removed+added pair as a rename rather than a breaking removal:
+
+```toml
+[model.assay_results.renames]
+result = "legacy_result"
+```
+
 Explicit contract files remain optional.
 
 ## 32. Inferred tests
@@ -1069,7 +1077,7 @@ phlo transform promote feature/new-assay --to main
 phlo transform promote --from feature/new-assay --to main
 ```
 
-Promotion is authorised by named gates, reported identically in human and JSON output: `run` (latest candidate run passed and validated the exact commit being promoted — a passed run is bound to the candidate head **after** its writes land, and a candidate that advanced since its run, or a run recorded before commit binding, fails), `tests` (no failed tests), `blocked` (no blocked/cancelled model, seed or test work), `schema` (a fresh audited diff inspected this pair at these commits and found no unwaived breaking changes — absent or stale evidence fails closed, never reading "no evidence" as "no changes"), `data_diff` (when `--require-diff` is set: a passing `--full` audited diff bound to this candidate→target pair at the commits being promoted — the artifact records both refs' resolved heads and is rejected when they no longer match — still fresh per recorded versions, and never a self-comparison; a single-model `diff.json` is never promotion evidence), `base` (the target still equals the commit the evidence was established against — the hash-bound artifact's recorded base, else the candidate's `created_from` provenance; unknown provenance fails rather than redefining the base as the current head, and the merge asserts the evaluated target hash) and `conflicts` (the merge check is clean). A candidate that advanced between gate evaluation and merge is refused rather than promoted unaudited. `--check` evaluates gates without merging; a passing promotion merges and persists a `PromotionRecord` (refs, hashes, plan/run ids, gate results, timestamp) in the state store.
+Promotion is authorised by named gates, reported identically in human and JSON output: `run` (latest candidate run passed and validated the exact commit being promoted — a passed run is bound to the candidate head **after** its writes land, and a candidate that advanced since its run, or a run recorded before commit binding, fails), `tests` (no failed tests), `blocked` (no blocked/cancelled model, seed or test work), `schema` (a fresh audited diff inspected this pair at these commits and found no unwaived breaking changes — physical breaks come from the audited diff; contract breaks are computed live from the workspace's desired contracts against the target's recorded contracts, so a post-`diff` contract edit cannot slip past — and absent or stale evidence fails closed, never reading "no evidence" as "no changes"), `data_diff` (when `--require-diff` is set: a passing `--full` audited diff bound to this candidate→target pair at the commits being promoted — the artifact records both refs' resolved heads and is rejected when they no longer match — still fresh per recorded versions, and never a self-comparison; a single-model `diff.json` is never promotion evidence), `base` (the target still equals the commit the evidence was established against — the hash-bound artifact's recorded base, else the candidate's `created_from` provenance; unknown provenance fails rather than redefining the base as the current head, and the merge asserts the evaluated target hash) and `conflicts` (the merge check is clean). A candidate that advanced between gate evaluation and merge is refused rather than promoted unaudited. `--check` evaluates gates without merging; a passing promotion merges and persists a `PromotionRecord` (refs, hashes, plan/run ids, gate results, timestamp) in the state store.
 
 Preconditions may include successful plan, successful execution, required tests passing, no blocking schema changes, no stale state and optional approval.
 

crates/phlo-transform-cli/src/main.rs

@@ -24,11 +24,11 @@ use phlo_transform_engine::{
     adapter_default_schema, branch_diff, changed_models, collect_source_states, diff, diff_reasons,
     ensure_environment, evaluate_gates, materialized_for_environment, model_keys, promote,
     relation_for_source, retarget, seeds_for_environment, Adapter, ArtifactWriter,
-    BranchDiffReport, BranchDiffRequest, CancelHandle, DatasetKind, DiffPolicy, DiffRequest,
-    DiffStrategy, EnvironmentArtifact, EnvironmentSetup, EnvironmentSpec, ExecutionStatus,
-    GateInput, Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner, PostgresStateStore,
-    PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner, SqliteStateStore,
-    StateStore, SCHEMA_VERSION,
+    BranchDiffReport, BranchDiffRequest, CancelHandle, ContractSafety, DatasetKind, DiffPolicy,
+    DiffRequest, DiffStrategy, EnvironmentArtifact, EnvironmentSetup, EnvironmentSpec,
+    ExecutionStatus, GateInput, Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner,
+    PostgresStateStore, PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner,
+    SqliteStateStore, StateStore, SCHEMA_VERSION,
 };
 use phlo_transform_nessie::{NessieClient, NessieConfig, NessieRestClient};
 use phlo_transform_trino::{TrinoAdapter, TrinoConfig};
@@ -538,6 +538,7 @@ async fn run(cli: &Cli) -> Result<ExitCode, String> {
         } => {
             run_promote(
                 cli,
+                &compilation,
                 candidate.as_deref(),
                 to,
                 *check,
@@ -1128,8 +1129,10 @@ fn build_nessie(cli: &Cli) -> Result<Arc<dyn NessieClient>, String> {
     Ok(Arc::new(client))
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn run_promote(
     cli: &Cli,
+    compilation: &Compilation,
     candidate: Option<&str>,
     to: &str,
     check: bool,
@@ -1186,14 +1189,21 @@ async fn run_promote(
     };
 
     let environment = read_environment_for(cli, candidate);
-    let audit = audited_diff(
+    let mut audit = audited_diff(
         cli,
         state.as_deref(),
         candidate,
         to,
         Some(&candidate_reference.hash),
         Some(&target.hash),
     );
+    // Contract breaks are computed live — the workspace's desired contracts
+    // against what the target environment last recorded — so a contract
+    // edited after `diff` cannot sneak a break past the gate on a stale
+    // artifact's analysis.
+    audit
+        .breaking_schema_changes
+        .extend(contract_breaking_changes(state.as_deref(), compilation, to));
     let merge_check = nessie.can_merge(candidate, to).await.ok();
 
     // The `base` gate needs a target commit the evidence was established
@@ -1346,6 +1356,40 @@ struct AuditEvidence {
     schema_audited: bool,
 }
 
+/// Live contract analysis for the promotion gate: the workspace's desired
+/// contracts against the contracts the target environment last recorded.
+/// Returns the breaking subset in the same `model.column: detail` shape as
+/// physical schema breaks.
+fn contract_breaking_changes(
+    state: Option<&dyn StateStore>,
+    compilation: &Compilation,
+    to: &str,
+) -> Vec<String> {
+    let Some(state) = state else {
+        return Vec::new();
+    };
+    let base = materialized_for_environment(state, to).unwrap_or_default();
+    let mut breaking = Vec::new();
+    for model in &compilation.models {
+        let name = model.id.logical_name();
+        let changes = phlo_transform_engine::contract_diff(
+            base.get(&name).and_then(|record| record.contract.as_ref()),
+            model.contract.as_ref(),
+        );
+        for change in changes {
+            if change.safety == ContractSafety::Breaking {
+                let subject = if change.column.is_empty() {
+                    name.clone()
+                } else {
+                    format!("{name}.{}", change.column)
+                };
+                breaking.push(format!("{subject}: contract {}", change.detail));
+            }
+        }
+    }
+    breaking
+}
+
 /// Read the audited diff artifact (`branch_diff.json` preferred, single-model
 /// `diff.json` as fallback) and derive the evidence it carries for this
 /// promotion: the diff verdict, why the artifact cannot be used, the breaking
@@ -1849,6 +1893,28 @@ fn run_state(cli: &Cli, action: &StateAction) -> Result<ExitCode, String> {
                         println!("  sources:  {}", record.version.source_state_hash);
                         println!("  compiler: {}", record.version.compiler_version);
                         println!("  target:   {}", record.version.target_hash);
+                        if let Some(contract) = &record.contract {
+                            let columns = contract
+                                .columns
+                                .iter()
+                                .map(|column| column.name.clone())
+                                .collect::<Vec<_>>()
+                                .join(", ");
+                            println!(
+                                "Contract:      {} ({} columns{})",
+                                if contract.enforced {
+                                    "enforced"
+                                } else {
+                                    "advisory"
+                                },
+                                contract.columns.len(),
+                                if columns.is_empty() {
+                                    String::new()
+                                } else {
+                                    format!(": {columns}")
+                                }
+                            );
+                        }
                     }
                 }
             }
@@ -3295,6 +3361,11 @@ async fn run_diff(
         strategy,
         policy: diff_policy(compiled.config.diff.as_ref()),
         sample_fraction: sample,
+        renames: compiled
+            .contract
+            .as_ref()
+            .map(|contract| contract.renames.clone())
+            .unwrap_or_default(),
     };
 
     match diff(adapter, &request).await {
@@ -3480,6 +3551,24 @@ fn print_branch_diff_human(report: &BranchDiffReport) {
             }
         }
     }
+    if !report.contract_changes.is_empty() {
+        println!();
+        println!("Contracts");
+        for diff in &report.contract_changes {
+            for change in &diff.changes {
+                let subject = if change.column.is_empty() {
+                    diff.model.clone()
+                } else {
+                    format!("{}.{}", diff.model, change.column)
+                };
+                println!(
+                    "  {subject}: {} [{}]",
+                    change.detail,
+                    change.safety.as_str()
+                );
+            }
+        }
+    }
     if !report.rows.is_empty() {
         println!();
         println!("Rows");
@@ -4137,6 +4226,8 @@ mod tests {
             base_hash: None,
             datasets: Vec::new(),
             schema_changes: Vec::new(),
+            contract_changes: Vec::new(),
+            impacts: Vec::new(),
             rows: Vec::new(),
             diffs: Vec::new(),
             deep,
@@ -4286,6 +4377,7 @@ mod tests {
                 incremental_strategy: None,
                 incremental_key: None,
                 adapter: None,
+                contract: None,
                 run_id: "run-1".to_string(),
                 materialized_at: "t".to_string(),
             })
@@ -4330,6 +4422,7 @@ mod tests {
             incremental_strategy: None,
             incremental_key: None,
             adapter: None,
+            contract: None,
             run_id: "run-1".to_string(),
             materialized_at: "t".to_string(),
         }
@@ -4534,4 +4627,50 @@ mod tests {
         // Nor may a degenerate ref collapse onto the single-slot artifact.
         assert_ne!(environment_artifact_name("///"), "environment.json");
     }
+
+    fn contract(names: &[&str]) -> phlo_transform_core::ModelContract {
+        phlo_transform_core::ModelContract {
+            enforced: true,
+            columns: names
+                .iter()
+                .map(|name| phlo_transform_core::ColumnContract {
+                    name: (*name).to_string(),
+                    data_type: None,
+                    nullable: None,
+                })
+                .collect(),
+            renames: Default::default(),
+        }
+    }
+
+    #[test]
+    fn contract_breaking_changes_reads_recorded_contracts() {
+        let state = SqliteStateStore::in_memory().expect("state");
+        let mut main_record = record("main.base", Some("main"), "v1");
+        main_record.contract = Some(contract(&["id", "legacy"]));
+        state.record_materialized(&main_record).expect("record");
+
+        // The workspace now declares a contract without `legacy` — removing
+        // it is a breaking change against the recorded base contract.
+        let mut base = phlo_transform_core::SemanticModel::in_memory(
+            ModelId::parse("main.base").expect("model id"),
+            "select 1 as id, 2 as legacy",
+        );
+        base.contract = Some(contract(&["id"]));
+        let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![base]));
+        assert!(compilation.is_ok(), "{:?}", compilation.diagnostics);
+
+        let breaks = contract_breaking_changes(Some(&state), &compilation, "main");
+        assert_eq!(breaks.len(), 1, "{breaks:?}");
+        assert!(breaks[0].contains("legacy"), "{breaks:?}");
+
+        // The default environment folds into `main`, so a default-env
+        // record feeds the same gate.
+        let state = SqliteStateStore::in_memory().expect("state");
+        let mut default_record = record("main.base", None, "v1");
+        default_record.contract = Some(contract(&["id", "legacy"]));
+        state.record_materialized(&default_record).expect("record");
+        let breaks = contract_breaking_changes(Some(&state), &compilation, "main");
+        assert_eq!(breaks.len(), 1, "{breaks:?}");
+    }
 }

crates/phlo-transform-core/src/compile.rs

@@ -420,10 +420,16 @@ fn version_inputs(
                     )
                 })
                 .collect();
+            let renames: Vec<String> = contract
+                .renames
+                .iter()
+                .map(|(new, old)| format!("{new}={old}"))
+                .collect();
             format!(
-                "enforced={};columns={}",
+                "enforced={};columns={};renames={}",
                 contract.enforced,
-                columns.join(",")
+                columns.join(","),
+                renames.join(",")
             )
         })
         .unwrap_or_default();

crates/phlo-transform-core/src/config.rs

@@ -70,6 +70,9 @@ pub struct ModelContractConfig {
     pub incremental: IncrementalModelConfig,
     /// `[model.<name>.diff]`.
     pub diff: DiffConfig,
+    /// `[model.<name>.renames]` — column renames the model declares
+    /// (`new name = "old name"`).
+    pub renames: BTreeMap<String, String>,
 }
 
 /// The `[model.<name>.diff]` settings.

crates/phlo-transform-core/src/discovery.rs

@@ -206,7 +206,7 @@ fn normalized_sections(config: &PhloConfig) -> BTreeMap<String, ModelContractCon
 }
 
 fn contract_from_section(section: &ModelContractConfig) -> Option<ModelContract> {
-    if !section.contract.enforced && section.columns.is_empty() {
+    if !section.contract.enforced && section.columns.is_empty() && section.renames.is_empty() {
         return None;
     }
     Some(ModelContract {
@@ -223,6 +223,7 @@ fn contract_from_section(section: &ModelContractConfig) -> Option<ModelContract>
                 nullable: column.nullable,
             })
             .collect(),
+        renames: section.renames.clone(),
     })
 }
 

crates/phlo-transform-core/src/semantic.rs

@@ -7,7 +7,7 @@
 use crate::identity::{ModelId, SourceId};
 
 /// A SQL scalar or nested type as understood by the compiler.
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub enum DataType {
     Boolean,
     TinyInt,
@@ -176,7 +176,7 @@ impl std::fmt::Display for DataType {
 }
 
 /// Nullability of a column. `Unknown` is distinct from `Nullable`.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub enum Nullability {
     NotNull,
     Nullable,
@@ -392,18 +392,23 @@ impl Assertion {
 }
 
 /// An explicit column-level contract.
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct ColumnContract {
     pub name: String,
     pub data_type: Option<DataType>,
     pub nullable: Option<bool>,
 }
 
 /// An explicit model contract.
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct ModelContract {
     pub enforced: bool,
     pub columns: Vec<ColumnContract>,
+    /// Column renames the model declares (`new name` → `old name`), from
+    /// `[model.<name>.renames]`: a removed+added pair that is a declared
+    /// rename classifies as a rename, not a breaking removal.
+    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
+    pub renames: std::collections::BTreeMap<String, String>,
 }
 
 /// Numeric tolerance for a diffed column.

crates/phlo-transform-core/tests/analysis.rs

@@ -330,6 +330,7 @@ fn enforced_contract_violation_is_an_error() {
             data_type: None,
             nullable: None,
         }],
+        renames: Default::default(),
     });
     let compilation = compile(vec![model]);
     assert!(!compilation.is_ok());
@@ -349,6 +350,7 @@ fn contract_type_mismatch_is_reported() {
             data_type: Some(DataType::Varchar),
             nullable: Some(false),
         }],
+        renames: Default::default(),
     });
     let compilation = compile(vec![model]);
     assert!(!compilation.is_ok());

crates/phlo-transform-core/tests/compilation.rs

@@ -484,6 +484,25 @@ fn discovery_attaches_contracts_and_generates_tests() {
     }));
 }
 
+/// `[model.<name>.renames]` lands on the contract so schema and contract
+/// diffs can classify a remove+add pair as a rename.
+#[test]
+fn discovery_attaches_declared_renames() {
+    let compilation = compile_fixture("contract-renames");
+    assert!(compilation.is_ok(), "{:?}", compilation.diagnostics);
+
+    let contract = compilation
+        .model(&ModelId::parse("assay.results").unwrap())
+        .expect("model")
+        .contract
+        .as_ref()
+        .expect("contract attached");
+    assert_eq!(
+        contract.renames.get("result_value").map(String::as_str),
+        Some("concentration")
+    );
+}
+
 #[test]
 fn models_expose_workflow_ownership() {
     let compilation = compile_fixture("cross-workflow");