diff --git a/Cargo.lock b/Cargo.lock
index daada9d..17c5f05 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8187,7 +8187,7 @@ dependencies = [
 
 [[package]]
 name = "schema-forge-cli"
-version = "0.37.3"
+version = "0.37.4"
 dependencies = [
  "acton-service",
  "assert_cmd",
diff --git a/crates/schema-forge-cli/Cargo.toml b/crates/schema-forge-cli/Cargo.toml
index 9a2372a..00c5c75 100644
--- a/crates/schema-forge-cli/Cargo.toml
+++ b/crates/schema-forge-cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "schema-forge-cli"
-version = "0.37.3"
+version = "0.37.4"
 edition = "2021"
 
 [[bin]]
diff --git a/crates/schema-forge-cli/src/commands/apply.rs b/crates/schema-forge-cli/src/commands/apply.rs
index 0db6da5..56576f4 100644
--- a/crates/schema-forge-cli/src/commands/apply.rs
+++ b/crates/schema-forge-cli/src/commands/apply.rs
@@ -1,5 +1,8 @@
 use console::Term;
-use schema_forge_core::migration::{DiffEngine, MigrationSafety};
+use schema_forge_acton::DynSchemaBackend;
+use schema_forge_core::{migration::MigrationSafety, types::SchemaDefinition};
+
+use super::schema_update::SchemaUpdate;
 
 use crate::cli::{ApplyArgs, GlobalOpts};
 use crate::commands::parse::parse_all_schemas_with_global;
@@ -22,19 +25,25 @@ pub async fn run(
 
     let backend = super::connect_backend(&db_params, output).await?;
 
+    apply_to_backend(&args, &schemas, backend.as_ref(), output).await
+}
+
+pub(super) async fn apply_to_backend(
+    args: &ApplyArgs,
+    schemas: &[SchemaDefinition],
+    backend: &dyn DynSchemaBackend,
+    output: &OutputContext,
+) -> Result<(), CliError> {
     let mut total_steps = 0usize;
     let mut applied_schemas = 0usize;
+    let mut metadata_only_updates = 0usize;
 
-    for schema in &schemas {
+    for schema in schemas {
         let existing = backend.load_schema_metadata(&schema.name).await?;
 
-        let plan = if let Some(old) = existing {
-            DiffEngine::diff(&old, schema)
-        } else {
-            DiffEngine::create_new(schema)
-        };
-
-        if plan.is_empty() {
+        let update = SchemaUpdate::plan(existing.as_ref(), schema);
+        let plan = &update.migration;
+        if update.is_empty() {
             output.status(&format!("  {} .... no changes", schema.name.as_str()));
             continue;
         }
@@ -75,7 +84,12 @@ pub async fn run(
         let safety_label = plan.overall_safety();
         match output.mode {
             OutputMode::Human => {
-                if plan.steps.len() == 1
+                if plan.is_empty() {
+                    output.status(&format!(
+                        "  {:<16} METADATA UPDATE (0 migration steps)",
+                        schema.name.as_str()
+                    ));
+                } else if plan.steps.len() == 1
                     && matches!(
                         &plan.steps[0],
                         schema_forge_core::migration::MigrationStep::CreateSchema { .. }
@@ -102,17 +116,19 @@ pub async fn run(
         }
 
         if !args.dry_run {
-            backend.apply_migration(&schema.name, &plan.steps).await?;
-            backend.store_schema_metadata(schema).await?;
+            update.persist(backend).await?;
         }
 
+        if plan.is_empty() {
+            metadata_only_updates += 1;
+        }
         total_steps += plan.steps.len();
         applied_schemas += 1;
     }
 
     // Generate policies if requested
     if args.with_policies && !args.dry_run {
-        for schema in &schemas {
+        for schema in schemas {
             let policies = schema_forge_acton::cedar::generate_cedar_policies(schema);
             output.status(&format!(
                 "  Generated {} Cedar policies for {}",
@@ -140,6 +156,7 @@ pub async fn run(
                 "dry_run": args.dry_run,
                 "schemas_applied": applied_schemas,
                 "total_steps": total_steps,
+                "metadata_only_updates": metadata_only_updates,
             });
             output.print_json(&json);
         }
diff --git a/crates/schema-forge-cli/src/commands/migrate.rs b/crates/schema-forge-cli/src/commands/migrate.rs
index 8673b01..c8a5d25 100644
--- a/crates/schema-forge-cli/src/commands/migrate.rs
+++ b/crates/schema-forge-cli/src/commands/migrate.rs
@@ -1,5 +1,8 @@
 use console::Term;
-use schema_forge_core::migration::DiffEngine;
+use schema_forge_acton::DynSchemaBackend;
+use schema_forge_core::types::SchemaDefinition;
+
+use super::schema_update::SchemaUpdate;
 
 use crate::cli::{GlobalOpts, MigrateArgs};
 use crate::commands::parse::parse_all_schemas_with_global;
@@ -20,11 +23,20 @@ pub async fn run(
 
     let backend = super::connect_backend(&db_params, output).await?;
 
+    migrate_on_backend(&args, &schemas, backend.as_ref(), output).await
+}
+
+pub(super) async fn migrate_on_backend(
+    args: &MigrateArgs,
+    schemas: &[SchemaDefinition],
+    backend: &dyn DynSchemaBackend,
+    output: &OutputContext,
+) -> Result<(), CliError> {
     let mut plans = Vec::new();
     let mut total_steps = 0usize;
     let mut schemas_affected = 0usize;
 
-    for schema in &schemas {
+    for schema in schemas {
         // Filter by --schema if specified
         if let Some(ref filter) = args.schema {
             if schema.name.as_str() != filter {
@@ -33,13 +45,10 @@ pub async fn run(
         }
 
         let existing = backend.load_schema_metadata(&schema.name).await?;
-        let plan = if let Some(old) = existing {
-            DiffEngine::diff(&old, schema)
-        } else {
-            DiffEngine::create_new(schema)
-        };
+        let update = SchemaUpdate::plan(existing.as_ref(), schema);
+        let plan = &update.migration;
 
-        if plan.is_empty() {
+        if update.is_empty() {
             if output.mode == OutputMode::Human {
                 output.status(&format!("{} (no changes)", schema.name.as_str()));
             }
@@ -48,7 +57,7 @@ pub async fn run(
             schemas_affected += 1;
         }
 
-        plans.push((schema, plan));
+        plans.push(update);
     }
 
     // Render plan
@@ -56,8 +65,18 @@ pub async fn run(
         OutputMode::Human => {
             println!("Migration plan for {} schemas:", plans.len());
             println!();
-            for (schema, plan) in &plans {
+            for update in &plans {
+                let schema = &update.schema;
+                let plan = &update.migration;
+                if update.is_empty() {
+                    continue;
+                }
                 if plan.is_empty() {
+                    println!(
+                        "{} (metadata update, 0 migration steps)",
+                        schema.name.as_str()
+                    );
+                    println!();
                     continue;
                 }
                 println!(
@@ -79,8 +98,10 @@ pub async fn run(
         OutputMode::Json => {
             let json_plans: Vec<serde_json::Value> = plans
                 .iter()
-                .filter(|(_, p)| !p.is_empty())
-                .map(|(schema, plan)| {
+                .filter(|update| !update.is_empty())
+                .map(|update| {
+                    let schema = &update.schema;
+                    let plan = &update.migration;
                     let steps: Vec<serde_json::Value> = plan
                         .steps
                         .iter()
@@ -93,6 +114,7 @@ pub async fn run(
                         .collect();
                     serde_json::json!({
                         "schema": schema.name.as_str(),
+                        "metadata_changed": update.metadata_changed,
                         "safety": plan.overall_safety().to_string(),
                         "steps": steps,
                     })
@@ -106,10 +128,15 @@ pub async fn run(
             output.print_json(&json);
         }
         OutputMode::Plain => {
-            for (schema, plan) in &plans {
-                if plan.is_empty() {
+            for update in &plans {
+                let schema = &update.schema;
+                let plan = &update.migration;
+                if update.is_empty() {
                     continue;
                 }
+                if plan.is_empty() {
+                    println!("{}\tmetadata update\tsafe", schema.name.as_str());
+                }
                 for step in &plan.steps {
                     println!("{}\t{}\t{}", schema.name.as_str(), step, step.safety());
                 }
@@ -119,8 +146,10 @@ pub async fn run(
 
     // Execute if requested
     if args.execute {
-        for (schema, plan) in &plans {
-            if plan.is_empty() {
+        for update in &plans {
+            let schema = &update.schema;
+            let plan = &update.migration;
+            if update.is_empty() {
                 continue;
             }
 
@@ -146,8 +175,7 @@ pub async fn run(
                 }
             }
 
-            backend.apply_migration(&schema.name, &plan.steps).await?;
-            backend.store_schema_metadata(schema).await?;
+            update.persist(backend).await?;
         }
 
         output.success(&format!(
diff --git a/crates/schema-forge-cli/src/commands/mod.rs b/crates/schema-forge-cli/src/commands/mod.rs
index 630d0e4..fea3b22 100644
--- a/crates/schema-forge-cli/src/commands/mod.rs
+++ b/crates/schema-forge-cli/src/commands/mod.rs
@@ -11,6 +11,7 @@ pub mod login;
 pub mod migrate;
 pub mod parse;
 pub mod policies;
+mod schema_update;
 pub mod serve;
 #[cfg(feature = "embedded-console")]
 pub mod serve_console;
diff --git a/crates/schema-forge-cli/src/commands/schema_update.rs b/crates/schema-forge-cli/src/commands/schema_update.rs
new file mode 100644
index 0000000..09339a2
--- /dev/null
+++ b/crates/schema-forge-cli/src/commands/schema_update.rs
@@ -0,0 +1,367 @@
+//! Plan metadata persistence separately from physical database changes.
+
+use schema_forge_acton::DynSchemaBackend;
+use schema_forge_core::{
+    migration::{DiffEngine, MigrationPlan},
+    types::SchemaDefinition,
+};
+
+use crate::error::CliError;
+
+pub(super) struct SchemaUpdate {
+    pub schema: SchemaDefinition,
+    pub migration: MigrationPlan,
+    pub metadata_changed: bool,
+}
+
+impl SchemaUpdate {
+    /// Preserve stored identity when comparing freshly parsed definitions.
+    pub fn plan(existing: Option<&SchemaDefinition>, desired: &SchemaDefinition) -> Self {
+        let mut schema = desired.clone();
+        if let Some(existing) = existing {
+            schema.id = existing.id.clone();
+        }
+        let metadata_changed = existing != Some(&schema);
+        let migration = existing.map_or_else(
+            || DiffEngine::create_new(&schema),
+            |existing| DiffEngine::diff(existing, &schema),
+        );
+        Self {
+            schema,
+            migration,
+            metadata_changed,
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.migration.is_empty() && !self.metadata_changed
+    }
+
+    /// Never submit an empty migration merely to persist runtime metadata.
+    pub async fn persist(&self, backend: &dyn DynSchemaBackend) -> Result<(), CliError> {
+        if !self.migration.is_empty() {
+            backend
+                .apply_migration(&self.schema.name, &self.migration.steps)
+                .await?;
+        }
+        if self.metadata_changed {
+            backend.store_schema_metadata(&self.schema).await?;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        cli::{ApplyArgs, GlobalOpts, MigrateArgs},
+        output::{OutputContext, OutputMode},
+    };
+    use schema_forge_backend::BackendError;
+    use schema_forge_core::{migration::MigrationStep, types::SchemaName};
+    use std::{future::Future, pin::Pin, sync::Mutex};
+
+    #[derive(Default)]
+    struct Stored {
+        schema: Option<SchemaDefinition>,
+        migrations: usize,
+        writes: usize,
+    }
+
+    #[derive(Default)]
+    struct Backend {
+        stored: Mutex<Stored>,
+        fail_migration: bool,
+    }
+
+    impl Backend {
+        fn seeded(schema: SchemaDefinition) -> Self {
+            Self {
+                stored: Mutex::new(Stored {
+                    schema: Some(schema),
+                    ..Stored::default()
+                }),
+                fail_migration: false,
+            }
+        }
+    }
+
+    impl DynSchemaBackend for Backend {
+        fn apply_migration<'a>(
+            &'a self,
+            _: &'a SchemaName,
+            steps: &'a [MigrationStep],
+        ) -> Pin<Box<dyn Future<Output = Result<(), BackendError>> + Send + Sync + 'a>> {
+            Box::pin(async move {
+                assert!(
+                    !steps.is_empty(),
+                    "metadata changes must not issue empty migrations"
+                );
+                if self.fail_migration {
+                    return Err(BackendError::MigrationFailed {
+                        step: "test".into(),
+                        reason: "controlled failure".into(),
+                    });
+                }
+                self.stored.lock().unwrap().migrations += 1;
+                Ok(())
+            })
+        }
+        fn store_schema_metadata<'a>(
+            &'a self,
+            schema: &'a SchemaDefinition,
+        ) -> Pin<Box<dyn Future<Output = Result<(), BackendError>> + Send + Sync + 'a>> {
+            Box::pin(async move {
+                let mut stored = self.stored.lock().unwrap();
+                stored.schema = Some(schema.clone());
+                stored.writes += 1;
+                Ok(())
+            })
+        }
+        fn load_schema_metadata<'a>(
+            &'a self,
+            _: &'a SchemaName,
+        ) -> Pin<
+            Box<
+                dyn Future<Output = Result<Option<SchemaDefinition>, BackendError>>
+                    + Send
+                    + Sync
+                    + 'a,
+            >,
+        > {
+            Box::pin(async move { Ok(self.stored.lock().unwrap().schema.clone()) })
+        }
+        fn list_schema_metadata(
+            &self,
+        ) -> Pin<
+            Box<
+                dyn Future<Output = Result<Vec<SchemaDefinition>, BackendError>> + Send + Sync + '_,
+            >,
+        > {
+            Box::pin(
+                async move { Ok(self.stored.lock().unwrap().schema.iter().cloned().collect()) },
+            )
+        }
+    }
+
+    fn schema(source: &str) -> SchemaDefinition {
+        schema_forge_dsl::parse(source).unwrap().remove(0)
+    }
+
+    fn output() -> OutputContext {
+        OutputContext {
+            mode: OutputMode::Human,
+            verbose: 0,
+            quiet: true,
+            use_color: false,
+        }
+    }
+
+    #[derive(Clone, Copy, Debug)]
+    enum Command {
+        Apply,
+        Migrate,
+    }
+
+    impl Command {
+        async fn run(
+            self,
+            backend: &Backend,
+            desired: SchemaDefinition,
+            execute: bool,
+        ) -> Result<(), CliError> {
+            match self {
+                Self::Apply => {
+                    super::super::apply::apply_to_backend(
+                        &ApplyArgs {
+                            paths: vec![],
+                            dry_run: !execute,
+                            force: false,
+                            with_policies: false,
+                        },
+                        &[desired],
+                        backend,
+                        &output(),
+                    )
+                    .await
+                }
+                Self::Migrate => {
+                    super::super::migrate::migrate_on_backend(
+                        &MigrateArgs {
+                            paths: vec![],
+                            execute,
+                            force: false,
+                            schema: None,
+                        },
+                        &[desired],
+                        backend,
+                        &output(),
+                    )
+                    .await
+                }
+            }
+        }
+    }
+
+    const ORIGINAL: &str = "@version(1) schema Person { age: integer }";
+    const METADATA_CHANGES: [&str; 3] = [
+        "@version(2) schema Person { age: integer }",
+        "@version(1) @access(read: [\"reader\"]) schema Person { age: integer }",
+        "@version(1) schema Person { age: integer @require(\"age >= 18\", \"must be adult\") }",
+    ];
+
+    #[tokio::test]
+    async fn commands_persist_metadata_only_changes_and_preserve_identity() {
+        for command in [Command::Apply, Command::Migrate] {
+            for source in METADATA_CHANGES {
+                let original = schema(ORIGINAL);
+                let backend = Backend::seeded(original.clone());
+                let mut desired = schema(source);
+                command.run(&backend, desired.clone(), true).await.unwrap();
+                desired.id = original.id;
+                let stored = backend.stored.lock().unwrap();
+                assert_eq!(
+                    stored.schema.as_ref(),
+                    Some(&desired),
+                    "{command:?} {source}"
+                );
+                assert_eq!(stored.migrations, 0);
+                assert_eq!(stored.writes, 1);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn repeated_parses_are_noops_after_metadata_update() {
+        for command in [Command::Apply, Command::Migrate] {
+            let original = schema(ORIGINAL);
+            let backend = Backend::seeded(original.clone());
+            command.run(&backend, schema(ORIGINAL), true).await.unwrap();
+            assert_eq!(backend.stored.lock().unwrap().writes, 0);
+            command
+                .run(&backend, schema(METADATA_CHANGES[2]), true)
+                .await
+                .unwrap();
+            command
+                .run(&backend, schema(METADATA_CHANGES[2]), true)
+                .await
+                .unwrap();
+            let stored = backend.stored.lock().unwrap();
+            assert_eq!(stored.writes, 1);
+            assert_eq!(stored.migrations, 0);
+            assert_eq!(stored.schema.as_ref().unwrap().id, original.id);
+        }
+    }
+
+    #[tokio::test]
+    async fn dry_runs_never_write_metadata_or_ddl() {
+        for command in [Command::Apply, Command::Migrate] {
+            for source in METADATA_CHANGES {
+                let original = schema(ORIGINAL);
+                let backend = Backend::seeded(original.clone());
+                command.run(&backend, schema(source), false).await.unwrap();
+                let stored = backend.stored.lock().unwrap();
+                assert_eq!(stored.schema.as_ref(), Some(&original));
+                assert_eq!(stored.writes, 0);
+                assert_eq!(stored.migrations, 0);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn physical_changes_still_migrate_before_persisting_metadata() {
+        for command in [Command::Apply, Command::Migrate] {
+            let backend = Backend::default();
+            command.run(&backend, schema(ORIGINAL), true).await.unwrap();
+            command
+                .run(
+                    &backend,
+                    schema("@version(2) schema Person { age: integer name: text }"),
+                    true,
+                )
+                .await
+                .unwrap();
+            let stored = backend.stored.lock().unwrap();
+            assert_eq!(stored.migrations, 2);
+            assert_eq!(stored.writes, 2);
+            assert!(stored.schema.as_ref().unwrap().field("name").is_some());
+        }
+    }
+
+    #[tokio::test]
+    async fn failed_physical_migration_does_not_replace_metadata() {
+        for command in [Command::Apply, Command::Migrate] {
+            let original = schema(ORIGINAL);
+            let mut backend = Backend::seeded(original.clone());
+            backend.fail_migration = true;
+            let result = command
+                .run(
+                    &backend,
+                    schema("@version(2) schema Person { age: integer name: text }"),
+                    true,
+                )
+                .await;
+            assert!(matches!(
+                result,
+                Err(CliError::Backend(BackendError::MigrationFailed { .. }))
+            ));
+            let stored = backend.stored.lock().unwrap();
+            assert_eq!(stored.schema.as_ref(), Some(&original));
+            assert_eq!(stored.writes, 0);
+        }
+    }
+
+    #[tokio::test]
+    async fn invalid_rule_is_rejected_before_backend_connection() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("person.schema");
+        std::fs::write(
+            &path,
+            "@version(2) schema Person { age: integer @require(\"age\", \"must be adult\") }",
+        )
+        .unwrap();
+        let config = dir.path().join("config.toml");
+        std::fs::write(&config, "[schema_forge.signing]\nmode = \"off\"\n").unwrap();
+        let global = GlobalOpts {
+            config: Some(config),
+            format: "human".into(),
+            verbose: 0,
+            quiet: true,
+            no_color: true,
+            db_url: Some("postgres://localhost:1/unreachable".into()),
+            db_ns: None,
+            db_name: None,
+            trust_policy: None,
+            no_verify: false,
+        };
+        let apply = super::super::apply::run(
+            ApplyArgs {
+                paths: vec![path.clone()],
+                dry_run: false,
+                force: false,
+                with_policies: false,
+            },
+            &global,
+            &output(),
+        )
+        .await;
+        let migrate = super::super::migrate::run(
+            MigrateArgs {
+                paths: vec![path],
+                execute: true,
+                force: false,
+                schema: None,
+            },
+            &global,
+            &output(),
+        )
+        .await;
+        for result in [apply, migrate] {
+            assert!(
+                matches!(result, Err(CliError::Parse { .. })),
+                "must fail parsing before attempting connection: {result:?}"
+            );
+        }
+    }
+}