feat(schema): validated enum narrowing + String→enum migration

Enable the Validated-tier enum tightenings. The planner now emits
ChangeEnumConstraint (instead of UnsupportedChange) for:

- narrow (remove allowed variants) → OG-MF-105
- String→enum (constrain a free String) → OG-MF-107

and apply gains a read-only pre-publish scan: for every Validated-tier
ChangeEnumConstraint it opens the affected table at the base snapshot and
runs the existing loader::validate_enum_constraints over every batch. A
row holding a now-disallowed value aborts the whole apply with the
offending value, before any staging write, manifest publish, or schema
rename — so the graph is left untouched (invariant: integrity failures
are loud, never silent data loss). The scan keys targets by table_key in
a BTreeMap for deterministic order and a single scan per type, and runs
before the per-table rewrite loop so a narrow combined with an unrelated
AddProperty aborts before the rewrite advances any HEAD.

This wires the Validated tier the lint chassis reserved but never used,
which also lays the groundwork for OG-MF-104 (nullable tightening).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/omnigraph-compiler/src/catalog/schema_plan.rs

@@ -884,21 +884,15 @@ fn plan_enum_constraint_change(
     // equal `PropType` and never reach here — a reorder-only change is a
     // no-op upstream of this call.
     let code = match (from.enum_values.as_deref(), to.enum_values.as_deref()) {
-        // enum → enum: Safe widen, unless a variant was removed.
+        // enum → enum: Safe widen, unless a variant was removed (narrow is a
+        // Validated tightening — apply scans existing rows for the removed
+        // value before publishing).
         (Some(from_vals), Some(to_vals)) => {
             let removed = from_vals.iter().any(|v| !to_vals.contains(v));
-            if removed {
-                // Narrow is a Validated tightening (OG-MF-105). It is only
-                // accepted once the apply-time row scan exists — until then
-                // fall through to UnsupportedChange so we never admit a narrow
-                // we cannot validate against existing data.
-                return None;
-            }
-            None
+            removed.then_some(crate::lint::codes::OG_MF_105.code)
         }
-        // String → enum is a Validated tightening (OG-MF-107); same staging
-        // as narrow — accepted alongside the apply-time scan.
-        (None, Some(_)) => return None,
+        // String → enum: constraining tightening (Validated, scanned).
+        (None, Some(_)) => Some(crate::lint::codes::OG_MF_107.code),
         // enum → String: loosening, metadata-only, no scan.
         (Some(_), None) => None,
         // Neither side is an enum: not an enum delta.
@@ -1113,6 +1107,47 @@ node Person {
         );
     }
 
+    #[test]
+    fn plan_narrow_enum_is_validated_change() {
+        // Removing an allowed variant may invalidate an existing row, so it's
+        // a Validated change: ChangeEnumConstraint tagged OG-MF-105. The plan
+        // is supported (the row scan runs at apply, not plan).
+        let plan = plan_for(
+            "node Doc {\n    id: String @key\n    status: enum(open, closed, archived)\n}\n",
+            "node Doc {\n    id: String @key\n    status: enum(open, closed)\n}\n",
+        );
+        assert!(plan.supported, "narrow is supported (validated at apply): {plan:?}");
+        assert!(
+            plan.steps.iter().any(|step| matches!(
+                step,
+                ChangeEnumConstraint { property_name, code, .. }
+                    if property_name == "status"
+                        && code.as_deref() == Some(crate::lint::codes::OG_MF_105.code)
+            )),
+            "expected Validated ChangeEnumConstraint (OG-MF-105) for narrow: {plan:?}",
+        );
+    }
+
+    #[test]
+    fn plan_string_to_enum_is_validated_change() {
+        // Constraining a free String to an enum may invalidate an existing
+        // row: Validated, tagged OG-MF-107.
+        let plan = plan_for(
+            "node Doc {\n    id: String @key\n    status: String\n}\n",
+            "node Doc {\n    id: String @key\n    status: enum(open, closed)\n}\n",
+        );
+        assert!(plan.supported, "String->enum is supported (validated at apply): {plan:?}");
+        assert!(
+            plan.steps.iter().any(|step| matches!(
+                step,
+                ChangeEnumConstraint { property_name, code, .. }
+                    if property_name == "status"
+                        && code.as_deref() == Some(crate::lint::codes::OG_MF_107.code)
+            )),
+            "expected Validated ChangeEnumConstraint (OG-MF-107) for String->enum: {plan:?}",
+        );
+    }
+
     #[test]
     fn plan_supports_explicit_type_and_property_rename() {
         let accepted = build_schema_ir(

crates/omnigraph/src/db/omnigraph/schema_apply.rs

@@ -295,6 +295,49 @@ pub(super) async fn apply_schema_with_lock(
         }
     }
 
+    // Validated-tier enum tightenings (narrow / String→enum) require that no
+    // existing row holds a now-disallowed value. Enums are stored as String,
+    // so the committed column already carries the values to check — scan it
+    // read-only here, before any staging write or manifest publish, so a
+    // violation aborts the whole apply with the offending value and leaves the
+    // graph untouched (no HEAD movement, no sidecar, no schema rename). Keying
+    // by table_key in a BTreeMap gives deterministic scan order (stable error)
+    // and collapses multiple enum-prop changes on one type to a single scan.
+    {
+        let mut scan_targets: std::collections::BTreeMap<String, String> =
+            std::collections::BTreeMap::new();
+        for step in &plan.steps {
+            if let SchemaMigrationStep::ChangeEnumConstraint {
+                type_kind,
+                type_name,
+                ..
+            } = step
+            {
+                let needs_scan = step
+                    .diagnostic()
+                    .is_some_and(|d| d.tier == omnigraph_compiler::SafetyTier::Validated);
+                if needs_scan {
+                    scan_targets
+                        .insert(schema_table_key(*type_kind, type_name), type_name.clone());
+                }
+            }
+        }
+        for (table_key, type_name) in &scan_targets {
+            let properties = if table_key.starts_with("edge:") {
+                desired_catalog.edge_types.get(type_name).map(|t| &t.properties)
+            } else {
+                desired_catalog.node_types.get(type_name).map(|t| &t.properties)
+            };
+            let Some(properties) = properties else {
+                continue;
+            };
+            let ds = snapshot.open(table_key).await?;
+            for batch in db.table_store().scan_batches(&ds).await? {
+                crate::loader::validate_enum_constraints(&batch, properties, type_name)?;
+            }
+        }
+    }
+
     let mut table_registrations = HashMap::<String, String>::new();
     let mut table_updates = HashMap::<String, crate::db::SubTableUpdate>::new();
     let mut table_tombstones = HashMap::<String, u64>::new();

crates/omnigraph/tests/schema_apply.rs

@@ -647,6 +647,198 @@ async fn apply_schema_widen_enum_accepts_new_variant() {
     assert_eq!(roles, vec!["guest".to_string()], "role should now be the widened variant");
 }
 
+const PERSON_SET_ROLE: &str =
+    "\nquery set_role($name: String, $role: String) {\n    update Person set { role: $role } where name = $name\n}\n";
+
+#[tokio::test]
+async fn apply_schema_narrow_enum_rejects_when_existing_row_violates() {
+    // Narrowing away a variant that an existing row holds is a Validated
+    // change: the pre-publish scan finds the offending value and aborts.
+    // Nothing moves — manifest version unchanged and the wide enum is still
+    // the accepted contract (re-planning still shows the narrow as pending).
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let schema = "node Person {\n    name: String @key\n    role: enum(admin, member, guest)\n}\n";
+    let mut db = Omnigraph::init(uri, schema).await.unwrap();
+    load_jsonl(
+        &mut db,
+        r#"{"type":"Person","data":{"name":"Alice","role":"guest"}}"#,
+        LoadMode::Overwrite,
+    )
+    .await
+    .unwrap();
+    let before_version = db
+        .snapshot_of(ReadTarget::branch("main"))
+        .await
+        .unwrap()
+        .version();
+
+    let narrowed = "node Person {\n    name: String @key\n    role: enum(admin, member)\n}\n";
+    let plan = db.plan_schema(narrowed).await.unwrap();
+    assert!(plan.supported, "narrow plans as supported (scan at apply): {plan:?}");
+
+    let err = db.apply_schema(narrowed).await.unwrap_err();
+    assert!(
+        err.to_string().contains("invalid enum value 'guest'"),
+        "narrow must abort listing the offending value; got: {err}",
+    );
+
+    // Nothing moved: manifest unchanged, and the wide enum is still accepted.
+    assert_eq!(
+        db.snapshot_of(ReadTarget::branch("main"))
+            .await
+            .unwrap()
+            .version(),
+        before_version,
+        "a rejected narrow must not advance the manifest",
+    );
+    let replan = db.plan_schema(narrowed).await.unwrap();
+    assert!(
+        replan.steps.iter().any(|s| matches!(
+            s,
+            SchemaMigrationStep::ChangeEnumConstraint { property_name, .. } if property_name == "role"
+        )),
+        "the narrow must still be pending — schema contract was not persisted: {replan:?}",
+    );
+}
+
+#[tokio::test]
+async fn apply_schema_narrow_enum_succeeds_when_conforming_then_enforces() {
+    // When no existing row holds a removed value, the narrow applies; the
+    // removed variant is then rejected on subsequent writes.
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let schema = "node Person {\n    name: String @key\n    role: enum(admin, member, guest)\n}\n";
+    let mut db = Omnigraph::init(uri, schema).await.unwrap();
+    load_jsonl(
+        &mut db,
+        r#"{"type":"Person","data":{"name":"Alice","role":"admin"}}"#,
+        LoadMode::Overwrite,
+    )
+    .await
+    .unwrap();
+
+    let narrowed = "node Person {\n    name: String @key\n    role: enum(admin, member)\n}\n";
+    let result = db.apply_schema(narrowed).await.unwrap();
+    assert!(result.supported && result.applied, "conforming narrow must apply");
+
+    // The removed variant is now rejected.
+    let err = mutate_main(
+        &mut db,
+        PERSON_SET_ROLE,
+        "set_role",
+        &params(&[("$name", "Alice"), ("$role", "guest")]),
+    )
+    .await
+    .unwrap_err();
+    assert!(
+        err.to_string().contains("invalid enum value 'guest'"),
+        "narrowed-away variant must be rejected after apply; got: {err}",
+    );
+}
+
+#[tokio::test]
+async fn apply_schema_string_to_enum_validates_existing_rows() {
+    // Constraining a free String to an enum scans existing rows: a row
+    // outside the new set aborts the apply; once the data conforms, the
+    // migration applies and the constraint is enforced on later writes.
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let schema = "node Person {\n    name: String @key\n    role: String\n}\n";
+    let mut db = Omnigraph::init(uri, schema).await.unwrap();
+    load_jsonl(
+        &mut db,
+        r#"{"type":"Person","data":{"name":"Alice","role":"wizard"}}"#,
+        LoadMode::Overwrite,
+    )
+    .await
+    .unwrap();
+
+    let constrained = "node Person {\n    name: String @key\n    role: enum(admin, member)\n}\n";
+
+    // Alice's "wizard" is outside the new set → rejected.
+    let err = db.apply_schema(constrained).await.unwrap_err();
+    assert!(
+        err.to_string().contains("invalid enum value 'wizard'"),
+        "String->enum must scan and reject out-of-set rows; got: {err}",
+    );
+
+    // Bring the data into the set, then the migration applies.
+    mutate_main(
+        &mut db,
+        PERSON_SET_ROLE,
+        "set_role",
+        &params(&[("$name", "Alice"), ("$role", "admin")]),
+    )
+    .await
+    .unwrap();
+    let result = db.apply_schema(constrained).await.unwrap();
+    assert!(result.supported && result.applied, "conforming String->enum must apply");
+
+    // Now the enum is enforced.
+    let err = mutate_main(
+        &mut db,
+        PERSON_SET_ROLE,
+        "set_role",
+        &params(&[("$name", "Alice"), ("$role", "wizard")]),
+    )
+    .await
+    .unwrap_err();
+    assert!(
+        err.to_string().contains("invalid enum value 'wizard'"),
+        "enum must be enforced after String->enum; got: {err}",
+    );
+}
+
+#[tokio::test]
+async fn apply_schema_narrow_enum_with_add_property_aborts_before_rewrite() {
+    // A Validated enum narrow combined with an unrelated AddProperty on the
+    // SAME type: the scan runs before the AddProperty rewrite, so a violation
+    // aborts with nothing moved — the manifest is unchanged and the new
+    // column was never added.
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap();
+    let schema = "node Person {\n    name: String @key\n    role: enum(admin, member, guest)\n}\n";
+    let mut db = Omnigraph::init(uri, schema).await.unwrap();
+    load_jsonl(
+        &mut db,
+        r#"{"type":"Person","data":{"name":"Alice","role":"guest"}}"#,
+        LoadMode::Overwrite,
+    )
+    .await
+    .unwrap();
+    let before_version = db
+        .snapshot_of(ReadTarget::branch("main"))
+        .await
+        .unwrap()
+        .version();
+
+    // Narrow role (removing the held 'guest') AND add a nullable column.
+    let desired = "node Person {\n    name: String @key\n    role: enum(admin, member)\n    nickname: String?\n}\n";
+    let err = db.apply_schema(desired).await.unwrap_err();
+    assert!(
+        err.to_string().contains("invalid enum value 'guest'"),
+        "combined apply must abort on the enum violation; got: {err}",
+    );
+
+    // Nothing moved and the AddProperty rewrite never ran.
+    assert_eq!(
+        db.snapshot_of(ReadTarget::branch("main"))
+            .await
+            .unwrap()
+            .version(),
+        before_version,
+        "aborted combined apply must not advance the manifest",
+    );
+    let person = read_table(&db, "node:Person").await;
+    assert!(
+        person
+            .iter()
+            .all(|b| b.schema().column_with_name("nickname").is_none()),
+        "nickname must not have been added when the apply aborted",
+    );
+}
+
 #[tokio::test]
 async fn plan_schema_for_property_type_narrowing_is_not_supported() {
     // Symmetric companion to `apply_schema_unsupported_plan_does_not_advance_manifest`,