From f9ee2c0a850d930643c9231132d7992b5d067d7b Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 14:26:05 +0200
Subject: [PATCH 1/7] feat(schema-engine): add P3026 and a connection string
 sanitizer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A shadow database is reset before a migration history is replayed into it, so
one that holds data the user cares about is a configuration the engine has to
refuse rather than obey. P3026 says so, names the database in question, and
points at the two ways out.

Naming the database means printing a connection string, which is also where the
credentials live, so it goes through a sanitizer: userinfo, the query parameters
that carry secrets, and the JDBC credential properties are dropped, while the
server, port, database name and connection settings stay. SQL Server strings are
parsed as JDBC — the `url` crate reads their whole property list, password
included, as an opaque host — and a string that cannot be parsed is replaced by a
placeholder rather than echoed back with its secrets in unknown positions.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 libs/user-facing-errors/src/schema_engine.rs  |  12 ++
 .../sql-schema-connector/src/lib.rs           |  11 +-
 .../sql-schema-connector/src/sanitize.rs      | 166 ++++++++++++++++++
 3 files changed, 185 insertions(+), 4 deletions(-)
 create mode 100644 schema-engine/connectors/sql-schema-connector/src/sanitize.rs

diff --git a/libs/user-facing-errors/src/schema_engine.rs b/libs/user-facing-errors/src/schema_engine.rs
index d8140d1f0fd..29ceeb0e9af 100644
--- a/libs/user-facing-errors/src/schema_engine.rs
+++ b/libs/user-facing-errors/src/schema_engine.rs
@@ -307,6 +307,18 @@ pub struct UnexpectedNamespaceInExternalTables;
 )]
 pub struct ShadowDbSameAsMainDb;
 
+#[derive(Debug, UserFacingError, Serialize)]
+#[user_facing(
+    code = "P3026",
+    message = "The shadow database at `{shadow_database_location}` is not empty. Prisma Migrate resets the shadow database before it replays your migration history into it, which would destroy the data it currently holds.\n\nPoint `datasource.shadowDatabaseUrl` in your `prisma.config.ts` at an empty database, or, if you are certain that the data in this one can be destroyed, run the command again with `--reset-shadow-database`."
+)]
+pub struct ShadowDbNotEmpty {
+    /// The shadow database, with credentials removed. Reads
+    /// `(driver adapter shadow database)` when the shadow database is reached through a driver
+    /// adapter and has no connection string of its own.
+    pub shadow_database_location: String,
+}
+
 #[derive(Debug, SimpleUserFacingError)]
 #[user_facing(code = "P4001", message = "The introspected database was empty.")]
 pub struct IntrospectionResultEmpty;
diff --git a/schema-engine/connectors/sql-schema-connector/src/lib.rs b/schema-engine/connectors/sql-schema-connector/src/lib.rs
index bb7fdb3ccb3..808387f3647 100644
--- a/schema-engine/connectors/sql-schema-connector/src/lib.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/lib.rs
@@ -10,6 +10,7 @@ mod flavour;
 mod introspection;
 mod migration_pair;
 mod same_database;
+mod sanitize;
 mod sql_destructive_change_checker;
 mod sql_doc_parser;
 mod sql_migration;
@@ -35,6 +36,7 @@ use sql_schema_describer as sql;
 use std::{future, sync::Arc};
 
 pub use same_database::urls_denote_same_database;
+pub use sanitize::{DRIVER_ADAPTER_SHADOW_DATABASE, sanitize_connection_string};
 
 const MIGRATIONS_TABLE_NAME: &str = "_prisma_migrations";
 
@@ -188,10 +190,9 @@ impl SchemaDialect for SqlSchemaDialect {
                     feature = "postgresql-native",
                     feature = "sqlite-native"
                 )))]
-                ExternalShadowDatabase::DriverAdapter {
-                    factory,
-                    preview_features: _,
-                } => self.dialect.connect_to_shadow_db(factory).await?,
+                ExternalShadowDatabase::DriverAdapter { factory, .. } => {
+                    self.dialect.connect_to_shadow_db(factory).await?
+                }
                 #[cfg(any(
                     feature = "mssql-native",
                     feature = "mysql-native",
@@ -201,6 +202,7 @@ impl SchemaDialect for SqlSchemaDialect {
                 ExternalShadowDatabase::ConnectionString {
                     connection_string,
                     preview_features,
+                    ..
                 } => {
                     self.dialect
                         .connect_to_shadow_db(connection_string, preview_features)
@@ -494,6 +496,7 @@ impl SchemaConnector for SqlSchemaConnector {
                     let target = ExternalShadowDatabase::ConnectionString {
                         connection_string: connection_string.to_owned(),
                         preview_features: self.inner.preview_features(),
+                        reset_allowed: self.inner.reset_shadow_database(),
                     };
                     self.schema_dialect()
                         .schema_from_migrations_with_target(migrations, namespaces, filter, target)
diff --git a/schema-engine/connectors/sql-schema-connector/src/sanitize.rs b/schema-engine/connectors/sql-schema-connector/src/sanitize.rs
new file mode 100644
index 00000000000..0fefda2b542
--- /dev/null
+++ b/schema-engine/connectors/sql-schema-connector/src/sanitize.rs
@@ -0,0 +1,166 @@
+//! Naming a database in a user-facing message without leaking the credentials that reach it.
+
+use connection_string::JdbcString;
+use url::Url;
+
+/// How a shadow database reached through a driver adapter is named: there is no connection string
+/// to show.
+pub const DRIVER_ADAPTER_SHADOW_DATABASE: &str = "(driver adapter shadow database)";
+
+/// Stands in for a connection string that could not be parsed, which must not be echoed back
+/// verbatim: what cannot be parsed cannot be stripped of its secrets either.
+const UNPARSEABLE_CONNECTION_STRING: &str = "(unparseable connection string)";
+
+/// The query string parameters that carry a secret rather than a connection setting.
+const SECRET_QUERY_PARAMS: &[&str] = &["api_key", "password", "sslpassword"];
+
+/// The JDBC properties that carry credentials. The `connection-string` crate lower-cases property
+/// names while parsing.
+const CREDENTIAL_JDBC_PROPERTIES: &[&str] = &["user", "password"];
+
+/// Renders a connection string for a user-facing message: the server, the port and the database
+/// name are kept, so that the reader can tell which database is meant, while the credentials that
+/// reach it — userinfo, secret-bearing query parameters, JDBC credential properties — are dropped.
+pub fn sanitize_connection_string(connection_string: &str) -> String {
+    let sanitized = if is_jdbc_connection_string(connection_string) {
+        sanitize_jdbc_connection_string(connection_string)
+    } else {
+        sanitize_url(connection_string)
+    };
+
+    sanitized.unwrap_or_else(|| UNPARSEABLE_CONNECTION_STRING.to_owned())
+}
+
+/// SQL Server connection strings hold their properties in a `;`-separated list, which the `url`
+/// crate happily parses as part of an opaque host, credentials and all. They are recognized by
+/// their scheme instead.
+fn is_jdbc_connection_string(connection_string: &str) -> bool {
+    connection_string.starts_with("sqlserver:") || connection_string.starts_with("jdbc:sqlserver:")
+}
+
+fn sanitize_jdbc_connection_string(connection_string: &str) -> Option<String> {
+    let with_prefix = if connection_string.starts_with("jdbc:") {
+        connection_string.to_owned()
+    } else {
+        format!("jdbc:{connection_string}")
+    };
+
+    let mut jdbc: JdbcString = with_prefix.parse().ok()?;
+    let properties = jdbc.properties_mut();
+
+    for property in CREDENTIAL_JDBC_PROPERTIES {
+        properties.remove(*property);
+    }
+
+    let sanitized = jdbc.to_string();
+
+    Some(if connection_string.starts_with("jdbc:") {
+        sanitized
+    } else {
+        sanitized.trim_start_matches("jdbc:").to_owned()
+    })
+}
+
+fn sanitize_url(connection_string: &str) -> Option<String> {
+    let mut url: Url = connection_string.parse().ok()?;
+
+    if !url.username().is_empty() {
+        url.set_username("").ok()?;
+    }
+
+    if url.password().is_some() {
+        url.set_password(None).ok()?;
+    }
+
+    let params: Vec<(String, String)> = url
+        .query_pairs()
+        .filter(|(name, _)| {
+            !SECRET_QUERY_PARAMS
+                .iter()
+                .any(|secret| name.eq_ignore_ascii_case(secret))
+        })
+        .map(|(name, value)| (name.into_owned(), value.into_owned()))
+        .collect();
+
+    if params.is_empty() {
+        url.set_query(None);
+    } else {
+        url.query_pairs_mut().clear().extend_pairs(params);
+    }
+
+    Some(url.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn credentials_are_stripped_from_urls() {
+        assert_eq!(
+            sanitize_connection_string("postgresql://alice:hunter2@db.example.com:5432/shadow"),
+            "postgresql://db.example.com:5432/shadow"
+        );
+        assert_eq!(
+            sanitize_connection_string("mysql://alice@db.example.com:3306/shadow"),
+            "mysql://db.example.com:3306/shadow"
+        );
+    }
+
+    #[test]
+    fn secret_query_parameters_are_stripped_and_the_rest_is_kept() {
+        assert_eq!(
+            sanitize_connection_string("postgresql://db.example.com/shadow?schema=public&sslmode=require"),
+            "postgresql://db.example.com/shadow?schema=public&sslmode=require"
+        );
+        assert_eq!(
+            sanitize_connection_string("postgresql://db.example.com/shadow?sslpassword=hunter2&schema=public"),
+            "postgresql://db.example.com/shadow?schema=public"
+        );
+        assert_eq!(
+            sanitize_connection_string("prisma+postgres://localhost:51213/?api_key=c2VjcmV0"),
+            "prisma+postgres://localhost:51213/"
+        );
+        assert_eq!(
+            sanitize_connection_string("postgresql://db.example.com/shadow?PASSWORD=hunter2"),
+            "postgresql://db.example.com/shadow"
+        );
+    }
+
+    #[test]
+    fn sqlite_paths_are_kept() {
+        assert_eq!(
+            sanitize_connection_string("file:/tmp/prisma/shadow.db"),
+            "file:///tmp/prisma/shadow.db"
+        );
+    }
+
+    #[test]
+    fn credentials_are_stripped_from_sql_server_connection_strings() {
+        let sanitized =
+            sanitize_connection_string("sqlserver://db.example.com:1433;database=shadow;user=SA;password=hunter2");
+
+        assert!(sanitized.starts_with("sqlserver://db.example.com:1433"), "{sanitized}");
+        assert!(sanitized.contains("database=shadow"), "{sanitized}");
+        assert!(!sanitized.contains("hunter2"), "{sanitized}");
+        assert!(!sanitized.contains("SA"), "{sanitized}");
+    }
+
+    #[test]
+    fn credentials_are_stripped_from_sql_server_connection_strings_without_a_port() {
+        // Without a port, the `url` crate parses the whole property list as an opaque host, so this
+        // is the shape that a URL-only sanitizer would echo back verbatim.
+        let sanitized = sanitize_connection_string("sqlserver://db.example.com;database=shadow;password=hunter2");
+
+        assert!(!sanitized.contains("hunter2"), "{sanitized}");
+        assert!(sanitized.contains("database=shadow"), "{sanitized}");
+    }
+
+    #[test]
+    fn an_unparseable_connection_string_is_not_echoed_back() {
+        assert_eq!(
+            sanitize_connection_string("this is not a connection string"),
+            "(unparseable connection string)"
+        );
+    }
+}

From 949def554f912772a125f593f73697f8d53cff57 Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 14:26:34 +0200
Subject: [PATCH 2/7] feat(schema-engine): carry consent to reset a shadow
 database from the wire
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`resetShadowDatabase` rides the `--datasource` JSON that already carries the
shadow database URL, and the equivalent option rides the Wasm engine's
constructor. Both default to absent-means-no-consent, so callers that know
nothing about either keep working unchanged.

From there the bit reaches the two places that will need it: `ConnectorParams`,
for the connector that holds a shadow database URL of its own, and both
`ExternalShadowDatabase` variants, for the callers that hand a shadow database to
the dialect. Nothing reads it yet — this is the wiring, not the check.

`diff_target_to_dialect` takes the shadow database as one argument rather than
the adapter factory, the preview features and the consent bit separately: the
three only ever travel together into `ExternalShadowDatabase::DriverAdapter`.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 schema-engine/cli/tests/cli_tests.rs          |  1 +
 .../commands/src/commands/dev_diagnostic.rs   |  2 +
 .../commands/diagnose_migration_history.rs    |  2 +
 schema-engine/commands/src/commands/diff.rs   | 23 ++++----
 .../tests/introspection/test_api/mod.rs       |  1 +
 .../tests/migrations/test_api.rs              |  1 +
 .../schema-connector/src/connector_params.rs  |  5 +-
 .../schema-connector/src/schema_connector.rs  |  4 ++
 .../sql-schema-connector/src/flavour.rs       |  3 +
 .../sql-schema-connector/src/flavour/mssql.rs |  6 ++
 .../sql-schema-connector/src/flavour/mysql.rs |  6 ++
 .../src/flavour/postgres.rs                   |  4 ++
 .../flavour/postgres/connector/native/mod.rs  |  6 ++
 .../flavour/postgres/connector/wasm/mod.rs    |  4 ++
 .../src/flavour/sqlite.rs                     |  4 ++
 .../flavour/sqlite/connector/native/mod.rs    |  6 ++
 .../src/flavour/sqlite/connector/wasm/mod.rs  |  4 ++
 schema-engine/core/src/commands/diff_cli.rs   |  8 +++
 schema-engine/core/src/lib.rs                 |  7 +++
 schema-engine/core/src/url.rs                 | 55 +++++++++++++++++++
 .../schema-engine-wasm/src/wasm/engine.rs     | 13 +++++
 .../sql-introspection-tests/src/test_api.rs   |  6 ++
 .../tests/cockroachdb/mod.rs                  |  1 +
 .../sql-introspection-tests/tests/simple.rs   |  1 +
 .../tests/tables/mysql.rs                     |  1 +
 .../src/multi_engine_test_api.rs              |  2 +
 .../sql-migration-tests/src/test_api.rs       |  1 +
 .../tests/errors/error_tests.rs               |  2 +
 .../tests/migrations/diff.rs                  |  8 +++
 .../tests/migrations/jsonrpc.rs               |  1 +
 .../tests/migrations/mssql.rs                 |  1 +
 .../tests/single_migration_tests.rs           |  1 +
 32 files changed, 177 insertions(+), 13 deletions(-)

diff --git a/schema-engine/cli/tests/cli_tests.rs b/schema-engine/cli/tests/cli_tests.rs
index fb73c966397..fe6412495bb 100644
--- a/schema-engine/cli/tests/cli_tests.rs
+++ b/schema-engine/cli/tests/cli_tests.rs
@@ -944,6 +944,7 @@ async fn test_missing_datasource_url_gives_proper_error() {
     let datasource_urls = DatasourceUrls {
         url: None,
         shadow_database_url: None,
+        reset_shadow_database: false,
     };
 
     let output = Command::new(schema_engine_bin_path())
diff --git a/schema-engine/commands/src/commands/dev_diagnostic.rs b/schema-engine/commands/src/commands/dev_diagnostic.rs
index c5228ff0c9c..7f3100582d8 100644
--- a/schema-engine/commands/src/commands/dev_diagnostic.rs
+++ b/schema-engine/commands/src/commands/dev_diagnostic.rs
@@ -17,6 +17,7 @@ pub async fn dev_diagnostic(
     namespaces: Option<Namespaces>,
     connector: &mut dyn SchemaConnector,
     adapter_factory: Arc<dyn ExternalConnectorFactory>,
+    reset_shadow_database: bool,
     migration_schema_cache: &mut MigrationSchemaCache,
 ) -> ConnectorResult<DevDiagnosticOutput> {
     migrations_directory::error_on_changed_provider(&input.migrations_list.lockfile, connector.connector_type())?;
@@ -32,6 +33,7 @@ pub async fn dev_diagnostic(
         namespaces,
         connector,
         adapter_factory,
+        reset_shadow_database,
         migration_schema_cache,
     )
     .await?;
diff --git a/schema-engine/commands/src/commands/diagnose_migration_history.rs b/schema-engine/commands/src/commands/diagnose_migration_history.rs
index 99ca4e2f651..685a35eb49b 100644
--- a/schema-engine/commands/src/commands/diagnose_migration_history.rs
+++ b/schema-engine/commands/src/commands/diagnose_migration_history.rs
@@ -73,6 +73,7 @@ pub async fn diagnose_migration_history(
     namespaces: Option<Namespaces>,
     connector: &mut dyn SchemaConnector,
     adapter_factory: Arc<dyn ExternalConnectorFactory>,
+    reset_shadow_database: bool,
     migration_schema_cache: &mut MigrationSchemaCache,
 ) -> CoreResult<DiagnoseMigrationHistoryOutput> {
     tracing::debug!("Diagnosing migration history");
@@ -149,6 +150,7 @@ pub async fn diagnose_migration_history(
             let target = ExternalShadowDatabase::DriverAdapter {
                 factory: adapter_factory,
                 preview_features: connector.preview_features(),
+                reset_allowed: reset_shadow_database,
             };
             let from = migration_schema_cache
                 .get_or_insert(&applied_migrations.migration_directories, || async {
diff --git a/schema-engine/commands/src/commands/diff.rs b/schema-engine/commands/src/commands/diff.rs
index 9df2dc8040e..4e6f495da1a 100644
--- a/schema-engine/commands/src/commands/diff.rs
+++ b/schema-engine/commands/src/commands/diff.rs
@@ -18,6 +18,7 @@ pub async fn diff(
     params: DiffParams,
     connector: &mut dyn SchemaConnector,
     adapter_factory: Arc<dyn ExternalConnectorFactory>,
+    reset_shadow_database: bool,
     extension_types: &dyn ExtensionTypes,
 ) -> CoreResult<DiffResult> {
     // In order to properly handle MultiSchema, we need to make sure the preview feature is
@@ -30,10 +31,16 @@ pub async fn diff(
     let filter: SchemaFilter = params.filters.into();
     filter.validate(&*connector.schema_dialect())?;
 
+    let shadow_database = ExternalShadowDatabase::DriverAdapter {
+        factory: adapter_factory,
+        preview_features,
+        reset_allowed: reset_shadow_database,
+    };
+
     let (conn_from, schema_from) = diff_target_to_dialect(
         &params.from,
         connector,
-        adapter_factory.clone(),
+        shadow_database.clone(),
         namespaces.clone(),
         &filter,
         preview_features,
@@ -45,7 +52,7 @@ pub async fn diff(
     let (conn_to, schema_to) = diff_target_to_dialect(
         &params.to,
         connector,
-        adapter_factory,
+        shadow_database,
         namespaces,
         &filter,
         preview_features,
@@ -118,7 +125,7 @@ fn namespaces_and_preview_features_from_diff_targets(
 async fn diff_target_to_dialect(
     target: &DiffTarget,
     connector: &mut dyn SchemaConnector,
-    adapter_factory: Arc<dyn ExternalConnectorFactory>,
+    shadow_database: ExternalShadowDatabase,
     namespaces: Option<Namespaces>,
     filter: &SchemaFilter,
     preview_features: BitFlags<psl::PreviewFeature>,
@@ -160,15 +167,7 @@ async fn diff_target_to_dialect(
 
                     // TODO: enable Driver Adapter for shadow database, using the AdapterFactory.
                     let schema = dialect
-                        .schema_from_migrations_with_target(
-                            &migrations,
-                            namespaces,
-                            filter,
-                            ExternalShadowDatabase::DriverAdapter {
-                                factory: adapter_factory,
-                                preview_features,
-                            },
-                        )
+                        .schema_from_migrations_with_target(&migrations, namespaces, filter, shadow_database)
                         .await?;
                     Ok(Some((dialect, schema)))
                 }
diff --git a/schema-engine/connectors/mongodb-schema-connector/tests/introspection/test_api/mod.rs b/schema-engine/connectors/mongodb-schema-connector/tests/introspection/test_api/mod.rs
index 544b0067b10..6502f406fd4 100644
--- a/schema-engine/connectors/mongodb-schema-connector/tests/introspection/test_api/mod.rs
+++ b/schema-engine/connectors/mongodb-schema-connector/tests/introspection/test_api/mod.rs
@@ -114,6 +114,7 @@ where
             connection_string: connection_string.clone(),
             preview_features,
             shadow_database_connection_string: None,
+            reset_shadow_database: false,
         };
 
         let connector = MongoDbSchemaConnector::new(params);
diff --git a/schema-engine/connectors/mongodb-schema-connector/tests/migrations/test_api.rs b/schema-engine/connectors/mongodb-schema-connector/tests/migrations/test_api.rs
index 4174a701e62..3efa930fc93 100644
--- a/schema-engine/connectors/mongodb-schema-connector/tests/migrations/test_api.rs
+++ b/schema-engine/connectors/mongodb-schema-connector/tests/migrations/test_api.rs
@@ -82,6 +82,7 @@ fn new_connector(preview_features: BitFlags<PreviewFeature>) -> (String, MongoDb
         connection_string: url.to_string(),
         preview_features,
         shadow_database_connection_string: None,
+        reset_shadow_database: false,
     };
     (db_name, MongoDbSchemaConnector::new(params))
 }
diff --git a/schema-engine/connectors/schema-connector/src/connector_params.rs b/schema-engine/connectors/schema-connector/src/connector_params.rs
index 416da39848b..83f736d0e10 100644
--- a/schema-engine/connectors/schema-connector/src/connector_params.rs
+++ b/schema-engine/connectors/schema-connector/src/connector_params.rs
@@ -10,10 +10,12 @@ pub struct ConnectorParams {
     pub preview_features: BitFlags<PreviewFeature>,
     /// The shadow database connection string.
     pub shadow_database_connection_string: Option<String>,
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    pub reset_shadow_database: bool,
 }
 
 impl ConnectorParams {
-    /// Creates new [`ConnectorParams`].
+    /// Creates new [`ConnectorParams`], without consent to reset a non-empty shadow database.
     pub fn new(
         connection_string: String,
         preview_features: BitFlags<PreviewFeature>,
@@ -23,6 +25,7 @@ impl ConnectorParams {
             connection_string,
             preview_features,
             shadow_database_connection_string,
+            reset_shadow_database: false,
         }
     }
 }
diff --git a/schema-engine/connectors/schema-connector/src/schema_connector.rs b/schema-engine/connectors/schema-connector/src/schema_connector.rs
index 6727caa982b..2eeb222c274 100644
--- a/schema-engine/connectors/schema-connector/src/schema_connector.rs
+++ b/schema-engine/connectors/schema-connector/src/schema_connector.rs
@@ -231,6 +231,8 @@ pub enum ExternalShadowDatabase {
         factory: Arc<dyn ExternalConnectorFactory>,
         /// The preview features to use while building the shadow schema.
         preview_features: PreviewFeatures,
+        /// Whether the user consented to the shadow database being reset even when it is not empty.
+        reset_allowed: bool,
     },
     /// A shadow database connection string and preview features.
     ConnectionString {
@@ -238,5 +240,7 @@ pub enum ExternalShadowDatabase {
         connection_string: String,
         /// The preview features.
         preview_features: PreviewFeatures,
+        /// Whether the user consented to the shadow database being reset even when it is not empty.
+        reset_allowed: bool,
     },
 }
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour.rs b/schema-engine/connectors/sql-schema-connector/src/flavour.rs
index 6b105b7bf66..9cab461223a 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour.rs
@@ -165,6 +165,9 @@ pub(crate) trait SqlConnector: Send + Sync + Debug {
 
     fn shadow_db_url(&self) -> Option<&str>;
 
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    fn reset_shadow_database(&self) -> bool;
+
     fn acquire_lock(&mut self) -> BoxFuture<'_, ConnectorResult<()>>;
 
     fn apply_migration_script<'a>(
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/mssql.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/mssql.rs
index 8cd6a75b8cc..03304477406 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/mssql.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/mssql.rs
@@ -180,6 +180,12 @@ impl SqlConnector for MssqlConnector {
             .as_deref()
     }
 
+    fn reset_shadow_database(&self) -> bool {
+        self.state
+            .params()
+            .is_some_and(|params| params.connector_params.reset_shadow_database)
+    }
+
     fn acquire_lock(&mut self) -> BoxFuture<'_, ConnectorResult<()>> {
         // see
         // https://docs.microsoft.com/en-us/sql/relational-databases/system-stored-procedures/sp-getapplock-transact-sql?view=sql-server-ver15
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/mysql.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/mysql.rs
index d901420cd4d..efd08fa9efd 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/mysql.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/mysql.rs
@@ -160,6 +160,12 @@ impl SqlConnector for MysqlConnector {
             .as_deref()
     }
 
+    fn reset_shadow_database(&self) -> bool {
+        self.state
+            .params()
+            .is_some_and(|params| params.connector_params.reset_shadow_database)
+    }
+
     fn acquire_lock(&mut self) -> BoxFuture<'_, ConnectorResult<()>> {
         with_connection(&mut self.state, |params, _, connection| async move {
             // We do not acquire advisory locks on PlanetScale instances.
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres.rs
index be48eddd1f5..bbb639312bf 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres.rs
@@ -360,6 +360,10 @@ impl SqlConnector for PostgresConnector {
         imp::get_shadow_db_url(&self.state)
     }
 
+    fn reset_shadow_database(&self) -> bool {
+        imp::get_reset_shadow_database(&self.state)
+    }
+
     fn acquire_lock(&mut self) -> BoxFuture<'_, ConnectorResult<()>> {
         // They do not support advisory locking:
         // https://github.com/cockroachdb/cockroach/issues/13546
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/mod.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/mod.rs
index 135a6beb381..de4f38ba05b 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/mod.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/mod.rs
@@ -421,6 +421,12 @@ pub fn get_shadow_db_url(state: &State) -> Option<&str> {
         .as_deref()
 }
 
+pub fn get_reset_shadow_database(state: &State) -> bool {
+    state
+        .params()
+        .is_some_and(|params| params.connector_params.reset_shadow_database)
+}
+
 pub async fn dispose(state: &mut State) -> ConnectorResult<()> {
     if let State::Connected(_, (_, conn)) = std::mem::replace(state, State::Initial) {
         conn.close().await;
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/wasm/mod.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/wasm/mod.rs
index 15877f367cd..1faaa2e8318 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/wasm/mod.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/wasm/mod.rs
@@ -141,6 +141,10 @@ pub fn get_shadow_db_url(_state: &State) -> Option<&str> {
     None
 }
 
+pub fn get_reset_shadow_database(_state: &State) -> bool {
+    false
+}
+
 pub async fn dispose(state: &State) -> ConnectorResult<()> {
     state.connection.dispose().await
 }
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite.rs
index 1b12e2cd7e0..b691f493576 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite.rs
@@ -139,6 +139,10 @@ impl SqlConnector for SqliteConnector {
         imp::get_shadow_db_url(&self.state)
     }
 
+    fn reset_shadow_database(&self) -> bool {
+        imp::get_reset_shadow_database(&self.state)
+    }
+
     fn acquire_lock(&mut self) -> BoxFuture<'_, ConnectorResult<()>> {
         self.with_connection(|conn, _| acquire_lock(conn))
     }
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/native/mod.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/native/mod.rs
index 7b3d78d77e2..719e88d63ff 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/native/mod.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/native/mod.rs
@@ -315,6 +315,12 @@ pub fn get_shadow_db_url(state: &State) -> Option<&str> {
         .as_deref()
 }
 
+pub fn get_reset_shadow_database(state: &State) -> bool {
+    state
+        .params()
+        .is_some_and(|params| params.connector_params.reset_shadow_database)
+}
+
 pub async fn dispose(_state: &State) -> ConnectorResult<()> {
     // Nothing to on dispose, the connection is disposed in Drop
     Ok(())
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/wasm/mod.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/wasm/mod.rs
index ce0f86c875b..a0cb5ee49bc 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/wasm/mod.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/sqlite/connector/wasm/mod.rs
@@ -174,6 +174,10 @@ pub fn get_shadow_db_url(_state: &State) -> Option<&str> {
     None
 }
 
+pub fn get_reset_shadow_database(_state: &State) -> bool {
+    false
+}
+
 pub async fn dispose(state: &State) -> ConnectorResult<()> {
     state.connection.dispose().await
 }
diff --git a/schema-engine/core/src/commands/diff_cli.rs b/schema-engine/core/src/commands/diff_cli.rs
index ffee8f00b69..ae563e046ae 100644
--- a/schema-engine/core/src/commands/diff_cli.rs
+++ b/schema-engine/core/src/commands/diff_cli.rs
@@ -263,6 +263,7 @@ async fn json_rpc_diff_target_to_dialect(
                             ExternalShadowDatabase::ConnectionString {
                                 connection_string: shadow_database_url.to_owned(),
                                 preview_features,
+                                reset_allowed: datasource_urls.reset_shadow_database,
                             },
                         )
                         .await?;
@@ -344,6 +345,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: Some(MAIN_URL.to_owned()),
+                reset_shadow_database: false,
             },
         ));
     }
@@ -356,6 +358,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: Some("postgres://user:password@LOCALHOST/maindb?schema=shadow".to_owned()),
+                reset_shadow_database: false,
             },
         ));
     }
@@ -368,6 +371,7 @@ mod tests {
             DatasourceUrls {
                 url: None,
                 shadow_database_url: Some(MAIN_URL.to_owned()),
+                reset_shadow_database: false,
             },
         ));
     }
@@ -380,6 +384,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: Some("postgresql://user:password@localhost:5432/shadowdb".to_owned()),
+                reset_shadow_database: false,
             },
         )
         .unwrap();
@@ -393,6 +398,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: Some(MAIN_URL.to_owned()),
+                reset_shadow_database: false,
             },
         )
         .unwrap();
@@ -406,6 +412,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: None,
+                reset_shadow_database: false,
             },
         )
         .unwrap();
@@ -421,6 +428,7 @@ mod tests {
             DatasourceUrls {
                 url: Some(MAIN_URL.to_owned()),
                 shadow_database_url: Some("postgres://user:password@LOCALHOST/maindb".to_owned()),
+                reset_shadow_database: false,
             },
         )
         .unwrap();
diff --git a/schema-engine/core/src/lib.rs b/schema-engine/core/src/lib.rs
index 356a93b2986..adf977cbcc6 100644
--- a/schema-engine/core/src/lib.rs
+++ b/schema-engine/core/src/lib.rs
@@ -49,6 +49,7 @@ fn connector_for_connection_string(
                 connection_string,
                 preview_features,
                 shadow_database_connection_string,
+                reset_shadow_database: false,
             };
             Ok(Box::new(SqlSchemaConnector::new_postgres_like(params)?))
         }
@@ -57,6 +58,7 @@ fn connector_for_connection_string(
                 connection_string,
                 preview_features,
                 shadow_database_connection_string,
+                reset_shadow_database: false,
             };
             Ok(Box::new(SqlSchemaConnector::new_sqlite(params)?))
         }
@@ -65,6 +67,7 @@ fn connector_for_connection_string(
                 connection_string,
                 preview_features,
                 shadow_database_connection_string,
+                reset_shadow_database: false,
             };
             Ok(Box::new(SqlSchemaConnector::new_mysql(params)?))
         }
@@ -73,6 +76,7 @@ fn connector_for_connection_string(
                 connection_string,
                 preview_features,
                 shadow_database_connection_string,
+                reset_shadow_database: false,
             };
             Ok(Box::new(SqlSchemaConnector::new_mssql(params)?))
         }
@@ -81,6 +85,7 @@ fn connector_for_connection_string(
                 connection_string,
                 preview_features,
                 shadow_database_connection_string,
+                reset_shadow_database: false,
             };
             let connector = MongoDbSchemaConnector::new(params);
             Ok(Box::new(connector))
@@ -133,6 +138,7 @@ fn schema_to_connector(
         connection_string,
         preview_features,
         shadow_database_connection_string,
+        reset_shadow_database: datasource_urls.reset_shadow_database(),
     };
 
     connector_for_provider(datasource.active_provider, params)
@@ -152,6 +158,7 @@ fn initial_datamodel_to_connector(
             .to_owned(),
         preview_features,
         shadow_database_connection_string: datasource_urls.shadow_database_url().map(<_>::to_owned),
+        reset_shadow_database: datasource_urls.reset_shadow_database(),
     };
 
     connector_for_provider(datasource.active_provider, params)
diff --git a/schema-engine/core/src/url.rs b/schema-engine/core/src/url.rs
index 915c389d9bc..23c05a75c90 100644
--- a/schema-engine/core/src/url.rs
+++ b/schema-engine/core/src/url.rs
@@ -15,6 +15,10 @@ pub struct DatasourceUrls {
     pub url: Option<String>,
     /// The URL to a live shadow database, if Prisma should use it instead of creating one.
     pub shadow_database_url: Option<String>,
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    /// Absent means no consent was given.
+    #[serde(default)]
+    pub reset_shadow_database: bool,
 }
 
 /// Datasource URLs that have passed validation and are safe to consume by the schema engine.
@@ -22,6 +26,7 @@ pub struct DatasourceUrls {
 pub struct ValidatedDatasourceUrls {
     url: Option<String>,
     shadow_database_url: Option<String>,
+    reset_shadow_database: bool,
 }
 
 /// Errors produced while validating datasource URLs supplied from configuration.
@@ -45,6 +50,7 @@ impl DatasourceUrls {
         Self {
             url: Some(url.into()),
             shadow_database_url: None,
+            reset_shadow_database: false,
         }
     }
 
@@ -64,6 +70,7 @@ impl DatasourceUrls {
         Ok(ValidatedDatasourceUrls {
             url: self.url.clone(),
             shadow_database_url: self.shadow_database_url.clone(),
+            reset_shadow_database: self.reset_shadow_database,
         })
     }
 }
@@ -91,6 +98,7 @@ impl From<ValidatedDatasourceUrls> for DatasourceUrls {
         Self {
             url: urls.url,
             shadow_database_url: urls.shadow_database_url,
+            reset_shadow_database: urls.reset_shadow_database,
         }
     }
 }
@@ -118,6 +126,11 @@ impl ValidatedDatasourceUrls {
         self.shadow_database_url.as_deref()
     }
 
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    pub fn reset_shadow_database(&self) -> bool {
+        self.reset_shadow_database
+    }
+
     /// Resolves relative paths in the URL against the provided configuration directory.
     pub fn url_with_config_dir(&self, flavour: Flavour, config_dir: &Path) -> Option<Cow<'_, str>> {
         self.url.as_deref().map(|url| set_config_dir(flavour, config_dir, url))
@@ -250,3 +263,45 @@ fn set_config_dir_sqlite<'a>(config_dir: &Path, url: &'a str) -> Cow<'a, str> {
 
     Cow::Borrowed(url)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn datasource_urls_without_reset_shadow_database_parse_as_before() {
+        let urls: DatasourceUrls = serde_json::from_str(
+            r#"{ "url": "postgresql://localhost:5432/main", "shadowDatabaseUrl": "postgresql://localhost:5432/shadow" }"#,
+        )
+        .unwrap();
+
+        assert_eq!(urls.url.as_deref(), Some("postgresql://localhost:5432/main"));
+        assert_eq!(
+            urls.shadow_database_url.as_deref(),
+            Some("postgresql://localhost:5432/shadow")
+        );
+        assert!(!urls.reset_shadow_database);
+    }
+
+    #[test]
+    fn reset_shadow_database_is_read_from_the_wire() {
+        let urls: DatasourceUrls =
+            serde_json::from_str(r#"{ "url": "postgresql://localhost:5432/main", "resetShadowDatabase": true }"#)
+                .unwrap();
+
+        assert!(urls.reset_shadow_database);
+    }
+
+    #[test]
+    fn reset_shadow_database_survives_validation() {
+        let urls = DatasourceUrls {
+            url: Some("postgresql://localhost:5432/main".to_owned()),
+            shadow_database_url: Some("postgresql://localhost:5432/shadow".to_owned()),
+            reset_shadow_database: true,
+        };
+
+        let validated = urls.validate(psl::builtin_connectors::POSTGRES).unwrap();
+
+        assert!(validated.reset_shadow_database());
+    }
+}
diff --git a/schema-engine/schema-engine-wasm/src/wasm/engine.rs b/schema-engine/schema-engine-wasm/src/wasm/engine.rs
index 4aca970fb1d..3ba6d55fc31 100644
--- a/schema-engine/schema-engine-wasm/src/wasm/engine.rs
+++ b/schema-engine/schema-engine-wasm/src/wasm/engine.rs
@@ -56,6 +56,11 @@ pub fn version() -> String {
 pub struct ConstructorOptions {
     /// The initial datamodels to use.
     datamodels: Option<Vec<(String, String)>>,
+
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    /// Absent means no consent was given.
+    #[serde(default)]
+    reset_shadow_database: bool,
 }
 
 /// The main query engine used by JS
@@ -70,6 +75,9 @@ pub struct SchemaEngine {
     /// The inferred database namespaces (used for the `multiSchema` preview feature).
     namespaces: Option<Namespaces>,
 
+    /// Whether the user consented to the shadow database being reset even when it is not empty.
+    reset_shadow_database: bool,
+
     /// The dispatcher for tracing.
     dispatch: Dispatch,
 
@@ -95,6 +103,7 @@ impl SchemaEngine {
         async move {
             let ConstructorOptions {
                 datamodels: initial_datamodels,
+                reset_shadow_database,
             } = options;
 
             let adapter_factory = Arc::new(adapter_factory_from_js(adapter));
@@ -125,6 +134,7 @@ impl SchemaEngine {
                 adapter_factory,
                 connector,
                 namespaces,
+                reset_shadow_database,
                 dispatch,
                 migration_schema_cache: MigrationSchemaCache::new(),
             })
@@ -218,6 +228,7 @@ impl SchemaEngine {
                 namespaces,
                 &mut self.connector,
                 self.adapter_factory.clone(),
+                self.reset_shadow_database,
                 &mut self.migration_schema_cache,
             )
             .instrument(tracing::info_span!("DevDiagnostic"))
@@ -237,6 +248,7 @@ impl SchemaEngine {
                 params,
                 &mut self.connector,
                 self.adapter_factory.clone(),
+                self.reset_shadow_database,
                 &NoExtensionTypes,
             )
             .instrument(tracing::info_span!("Diff"))
@@ -261,6 +273,7 @@ impl SchemaEngine {
                 namespaces,
                 &mut self.connector,
                 self.adapter_factory.clone(),
+                self.reset_shadow_database,
                 &mut self.migration_schema_cache,
             )
             .instrument(tracing::info_span!("DiagnoseMigrationHistory"))
diff --git a/schema-engine/sql-introspection-tests/src/test_api.rs b/schema-engine/sql-introspection-tests/src/test_api.rs
index 69fecd41f9e..daa219a0458 100644
--- a/schema-engine/sql-introspection-tests/src/test_api.rs
+++ b/schema-engine/sql-introspection-tests/src/test_api.rs
@@ -53,6 +53,7 @@ impl TestApi {
                 connection_string: connection_string.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let mut me = SqlSchemaConnector::new_mysql(params).unwrap();
 
@@ -75,6 +76,7 @@ impl TestApi {
                 connection_string: cs.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let me = SqlSchemaConnector::new_mysql(params).unwrap();
 
@@ -85,6 +87,7 @@ impl TestApi {
                 connection_string: cs.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let me = SqlSchemaConnector::new_postgres(params).unwrap();
 
@@ -104,6 +107,7 @@ impl TestApi {
                 connection_string: cs.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let me = SqlSchemaConnector::new_cockroach(params).unwrap();
 
@@ -115,6 +119,7 @@ impl TestApi {
                 connection_string: cs.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let me = SqlSchemaConnector::new_mssql(params).unwrap();
 
@@ -126,6 +131,7 @@ impl TestApi {
                 connection_string: url.to_owned(),
                 preview_features,
                 shadow_database_connection_string: None,
+                reset_shadow_database: false,
             };
             let me = SqlSchemaConnector::new_sqlite(params).unwrap();
 
diff --git a/schema-engine/sql-introspection-tests/tests/cockroachdb/mod.rs b/schema-engine/sql-introspection-tests/tests/cockroachdb/mod.rs
index d061fdf1514..f3d05d64d2a 100644
--- a/schema-engine/sql-introspection-tests/tests/cockroachdb/mod.rs
+++ b/schema-engine/sql-introspection-tests/tests/cockroachdb/mod.rs
@@ -40,6 +40,7 @@ async fn introspecting_cockroach_db_with_postgres_provider_fails(api: TestApi) {
         connection_string: api.connection_string().to_owned(),
         preview_features: api.preview_features(),
         shadow_database_connection_string: None,
+        reset_shadow_database: false,
     };
     let mut engine = SqlSchemaConnector::new_postgres(params).unwrap();
 
diff --git a/schema-engine/sql-introspection-tests/tests/simple.rs b/schema-engine/sql-introspection-tests/tests/simple.rs
index e1f8ce26de7..7df00272f11 100644
--- a/schema-engine/sql-introspection-tests/tests/simple.rs
+++ b/schema-engine/sql-introspection-tests/tests/simple.rs
@@ -139,6 +139,7 @@ source .test_database_urls/mysql_5_6
         connection_string: database_url,
         preview_features,
         shadow_database_connection_string: None,
+        reset_shadow_database: false,
     };
 
     let mut api = match provider {
diff --git a/schema-engine/sql-introspection-tests/tests/tables/mysql.rs b/schema-engine/sql-introspection-tests/tests/tables/mysql.rs
index 019e22c06aa..4c2b2af1925 100644
--- a/schema-engine/sql-introspection-tests/tests/tables/mysql.rs
+++ b/schema-engine/sql-introspection-tests/tests/tables/mysql.rs
@@ -377,6 +377,7 @@ async fn missing_select_rights(api: &mut TestApi) -> TestResult {
         connection_string: url.to_string(),
         preview_features: Default::default(),
         shadow_database_connection_string: None,
+        reset_shadow_database: false,
     };
 
     let mut conn = SqlSchemaConnector::new_mysql(params).unwrap();
diff --git a/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs b/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
index 155b382ea93..079c0541e50 100644
--- a/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
+++ b/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
@@ -50,6 +50,7 @@ impl TestApi {
                 connection_string: args.database_url().to_owned(),
                 preview_features,
                 shadow_database_connection_string: args.shadow_database_url().map(String::from),
+                reset_shadow_database: false,
             };
             let mut conn = SqlSchemaConnector::new_mysql(params).unwrap();
             tok(conn.reset(false, None, &schema_connector::SchemaFilter::default())).unwrap();
@@ -214,6 +215,7 @@ impl TestApi {
             connection_string,
             preview_features: self.preview_features,
             shadow_database_connection_string,
+            reset_shadow_database: false,
         };
 
         let connector = match &connection_info {
diff --git a/schema-engine/sql-migration-tests/src/test_api.rs b/schema-engine/sql-migration-tests/src/test_api.rs
index caee08cd62a..35487d5f6e5 100644
--- a/schema-engine/sql-migration-tests/src/test_api.rs
+++ b/schema-engine/sql-migration-tests/src/test_api.rs
@@ -219,6 +219,7 @@ impl TestApi {
             &DatasourceUrls {
                 url: Some(self.connection_string().to_owned()),
                 shadow_database_url: self.shadow_database_connection_string().map(<_>::to_owned),
+                reset_shadow_database: false,
             },
             params,
         )
diff --git a/schema-engine/sql-migration-tests/tests/errors/error_tests.rs b/schema-engine/sql-migration-tests/tests/errors/error_tests.rs
index 4d8bf0b5fbf..0eb9e25c55d 100644
--- a/schema-engine/sql-migration-tests/tests/errors/error_tests.rs
+++ b/schema-engine/sql-migration-tests/tests/errors/error_tests.rs
@@ -516,6 +516,7 @@ async fn missing_datasource_url_gives_proper_error() {
     let datasource_urls = DatasourceUrls {
         url: None,
         shadow_database_url: None,
+        reset_shadow_database: false,
     };
 
     let mut api = schema_core::schema_api_without_extensions(Some(dm.to_owned()), datasource_urls, None).unwrap();
@@ -563,6 +564,7 @@ async fn diff_from_empty_schema_to_datamodel_should_not_require_url() {
     let datasource_urls = DatasourceUrls {
         url: None,
         shadow_database_url: None,
+        reset_shadow_database: false,
     };
 
     let mut api = schema_core::schema_api_without_extensions(Some(dm.to_owned()), datasource_urls, None).unwrap();
diff --git a/schema-engine/sql-migration-tests/tests/migrations/diff.rs b/schema-engine/sql-migration-tests/tests/migrations/diff.rs
index bfd2d95af21..690d4106c0d 100644
--- a/schema-engine/sql-migration-tests/tests/migrations/diff.rs
+++ b/schema-engine/sql-migration-tests/tests/migrations/diff.rs
@@ -377,6 +377,7 @@ fn from_empty_to_migrations_directory(mut api: TestApi) {
         &DatasourceUrls {
             url: Some("postgres://not-used".to_string()),
             shadow_database_url: Some(api.connection_string().to_owned()),
+            reset_shadow_database: false,
         },
         host.clone(),
         BitFlags::empty(),
@@ -427,6 +428,7 @@ fn from_empty_to_migrations_folder_without_shadow_db_url_must_error(mut api: Tes
             &DatasourceUrls {
                 url: Some(api.connection_string().to_owned()),
                 shadow_database_url: None,
+                reset_shadow_database: false,
             },
             params,
         )
@@ -457,6 +459,7 @@ fn from_migrations_with_the_datasource_database_as_shadow_db_must_error(mut api:
                 &DatasourceUrls {
                     url: Some(main_url.clone()),
                     shadow_database_url: Some(shadow_database_url.clone()),
+                    reset_shadow_database: false,
                 },
                 DiffParams {
                     exit_code: None,
@@ -491,6 +494,7 @@ fn from_url_to_migrations_with_that_url_as_shadow_db_must_error(mut api: TestApi
             &DatasourceUrls {
                 url: None,
                 shadow_database_url: Some(main_url.clone()),
+                reset_shadow_database: false,
             },
             DiffParams {
                 exit_code: None,
@@ -514,6 +518,7 @@ fn from_migrations_with_a_separate_shadow_db_on_the_same_server_works(mut api: T
         DatasourceUrls {
             url: Some(api.connection_string().to_owned()),
             shadow_database_url: Some(api.create_external_shadow_database()),
+            reset_shadow_database: false,
         },
         DiffParams {
             exit_code: Some(true),
@@ -1040,6 +1045,7 @@ fn from_migrations_to_schema_datamodel_ignores_manual_partial_indexes_without_pr
         DatasourceUrls {
             url: Some(api.connection_string().to_owned()),
             shadow_database_url: Some(api.create_external_shadow_database()),
+            reset_shadow_database: false,
         },
         DiffParams {
             exit_code: Some(true),
@@ -1104,6 +1110,7 @@ fn from_schema_datamodel_to_migrations_ignores_manual_partial_indexes_without_pr
         DatasourceUrls {
             url: Some(api.connection_string().to_owned()),
             shadow_database_url: Some(api.create_external_shadow_database()),
+            reset_shadow_database: false,
         },
         DiffParams {
             exit_code: Some(true),
@@ -1179,6 +1186,7 @@ fn from_migrations_to_url_ignores_manual_partial_indexes_with_engine_seeded_sche
         DatasourceUrls {
             url: Some(api.connection_string().to_owned()),
             shadow_database_url: Some(api.create_external_shadow_database()),
+            reset_shadow_database: false,
         },
         DiffParams {
             exit_code: Some(true),
diff --git a/schema-engine/sql-migration-tests/tests/migrations/jsonrpc.rs b/schema-engine/sql-migration-tests/tests/migrations/jsonrpc.rs
index 2bec0465022..336dbf5de7e 100644
--- a/schema-engine/sql-migration-tests/tests/migrations/jsonrpc.rs
+++ b/schema-engine/sql-migration-tests/tests/migrations/jsonrpc.rs
@@ -19,6 +19,7 @@ impl TestApi {
                 DatasourceUrls {
                     url: Some(args.database_url().to_owned()),
                     shadow_database_url: args.shadow_database_url().map(ToOwned::to_owned),
+                    reset_shadow_database: false,
                 },
                 host,
                 Arc::new(ExtensionTypeConfig::default()),
diff --git a/schema-engine/sql-migration-tests/tests/migrations/mssql.rs b/schema-engine/sql-migration-tests/tests/migrations/mssql.rs
index 6842aba1461..16be8531d9c 100644
--- a/schema-engine/sql-migration-tests/tests/migrations/mssql.rs
+++ b/schema-engine/sql-migration-tests/tests/migrations/mssql.rs
@@ -398,6 +398,7 @@ fn apply_migrations_with_a_schema_in_url(mut api: TestApi) {
         connection_string: format!("{};schema=myschema", api.connection_string()),
         preview_features: PreviewFeatures::empty(),
         shadow_database_connection_string: None,
+        reset_shadow_database: false,
     })
     .unwrap();
 
diff --git a/schema-engine/sql-migration-tests/tests/single_migration_tests.rs b/schema-engine/sql-migration-tests/tests/single_migration_tests.rs
index 134e7927a4d..285da4ab7d5 100644
--- a/schema-engine/sql-migration-tests/tests/single_migration_tests.rs
+++ b/schema-engine/sql-migration-tests/tests/single_migration_tests.rs
@@ -79,6 +79,7 @@ fn run_single_migration_test(test_file_path: &str, test_function_name: &'static
             connection_string: test_api_args.database_url().to_owned(),
             preview_features: Default::default(),
             shadow_database_connection_string: None,
+            reset_shadow_database: false,
         };
         let mut conn = SqlSchemaConnector::new_mysql(params).unwrap();
         tok(conn.reset(false, None, &SchemaFilter::default().into())).unwrap();

From 56af9a7d4eceb6e444f991701cd2b120aef72296 Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 14:45:12 +0200
Subject: [PATCH 3/7] fix(schema-engine): read a connection string as its
 flavour when sanitizing it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Which syntax a connection string is written in cannot be inferred from the string
itself: `SQLSERVER://host;database=x;password=secret` is a perfectly good URL
whose entire property list parses as an opaque host, so a URL sanitizer hands the
password straight back, and the scheme check that was meant to catch it only
matched lower case. The flavour decides instead, the way it does for comparing
two connection strings, and every caller knows it.

SQL Server strings are read as JDBC whatever the case of their scheme, and SQLite
strings are read as the file paths they are — a bare path is no URL, and used to
render as the unparseable-connection-string placeholder even though a path holds
nothing to hide.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 .../sql-schema-connector/src/sanitize.rs      | 143 ++++++++++++++----
 1 file changed, 111 insertions(+), 32 deletions(-)

diff --git a/schema-engine/connectors/sql-schema-connector/src/sanitize.rs b/schema-engine/connectors/sql-schema-connector/src/sanitize.rs
index 0fefda2b542..a8b2ac8e784 100644
--- a/schema-engine/connectors/sql-schema-connector/src/sanitize.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/sanitize.rs
@@ -1,6 +1,7 @@
 //! Naming a database in a user-facing message without leaking the credentials that reach it.
 
 use connection_string::JdbcString;
+use psl::datamodel_connector::Flavour;
 use url::Url;
 
 /// How a shadow database reached through a driver adapter is named: there is no connection string
@@ -15,34 +16,41 @@ const UNPARSEABLE_CONNECTION_STRING: &str = "(unparseable connection string)";
 const SECRET_QUERY_PARAMS: &[&str] = &["api_key", "password", "sslpassword"];
 
 /// The JDBC properties that carry credentials. The `connection-string` crate lower-cases property
-/// names while parsing.
+/// names while parsing, so these match however the user spelled them.
 const CREDENTIAL_JDBC_PROPERTIES: &[&str] = &["user", "password"];
 
+const JDBC_PREFIX: &str = "jdbc:";
+
 /// Renders a connection string for a user-facing message: the server, the port and the database
 /// name are kept, so that the reader can tell which database is meant, while the credentials that
 /// reach it — userinfo, secret-bearing query parameters, JDBC credential properties — are dropped.
-pub fn sanitize_connection_string(connection_string: &str) -> String {
-    let sanitized = if is_jdbc_connection_string(connection_string) {
-        sanitize_jdbc_connection_string(connection_string)
-    } else {
-        sanitize_url(connection_string)
+///
+/// The flavour decides how the connection string is read, because the syntaxes are not mutually
+/// exclusive: a SQL Server connection string is a valid URL whose entire property list, password
+/// included, parses as an opaque host, and a SQLite connection string may be a bare file path that
+/// is no URL at all. A connection string that cannot be read as its flavour renders as a
+/// placeholder instead of being echoed back with its secrets in unknown positions.
+pub fn sanitize_connection_string(flavour: Flavour, connection_string: &str) -> String {
+    let sanitized = match flavour {
+        Flavour::Sqlserver => sanitize_jdbc_connection_string(connection_string),
+        Flavour::Sqlite => Some(sqlite_file_path(connection_string)),
+        Flavour::Postgres | Flavour::Cockroach | Flavour::Mysql | Flavour::Mongo => sanitize_url(connection_string),
     };
 
     sanitized.unwrap_or_else(|| UNPARSEABLE_CONNECTION_STRING.to_owned())
 }
 
-/// SQL Server connection strings hold their properties in a `;`-separated list, which the `url`
-/// crate happily parses as part of an opaque host, credentials and all. They are recognized by
-/// their scheme instead.
-fn is_jdbc_connection_string(connection_string: &str) -> bool {
-    connection_string.starts_with("sqlserver:") || connection_string.starts_with("jdbc:sqlserver:")
-}
-
 fn sanitize_jdbc_connection_string(connection_string: &str) -> Option<String> {
-    let with_prefix = if connection_string.starts_with("jdbc:") {
-        connection_string.to_owned()
+    let had_prefix = connection_string
+        .get(..JDBC_PREFIX.len())
+        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(JDBC_PREFIX));
+
+    // `JdbcString` matches the leading `jdbc` literal case-sensitively, and quaint prepends it to
+    // the connection strings that come without it.
+    let with_prefix = if had_prefix {
+        format!("{JDBC_PREFIX}{}", &connection_string[JDBC_PREFIX.len()..])
     } else {
-        format!("jdbc:{connection_string}")
+        format!("{JDBC_PREFIX}{connection_string}")
     };
 
     let mut jdbc: JdbcString = with_prefix.parse().ok()?;
@@ -54,13 +62,24 @@ fn sanitize_jdbc_connection_string(connection_string: &str) -> Option<String> {
 
     let sanitized = jdbc.to_string();
 
-    Some(if connection_string.starts_with("jdbc:") {
+    Some(if had_prefix {
         sanitized
     } else {
-        sanitized.trim_start_matches("jdbc:").to_owned()
+        sanitized[JDBC_PREFIX.len()..].to_owned()
     })
 }
 
+/// A SQLite connection string holds a file path, which carries no credentials to strip. The scheme
+/// and the connection parameters are trimmed the way quaint trims them when it opens the file.
+fn sqlite_file_path(connection_string: &str) -> String {
+    let path = connection_string
+        .strip_prefix("file:")
+        .or_else(|| connection_string.strip_prefix("sqlite:"))
+        .unwrap_or(connection_string);
+
+    path.split('?').next().unwrap_or(path).to_owned()
+}
+
 fn sanitize_url(connection_string: &str) -> Option<String> {
     let mut url: Url = connection_string.parse().ok()?;
 
@@ -98,47 +117,82 @@ mod tests {
     #[test]
     fn credentials_are_stripped_from_urls() {
         assert_eq!(
-            sanitize_connection_string("postgresql://alice:hunter2@db.example.com:5432/shadow"),
+            sanitize_connection_string(
+                Flavour::Postgres,
+                "postgresql://alice:hunter2@db.example.com:5432/shadow"
+            ),
             "postgresql://db.example.com:5432/shadow"
         );
         assert_eq!(
-            sanitize_connection_string("mysql://alice@db.example.com:3306/shadow"),
+            sanitize_connection_string(Flavour::Mysql, "mysql://alice@db.example.com:3306/shadow"),
             "mysql://db.example.com:3306/shadow"
         );
     }
 
+    #[test]
+    fn the_case_of_a_url_scheme_does_not_hide_credentials() {
+        let sanitized = sanitize_connection_string(
+            Flavour::Postgres,
+            "POSTGRESQL://alice:hunter2@db.example.com:5432/shadow",
+        );
+
+        assert!(!sanitized.contains("hunter2"), "{sanitized}");
+        assert!(sanitized.contains("db.example.com:5432/shadow"), "{sanitized}");
+    }
+
     #[test]
     fn secret_query_parameters_are_stripped_and_the_rest_is_kept() {
         assert_eq!(
-            sanitize_connection_string("postgresql://db.example.com/shadow?schema=public&sslmode=require"),
+            sanitize_connection_string(
+                Flavour::Postgres,
+                "postgresql://db.example.com/shadow?schema=public&sslmode=require"
+            ),
             "postgresql://db.example.com/shadow?schema=public&sslmode=require"
         );
         assert_eq!(
-            sanitize_connection_string("postgresql://db.example.com/shadow?sslpassword=hunter2&schema=public"),
+            sanitize_connection_string(
+                Flavour::Postgres,
+                "postgresql://db.example.com/shadow?sslpassword=hunter2&schema=public"
+            ),
             "postgresql://db.example.com/shadow?schema=public"
         );
         assert_eq!(
-            sanitize_connection_string("prisma+postgres://localhost:51213/?api_key=c2VjcmV0"),
+            sanitize_connection_string(Flavour::Postgres, "prisma+postgres://localhost:51213/?api_key=c2VjcmV0"),
             "prisma+postgres://localhost:51213/"
         );
         assert_eq!(
-            sanitize_connection_string("postgresql://db.example.com/shadow?PASSWORD=hunter2"),
+            sanitize_connection_string(Flavour::Postgres, "postgresql://db.example.com/shadow?PASSWORD=hunter2"),
             "postgresql://db.example.com/shadow"
         );
     }
 
     #[test]
-    fn sqlite_paths_are_kept() {
+    fn sqlite_paths_are_rendered_as_paths() {
+        assert_eq!(
+            sanitize_connection_string(Flavour::Sqlite, "file:/tmp/prisma/shadow.db"),
+            "/tmp/prisma/shadow.db"
+        );
+        assert_eq!(
+            sanitize_connection_string(Flavour::Sqlite, "sqlite:/tmp/prisma/shadow.db"),
+            "/tmp/prisma/shadow.db"
+        );
         assert_eq!(
-            sanitize_connection_string("file:/tmp/prisma/shadow.db"),
-            "file:///tmp/prisma/shadow.db"
+            sanitize_connection_string(Flavour::Sqlite, "file:./shadow.db?connection_limit=1"),
+            "./shadow.db"
         );
     }
 
+    #[test]
+    fn a_bare_sqlite_path_is_rendered_as_it_is() {
+        assert_eq!(sanitize_connection_string(Flavour::Sqlite, "shadow.db"), "shadow.db");
+    }
+
     #[test]
     fn credentials_are_stripped_from_sql_server_connection_strings() {
-        let sanitized =
-            sanitize_connection_string("sqlserver://db.example.com:1433;database=shadow;user=SA;password=hunter2");
+        let sanitized = sanitize_connection_string(
+            Flavour::Sqlserver,
+            "sqlserver://db.example.com:1433;database=shadow;user=SA;password=hunter2",
+        );
 
         assert!(sanitized.starts_with("sqlserver://db.example.com:1433"), "{sanitized}");
         assert!(sanitized.contains("database=shadow"), "{sanitized}");
@@ -149,17 +203,42 @@ mod tests {
     #[test]
     fn credentials_are_stripped_from_sql_server_connection_strings_without_a_port() {
         // Without a port, the `url` crate parses the whole property list as an opaque host, so this
-        // is the shape that a URL-only sanitizer would echo back verbatim.
-        let sanitized = sanitize_connection_string("sqlserver://db.example.com;database=shadow;password=hunter2");
+        // is the shape that a URL sanitizer would echo back verbatim.
+        let sanitized = sanitize_connection_string(
+            Flavour::Sqlserver,
+            "sqlserver://db.example.com;database=shadow;password=hunter2",
+        );
 
         assert!(!sanitized.contains("hunter2"), "{sanitized}");
         assert!(sanitized.contains("database=shadow"), "{sanitized}");
     }
 
+    #[test]
+    fn the_case_of_a_sql_server_scheme_does_not_hide_credentials() {
+        for connection_string in [
+            "SQLSERVER://db.example.com;database=shadow;password=hunter2",
+            "SqlServer://db.example.com:1433;database=shadow;PASSWORD=hunter2",
+            "JDBC:sqlserver://db.example.com:1433;database=shadow;password=hunter2",
+            "jdbc:SQLSERVER://db.example.com:1433;database=shadow;user=SA;password=hunter2",
+        ] {
+            let sanitized = sanitize_connection_string(Flavour::Sqlserver, connection_string);
+
+            assert!(!sanitized.contains("hunter2"), "{connection_string}: {sanitized}");
+            assert!(
+                sanitized.contains("database=shadow"),
+                "{connection_string}: {sanitized}"
+            );
+        }
+    }
+
     #[test]
     fn an_unparseable_connection_string_is_not_echoed_back() {
         assert_eq!(
-            sanitize_connection_string("this is not a connection string"),
+            sanitize_connection_string(Flavour::Postgres, "this is not a connection string"),
+            "(unparseable connection string)"
+        );
+        assert_eq!(
+            sanitize_connection_string(Flavour::Sqlserver, "this is not a connection string"),
             "(unparseable connection string)"
         );
     }

From 9c5663008aba7129b8245150ccfdf3b1d1b990d9 Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 14:58:32 +0200
Subject: [PATCH 4/7] feat(schema-engine): only reset a shadow database that is
 empty or consented
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A shadow database provided by the user is reset before the migration history is
replayed into it. Until now the engine did that to whatever it was pointed at, so
a shadow database that had been given a real one's URL, or one holding data
somebody cared about, was emptied without a word.

It is now described first: anything in it — a table, `_prisma_migrations`
included, a view, an enum, a user-defined type, a stored procedure — means the
replay only goes ahead if the user said the contents can be destroyed, and
otherwise the command stops with P3026 naming the database.

After a successful replay the shadow database is reset again, which is what keeps
the promise cheap: a command that replays the history several times finds it
empty every time after the first, and so does the next command, so there is no
consent to remember anywhere. A failed replay leaves its remains behind on
purpose — the next run finding them, and asking, is the signal that it failed.

Both ways in go through one place: the dialect-level funnel, and the local Prisma
Postgres shadow database, which comes with the main database instead of from the
user and used to reach the replay without passing the funnel at all.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 .../src/external_shadow_db.rs                 | 129 ++++++++++++++++++
 .../postgres/connector/native/shadow_db.rs    |  22 ++-
 .../sql-schema-connector/src/lib.rs           |  27 +++-
 schema-engine/sql-schema-describer/src/lib.rs |   4 +
 4 files changed, 175 insertions(+), 7 deletions(-)
 create mode 100644 schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs

diff --git a/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs b/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs
new file mode 100644
index 00000000000..49b50123b32
--- /dev/null
+++ b/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs
@@ -0,0 +1,129 @@
+//! Replaying a migration history into a shadow database the user provided.
+//!
+//! Such a database is reset before the history is replayed into it, which destroys whatever it
+//! holds, so it is only ever used when it holds nothing or when the user said that its contents
+//! can go. It is also left empty afterwards, which is what makes the promise hold for the next
+//! replay: a command that replays the history several times finds an empty database every time
+//! after the first, and so does the command after it.
+
+use crate::flavour::{SqlConnector, UsingExternalShadowDb};
+use schema_connector::{ConnectorError, ConnectorResult, Namespaces, SchemaFilter, migrations_directory::Migrations};
+use sql_schema_describer::SqlSchema;
+use user_facing_errors::schema_engine::ShadowDbNotEmpty;
+
+/// Replays `migrations` into the shadow database `shadow_db` is connected to, and leaves it empty.
+///
+/// `reset_allowed` carries the user's consent to destroy the contents of a shadow database that is
+/// not empty; `location` names the database in the error raised when that consent is missing, and
+/// must be safe to show (see [`crate::sanitize_connection_string`]).
+///
+/// Disposing of `shadow_db` is the caller's business, on this function's success and failure alike.
+pub(crate) async fn replay_migration_history(
+    shadow_db: &mut dyn SqlConnector,
+    migrations: &Migrations,
+    namespaces: Option<Namespaces>,
+    filter: &SchemaFilter,
+    reset_allowed: bool,
+    location: &str,
+) -> ConnectorResult<SqlSchema> {
+    ensure_shadow_db_may_be_reset(shadow_db, namespaces.clone(), reset_allowed, location).await?;
+
+    let schema = shadow_db
+        .sql_schema_from_migration_history(migrations, namespaces.clone(), filter, UsingExternalShadowDb::Yes)
+        .await?;
+
+    shadow_db.reset(namespaces).await?;
+
+    Ok(schema)
+}
+
+/// A failed replay leaves the shadow database as it was when it failed: the next run finding it
+/// dirty, and asking about it, is the signal that something went wrong here.
+async fn ensure_shadow_db_may_be_reset(
+    shadow_db: &mut dyn SqlConnector,
+    namespaces: Option<Namespaces>,
+    reset_allowed: bool,
+    location: &str,
+) -> ConnectorResult<()> {
+    if reset_allowed {
+        return Ok(());
+    }
+
+    let schema = shadow_db.describe_schema(namespaces).await?;
+
+    if holds_no_objects(&schema) {
+        return Ok(());
+    }
+
+    Err(ConnectorError::user_facing(ShadowDbNotEmpty {
+        shadow_database_location: location.to_owned(),
+    }))
+}
+
+/// Whether the described schema holds nothing at all — no table (`_prisma_migrations` included, a
+/// database that only holds one is the remains of an earlier replay), no view, no enum, no
+/// user-defined type, no stored procedure.
+fn holds_no_objects(schema: &SqlSchema) -> bool {
+    schema.tables_count() == 0
+        && schema.views_count() == 0
+        && schema.enum_walkers().len() == 0
+        && schema.udt_walkers().next().is_none()
+        && schema.procedures_count() == 0
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn schema_with_a_namespace() -> SqlSchema {
+        let mut schema = SqlSchema::default();
+        schema.push_namespace("public".to_owned());
+        schema
+    }
+
+    #[test]
+    fn a_schema_without_objects_holds_no_objects() {
+        assert!(holds_no_objects(&SqlSchema::default()));
+        assert!(holds_no_objects(&schema_with_a_namespace()));
+    }
+
+    #[test]
+    fn a_table_counts() {
+        let mut schema = schema_with_a_namespace();
+        schema.push_table("Cat".to_owned(), Default::default(), None);
+
+        assert!(!holds_no_objects(&schema));
+    }
+
+    #[test]
+    fn the_migrations_table_alone_counts() {
+        let mut schema = schema_with_a_namespace();
+        schema.push_table(crate::MIGRATIONS_TABLE_NAME.to_owned(), Default::default(), None);
+
+        assert!(!holds_no_objects(&schema));
+    }
+
+    #[test]
+    fn a_view_counts() {
+        let mut schema = schema_with_a_namespace();
+        schema.push_view("Cats".to_owned(), Default::default(), None, None);
+
+        assert!(!holds_no_objects(&schema));
+    }
+
+    #[test]
+    fn an_enum_counts() {
+        let mut schema = schema_with_a_namespace();
+        schema.push_enum(Default::default(), "Mood".to_owned(), None);
+
+        assert!(!holds_no_objects(&schema));
+    }
+
+    #[test]
+    fn a_user_defined_type_counts() {
+        let mut schema = schema_with_a_namespace();
+        schema.push_udt(Default::default(), "Whiskers".to_owned(), None);
+
+        assert!(!holds_no_objects(&schema));
+    }
+}
diff --git a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/shadow_db.rs b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/shadow_db.rs
index be105d943ff..b1efd96e7ca 100644
--- a/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/shadow_db.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connector/native/shadow_db.rs
@@ -1,6 +1,6 @@
 use crate::flavour::postgres::{MigratePostgresUrl, PpgParams, sql_schema_from_migrations_and_db};
 use crate::flavour::{PostgresConnector, SqlConnector, UsingExternalShadowDb};
-use psl::PreviewFeatures;
+use psl::{PreviewFeatures, datamodel_connector::Flavour};
 use quaint::connector::is_url_localhost;
 use schema_connector::{
     ConnectorError, ConnectorParams, ConnectorResult, Namespaces, SchemaFilter, migrations_directory::Migrations,
@@ -43,6 +43,7 @@ pub async fn sql_schema_from_migration_history(
                 return sql_schema_from_migration_history_for_local_ppg(
                     &shadow_database_url,
                     params.connector_params.preview_features,
+                    params.connector_params.reset_shadow_database,
                     migrations,
                     namespaces,
                     filter,
@@ -143,21 +144,34 @@ async fn sql_schema_from_migration_history_for_external_db(
         .await
 }
 
+/// The shadow database of a local Prisma Postgres comes with the main database rather than from the
+/// user, but it is still a database that is reset and replayed into, so it goes through the same
+/// gate as any other shadow database the engine did not create itself.
 async fn sql_schema_from_migration_history_for_local_ppg(
     url: &Url,
     preview_features: PreviewFeatures,
+    reset_allowed: bool,
     migrations: &Migrations,
     namespaces: Option<Namespaces>,
     filter: &SchemaFilter,
 ) -> Result<SqlSchema, ConnectorError> {
     let ppg_params = PpgParams::parse_from(url)?;
     let shadow_db_url = ppg_params.local_shadow_database_url()?;
+    let location = crate::sanitize_connection_string(Flavour::Postgres, shadow_db_url.as_str());
 
-    let connector_params = ConnectorParams::new(shadow_db_url.to_string(), preview_features, None);
+    let mut connector_params = ConnectorParams::new(shadow_db_url.to_string(), preview_features, None);
+    connector_params.reset_shadow_database = reset_allowed;
     let mut shadow_database = PostgresConnector::new_with_params(connector_params)?;
 
-    let result =
-        sql_schema_from_migration_history_for_external_db(&mut shadow_database, migrations, namespaces, filter).await;
+    let result = crate::external_shadow_db::replay_migration_history(
+        &mut shadow_database,
+        migrations,
+        namespaces,
+        filter,
+        reset_allowed,
+        &location,
+    )
+    .await;
 
     shadow_database.dispose().await?;
 
diff --git a/schema-engine/connectors/sql-schema-connector/src/lib.rs b/schema-engine/connectors/sql-schema-connector/src/lib.rs
index 808387f3647..4de6950352e 100644
--- a/schema-engine/connectors/sql-schema-connector/src/lib.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/lib.rs
@@ -6,6 +6,7 @@
 mod apply_migration;
 mod database_schema;
 mod error;
+mod external_shadow_db;
 mod flavour;
 mod introspection;
 mod migration_pair;
@@ -183,6 +184,20 @@ impl SchemaDialect for SqlSchemaDialect {
                 | ExternalShadowDatabase::ConnectionString { preview_features, .. } => *preview_features,
             };
 
+            let (reset_allowed, location) = match &target {
+                ExternalShadowDatabase::DriverAdapter { reset_allowed, .. } => {
+                    (*reset_allowed, DRIVER_ADAPTER_SHADOW_DATABASE.to_owned())
+                }
+                ExternalShadowDatabase::ConnectionString {
+                    connection_string,
+                    reset_allowed,
+                    ..
+                } => (
+                    *reset_allowed,
+                    sanitize_connection_string(self.dialect.datamodel_connector().flavour(), connection_string),
+                ),
+            };
+
             let mut connector = match target {
                 #[cfg(not(any(
                     feature = "mssql-native",
@@ -214,9 +229,15 @@ impl SchemaDialect for SqlSchemaDialect {
                     ));
                 }
             };
-            let schema = connector
-                .sql_schema_from_migration_history(migrations, namespaces, filter, UsingExternalShadowDb::Yes)
-                .await;
+            let schema = external_shadow_db::replay_migration_history(
+                connector.as_mut(),
+                migrations,
+                namespaces,
+                filter,
+                reset_allowed,
+                &location,
+            )
+            .await;
             // dispose of the connector regardless of the result
             connector.dispose().await?;
             let schema = DatabaseSchema::new(SqlDatabaseSchema::from(schema?));
diff --git a/schema-engine/sql-schema-describer/src/lib.rs b/schema-engine/sql-schema-describer/src/lib.rs
index c731c5236b2..b7e19c0abc8 100644
--- a/schema-engine/sql-schema-describer/src/lib.rs
+++ b/schema-engine/sql-schema-describer/src/lib.rs
@@ -447,6 +447,10 @@ impl SqlSchema {
         self.views.len()
     }
 
+    pub fn procedures_count(&self) -> usize {
+        self.procedures.len()
+    }
+
     pub fn table_walker<'a>(&'a self, name: &str) -> Option<TableWalker<'a>> {
         let table_idx = self.tables.iter().position(|table| table.name == name)?;
         Some(self.walk(TableId(table_idx as u32)))

From 43bc49f56d7a428b375c69fcb1abd17c268541ba Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 14:58:45 +0200
Subject: [PATCH 5/7] chore(schema-engine): silence the unused-variable warning
 in a flavourless build

Every arm of the connection string comparison is behind a flavour feature, so a
build with none of them enabled has nothing left to compare and no use for the
arguments. That build is what `cargo check -p schema-engine-wasm` does by default.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 .../sql-schema-connector/src/same_database.rs         | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/schema-engine/connectors/sql-schema-connector/src/same_database.rs b/schema-engine/connectors/sql-schema-connector/src/same_database.rs
index 9019c72d76a..5a1dd3675eb 100644
--- a/schema-engine/connectors/sql-schema-connector/src/same_database.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/same_database.rs
@@ -20,6 +20,17 @@ pub fn urls_denote_same_database(flavour: Flavour, first: &str, second: &str) ->
 }
 
 /// `None` means "cannot tell", which leaves the decision to the caller of the comparison.
+// With no flavour enabled there is no arm left to compare anything in.
+#[cfg_attr(
+    not(any(
+        feature = "postgresql",
+        feature = "cockroachdb",
+        feature = "mysql",
+        feature = "mssql",
+        feature = "sqlite"
+    )),
+    allow(unused_variables)
+)]
 fn compare_urls(flavour: Flavour, first: &str, second: &str) -> Option<bool> {
     match flavour {
         #[cfg(any(feature = "postgresql", feature = "cockroachdb"))]

From cb64fbf55a8b9200995d5ee26a80d06768256638 Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 16:00:05 +0200
Subject: [PATCH 6/7] fix(schema-engine): reset a consented shadow database,
 and survive not owning it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Consent said the contents of the shadow database can go, and on PostgreSQL,
MySQL and SQL Server they went: those connectors reset an external shadow
database before replaying a migration history into it. SQLite and the Wasm
PostgreSQL connector do not, so the tables the user agreed to part with were
still there when the replay tried to create its own, and the run died on
`table "Cat" already exists` instead of doing what it was told. The reset now
happens where consent is read, so it means the same thing everywhere.

Both resets — the consented one and the one that leaves the database empty
afterwards — fall back to dropping the objects one by one, the way the
flavours' own pre-replay resets already did. A shadow database the engine may
create and drop tables in without owning its schema is a configuration the
tests have covered since long before this, and dropping and recreating the
schema fails there.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 .../src/external_shadow_db.rs                 | 37 ++++++++++++++-----
 1 file changed, 27 insertions(+), 10 deletions(-)

diff --git a/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs b/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs
index 49b50123b32..f681b0834bb 100644
--- a/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs
+++ b/schema-engine/connectors/sql-schema-connector/src/external_shadow_db.rs
@@ -19,36 +19,53 @@ use user_facing_errors::schema_engine::ShadowDbNotEmpty;
 ///
 /// Disposing of `shadow_db` is the caller's business, on this function's success and failure alike.
 pub(crate) async fn replay_migration_history(
-    shadow_db: &mut dyn SqlConnector,
+    shadow_db: &mut (dyn SqlConnector + Send + Sync),
     migrations: &Migrations,
     namespaces: Option<Namespaces>,
     filter: &SchemaFilter,
     reset_allowed: bool,
     location: &str,
 ) -> ConnectorResult<SqlSchema> {
-    ensure_shadow_db_may_be_reset(shadow_db, namespaces.clone(), reset_allowed, location).await?;
+    if reset_allowed {
+        // Consent is exercised here rather than left to the flavour: SQLite and the Wasm PostgreSQL
+        // connector replay a migration history into an external shadow database without resetting
+        // it first, and the contents the user agreed to part with would collide with the replay.
+        reset(shadow_db, namespaces.clone(), filter).await?;
+    } else {
+        ensure_shadow_db_is_empty(shadow_db, namespaces.clone(), location).await?;
+    }
 
     let schema = shadow_db
         .sql_schema_from_migration_history(migrations, namespaces.clone(), filter, UsingExternalShadowDb::Yes)
         .await?;
 
-    shadow_db.reset(namespaces).await?;
+    reset(shadow_db, namespaces, filter).await?;
 
     Ok(schema)
 }
 
+/// Empties the shadow database. A user may well have granted the engine the right to create and to
+/// drop tables in it without making it the owner of anything, which is not enough to drop and
+/// recreate the schema itself, so the objects are dropped one by one when that fails.
+async fn reset(
+    shadow_db: &mut (dyn SqlConnector + Send + Sync),
+    namespaces: Option<Namespaces>,
+    filter: &SchemaFilter,
+) -> ConnectorResult<()> {
+    if shadow_db.reset(namespaces.clone()).await.is_err() {
+        crate::best_effort_reset(shadow_db, namespaces, filter).await?;
+    }
+
+    Ok(())
+}
+
 /// A failed replay leaves the shadow database as it was when it failed: the next run finding it
 /// dirty, and asking about it, is the signal that something went wrong here.
-async fn ensure_shadow_db_may_be_reset(
-    shadow_db: &mut dyn SqlConnector,
+async fn ensure_shadow_db_is_empty(
+    shadow_db: &mut (dyn SqlConnector + Send + Sync),
     namespaces: Option<Namespaces>,
-    reset_allowed: bool,
     location: &str,
 ) -> ConnectorResult<()> {
-    if reset_allowed {
-        return Ok(());
-    }
-
     let schema = shadow_db.describe_schema(namespaces).await?;
 
     if holds_no_objects(&schema) {

From 28948adceee3e5882917894e4a2b40f4f415473e Mon Sep 17 00:00:00 2001
From: Alexey Orlenko's AI Agent <robot@aqrln.net>
Date: Thu, 30 Jul 2026 16:00:20 +0200
Subject: [PATCH 7/7] test(schema-engine): cover the shadow database consent
 gate
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The dirty shadow database is a table the tests put there themselves, standing in
for whatever a real database holds: replaying a migration history into it is
refused with P3026 on the dev path and on the diff path alike, and the table is
still there afterwards, which is the whole point. A shadow database holding
nothing but `_prisma_migrations` is refused too — a migration history in a
database the engine did not just write is either a real database or the remains
of a run that crashed.

The refusal names the database and not the way in: the diff-path test gives the
shadow database a login of its own and asserts that neither the user nor the
password reaches the message.

With consent the same setup goes through and the shadow database is left empty.
SQLite gets its own test for the shape it did not have before — a file holding a
table the migration history also creates is refused, and consented to, reset,
and replayed into cleanly.

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>
---
 .../src/multi_engine_test_api.rs              |  22 +++-
 .../sql-migration-tests/src/utils.rs          |  10 ++
 .../tests/migrations/diff.rs                  |  93 +++++++++++++-
 .../shadow_database_url_configuration.rs      | 121 +++++++++++++++++-
 4 files changed, 241 insertions(+), 5 deletions(-)

diff --git a/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs b/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
index 079c0541e50..6e1a43d1e67 100644
--- a/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
+++ b/schema-engine/sql-migration-tests/src/multi_engine_test_api.rs
@@ -208,6 +208,26 @@ impl TestApi {
         &self,
         connection_string: String,
         shadow_database_connection_string: Option<String>,
+    ) -> ConnectorResult<EngineTestApi> {
+        self.new_engine_with_shadow_db_consent_or_err(connection_string, shadow_database_connection_string, false)
+    }
+
+    /// Instantiate a new migration with the provided connection strings, consenting to the shadow
+    /// database being reset even when it is not empty.
+    pub fn new_engine_with_shadow_db_consent(
+        &self,
+        connection_string: String,
+        shadow_database_connection_string: Option<String>,
+    ) -> EngineTestApi {
+        self.new_engine_with_shadow_db_consent_or_err(connection_string, shadow_database_connection_string, true)
+            .unwrap()
+    }
+
+    fn new_engine_with_shadow_db_consent_or_err(
+        &self,
+        connection_string: String,
+        shadow_database_connection_string: Option<String>,
+        reset_shadow_database: bool,
     ) -> ConnectorResult<EngineTestApi> {
         let connection_info = ConnectionInfo::from_url(&connection_string).unwrap();
 
@@ -215,7 +235,7 @@ impl TestApi {
             connection_string,
             preview_features: self.preview_features,
             shadow_database_connection_string,
-            reset_shadow_database: false,
+            reset_shadow_database,
         };
 
         let connector = match &connection_info {
diff --git a/schema-engine/sql-migration-tests/src/utils.rs b/schema-engine/sql-migration-tests/src/utils.rs
index c6a95334cea..8e7dab657aa 100644
--- a/schema-engine/sql-migration-tests/src/utils.rs
+++ b/schema-engine/sql-migration-tests/src/utils.rs
@@ -6,6 +6,7 @@ use std::{
     path::Path,
 };
 
+use quaint::{prelude::Queryable, single::Quaint};
 use schema_core::json_rpc::types::{
     MigrationDirectory, MigrationFile, MigrationList, MigrationLockfile, SchemaContainer,
 };
@@ -177,3 +178,12 @@ fn shadow_database_name(test_function_name: &str) -> String {
 
     format!("{prefix}_{hash:x}_shadow")
 }
+
+/// Runs a SQL command against an arbitrary database, on a connection of its own. Used to set up and
+/// to inspect databases the engine under test is not connected to, such as a shadow database.
+pub fn raw_cmd_on(connection_string: &str, sql: &str) {
+    tok(async {
+        let connection = Quaint::new(connection_string).await.unwrap();
+        connection.raw_cmd(sql).await.unwrap();
+    })
+}
diff --git a/schema-engine/sql-migration-tests/tests/migrations/diff.rs b/schema-engine/sql-migration-tests/tests/migrations/diff.rs
index 690d4106c0d..e6d67ce3c63 100644
--- a/schema-engine/sql-migration-tests/tests/migrations/diff.rs
+++ b/schema-engine/sql-migration-tests/tests/migrations/diff.rs
@@ -9,10 +9,10 @@ use schema_core::{
 };
 use sql_migration_tests::{
     test_api::*,
-    utils::{list_migrations, to_schema_containers},
+    utils::{list_migrations, raw_cmd_on, to_schema_containers},
 };
 use std::sync::Arc;
-use user_facing_errors::schema_engine::ShadowDbSameAsMainDb;
+use user_facing_errors::schema_engine::{ShadowDbNotEmpty, ShadowDbSameAsMainDb};
 
 #[test_connector(tags(Sqlite, Mysql, Postgres, CockroachDb, Mssql))]
 fn from_unique_index_to_without(mut api: TestApi) {
@@ -541,6 +541,95 @@ fn from_migrations_with_a_separate_shadow_db_on_the_same_server_works(mut api: T
     expected_diff.assert_eq(&diff);
 }
 
+#[test_connector(tags(Postgres), exclude(CockroachDb))]
+fn diff_from_migrations_must_not_reset_a_shadow_db_that_is_not_empty(mut api: TestApi) {
+    // A shadow database with a login of its own: the refusal names the database, and must not carry
+    // the password that reaches it.
+    const SHADOW_DB_USER: &str = "shadowdbconsenttestuser";
+    const SHADOW_DB_PASSWORD: &str = "sh4d0w-p4ssw0rd";
+
+    let migrations_dir = migrations_directory_with_one_migration(&api);
+    let shadow_db_url = api.create_external_shadow_database();
+    raw_cmd_on(&shadow_db_url, "CREATE TABLE dirty_marker (id INTEGER PRIMARY KEY)");
+
+    api.raw_cmd(&format!("DROP USER IF EXISTS {SHADOW_DB_USER}"));
+    api.raw_cmd(&format!(
+        "CREATE USER {SHADOW_DB_USER} PASSWORD '{SHADOW_DB_PASSWORD}' LOGIN"
+    ));
+
+    let shadow_db_url = {
+        let mut url: url::Url = shadow_db_url.parse().unwrap();
+        url.set_username(SHADOW_DB_USER).unwrap();
+        url.set_password(Some(SHADOW_DB_PASSWORD)).unwrap();
+        url.to_string()
+    };
+    let shadow_db_name = url::Url::parse(&shadow_db_url)
+        .unwrap()
+        .path()
+        .trim_start_matches('/')
+        .to_owned();
+
+    let err = api
+        .diff_with_datasource(
+            &DatasourceUrls {
+                url: Some(api.connection_string().to_owned()),
+                shadow_database_url: Some(shadow_db_url.clone()),
+                reset_shadow_database: false,
+            },
+            DiffParams {
+                exit_code: None,
+                from: DiffTarget::Migrations(list_migrations(migrations_dir.path()).unwrap()),
+                to: DiffTarget::Empty,
+                script: false,
+                filters: SchemaFilter::default(),
+            },
+        )
+        .unwrap_err();
+
+    assert!(err.is_user_facing_error::<ShadowDbNotEmpty>(), "{err:?}");
+
+    let message = err.to_string();
+    assert!(message.contains(&shadow_db_name), "{message}");
+    assert!(message.contains("localhost"), "{message}");
+    assert!(!message.contains(SHADOW_DB_PASSWORD), "{message}");
+    assert!(!message.contains(SHADOW_DB_USER), "{message}");
+
+    api.new_engine_with_connection_strings(shadow_db_url, None)
+        .assert_schema()
+        .assert_has_table("dirty_marker");
+}
+
+#[test_connector(tags(Postgres), exclude(CockroachDb))]
+fn diff_from_migrations_resets_a_shadow_db_that_is_not_empty_with_consent(mut api: TestApi) {
+    let migrations_dir = migrations_directory_with_one_migration(&api);
+    let shadow_db_url = api.create_external_shadow_database();
+    raw_cmd_on(&shadow_db_url, "CREATE TABLE dirty_marker (id INTEGER PRIMARY KEY)");
+
+    let result = api
+        .diff_with_datasource(
+            &DatasourceUrls {
+                url: Some(api.connection_string().to_owned()),
+                shadow_database_url: Some(shadow_db_url.clone()),
+                reset_shadow_database: true,
+            },
+            DiffParams {
+                exit_code: Some(true),
+                from: DiffTarget::Empty,
+                to: DiffTarget::Migrations(list_migrations(migrations_dir.path()).unwrap()),
+                script: false,
+                filters: SchemaFilter::default(),
+            },
+        )
+        .unwrap();
+
+    // The migration history was replayed, so the diff against an empty schema is not empty.
+    assert_eq!(result.exit_code, 2);
+
+    api.new_engine_with_connection_strings(shadow_db_url, None)
+        .assert_schema()
+        .assert_tables_count(0);
+}
+
 fn migrations_directory_with_one_migration(api: &TestApi) -> tempfile::TempDir {
     let migrations_dir = tempfile::tempdir().unwrap();
     std::fs::write(
diff --git a/schema-engine/sql-migration-tests/tests/migrations/shadow_database_url_configuration.rs b/schema-engine/sql-migration-tests/tests/migrations/shadow_database_url_configuration.rs
index cf330b371ed..b099452dad5 100644
--- a/schema-engine/sql-migration-tests/tests/migrations/shadow_database_url_configuration.rs
+++ b/schema-engine/sql-migration-tests/tests/migrations/shadow_database_url_configuration.rs
@@ -1,8 +1,12 @@
 use expect_test::expect;
 use quaint::{prelude::Queryable, single::Quaint};
-use sql_migration_tests::multi_engine_test_api::*;
+use sql_migration_tests::{multi_engine_test_api::*, utils::raw_cmd_on};
+use tempfile::TempDir;
 use test_macros::test_connector;
-use user_facing_errors::{UserFacingError, schema_engine::ShadowDbSameAsMainDb};
+use user_facing_errors::{
+    UserFacingError,
+    schema_engine::{ShadowDbNotEmpty, ShadowDbSameAsMainDb},
+};
 
 // exclude: auth works differently in single-node insecure cockroach
 #[test_connector(tags(Postgres), exclude(CockroachDb))]
@@ -292,3 +296,116 @@ fn a_separate_shadow_db_on_the_same_server_is_accepted(api: TestApi) {
         .send_sync()
         .assert_migration_directories_count(1);
 }
+
+const DIRTY_MARKER_TABLE: &str = "CREATE TABLE dirty_marker (id INTEGER PRIMARY KEY)";
+const MIGRATIONS_TABLE_ONLY: &str = "CREATE TABLE _prisma_migrations (id VARCHAR(36) PRIMARY KEY)";
+
+/// A shadow database holding something of its own, as a database somebody else uses would.
+fn dirty_shadow_database(api: &TestApi, setup_sql: &str) -> String {
+    let shadow_db_url = api.create_external_shadow_database();
+    raw_cmd_on(&shadow_db_url, setup_sql);
+    shadow_db_url
+}
+
+/// Plans the second migration of a history, which is the point at which the engine replays the
+/// first one into the shadow database.
+fn migrations_directory_with_one_migration(api: &TestApi) -> TempDir {
+    let migrations_directory = api.create_migrations_directory();
+
+    api.new_engine()
+        .create_migration("01init", CAT_SCHEMA, &migrations_directory)
+        .send_sync();
+
+    migrations_directory
+}
+
+const CAT_SCHEMA: &str = r#"
+    model Cat {
+        id Int @id
+        litterConsumption Int
+    }
+"#;
+
+const CAT_SCHEMA_WITH_ONE_MORE_FIELD: &str = r#"
+    model Cat {
+        id Int @id
+        litterConsumption Int
+        hungry Boolean @default(true)
+    }
+"#;
+
+#[test_connector(tags(Postgres, Mysql, Mssql, Sqlite), exclude(CockroachDb, Vitess))]
+fn a_shadow_db_that_is_not_empty_must_not_be_reset_without_consent(api: TestApi) {
+    let migrations_directory = migrations_directory_with_one_migration(&api);
+    let shadow_db_url = dirty_shadow_database(&api, DIRTY_MARKER_TABLE);
+
+    let err = api
+        .new_engine_with_connection_strings(api.connection_string().to_owned(), Some(shadow_db_url.clone()))
+        .create_migration("02hungry", CAT_SCHEMA_WITH_ONE_MORE_FIELD, &migrations_directory)
+        .send_unwrap_err();
+
+    assert!(err.is_user_facing_error::<ShadowDbNotEmpty>(), "{err:?}");
+
+    // The whole point of refusing: whatever was in there is still in there.
+    api.new_engine_with_connection_strings(shadow_db_url, None)
+        .assert_schema()
+        .assert_has_table("dirty_marker");
+}
+
+#[test_connector(tags(Postgres, Mysql, Mssql, Sqlite), exclude(CockroachDb, Vitess))]
+fn a_shadow_db_holding_only_a_migrations_table_must_not_be_reset_without_consent(api: TestApi) {
+    let migrations_directory = migrations_directory_with_one_migration(&api);
+    let shadow_db_url = dirty_shadow_database(&api, MIGRATIONS_TABLE_ONLY);
+
+    let err = api
+        .new_engine_with_connection_strings(api.connection_string().to_owned(), Some(shadow_db_url))
+        .create_migration("02hungry", CAT_SCHEMA_WITH_ONE_MORE_FIELD, &migrations_directory)
+        .send_unwrap_err();
+
+    assert!(err.is_user_facing_error::<ShadowDbNotEmpty>(), "{err:?}");
+}
+
+#[test_connector(tags(Postgres, Mysql, Mssql, Sqlite), exclude(CockroachDb, Vitess))]
+fn a_shadow_db_that_is_not_empty_is_reset_with_consent_and_left_empty(api: TestApi) {
+    let migrations_directory = migrations_directory_with_one_migration(&api);
+    let shadow_db_url = dirty_shadow_database(&api, DIRTY_MARKER_TABLE);
+
+    api.new_engine_with_shadow_db_consent(api.connection_string().to_owned(), Some(shadow_db_url.clone()))
+        .create_migration("02hungry", CAT_SCHEMA_WITH_ONE_MORE_FIELD, &migrations_directory)
+        .send_sync()
+        .assert_migration_directories_count(2);
+
+    // The shadow database is left as an empty database, so the next command finds nothing to ask
+    // about.
+    api.new_engine_with_connection_strings(shadow_db_url, None)
+        .assert_schema()
+        .assert_tables_count(0);
+}
+
+#[test_connector(tags(Sqlite))]
+fn a_dirty_sqlite_shadow_db_file_is_refused_and_a_consented_one_is_reset(api: TestApi) {
+    let migrations_directory = migrations_directory_with_one_migration(&api);
+
+    // A table the migration history creates too: replaying the history into a file that already
+    // holds it fails, because nothing resets an external SQLite shadow database on the way in.
+    let shadow_db_url = dirty_shadow_database(&api, "CREATE TABLE Cat (id INTEGER PRIMARY KEY)");
+
+    let err = api
+        .new_engine_with_connection_strings(api.connection_string().to_owned(), Some(shadow_db_url.clone()))
+        .create_migration("02hungry", CAT_SCHEMA_WITH_ONE_MORE_FIELD, &migrations_directory)
+        .send_unwrap_err();
+
+    assert!(err.is_user_facing_error::<ShadowDbNotEmpty>(), "{err:?}");
+    api.new_engine_with_connection_strings(shadow_db_url.clone(), None)
+        .assert_schema()
+        .assert_has_table("Cat");
+
+    api.new_engine_with_shadow_db_consent(api.connection_string().to_owned(), Some(shadow_db_url.clone()))
+        .create_migration("02hungry", CAT_SCHEMA_WITH_ONE_MORE_FIELD, &migrations_directory)
+        .send_sync()
+        .assert_migration_directories_count(2);
+
+    api.new_engine_with_connection_strings(shadow_db_url, None)
+        .assert_schema()
+        .assert_tables_count(0);
+}