@@ -1105,21 +1105,45 @@ mod _sqlite3 {
                 ..
             } = args;
 
-            zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize;
-
-            zelf.detect_types.store(detect_types, Ordering::Relaxed);
-            zelf.check_same_thread
-                .store(check_same_thread, Ordering::Relaxed);
-            *zelf.autocommit.lock() = autocommit;
-            *zelf.thread_ident.lock() = std::thread::current().id();
-            let _ = unsafe { zelf.isolation_level.swap(isolation_level.0) };
-
-            let previous = zelf.db.lock().replace(db);
-            Radium::store(&zelf.initialized, true, Ordering::Release);
-            // An old callback's finalizer can reenter __init__ during teardown.
-            // Release any database it installed without holding the Rust lock.
-            with_deferred_drops(|| drop(previous));
-            Ok(())
+            loop {
+                let mut current = zelf.db.lock();
+                if current.is_none() {
+                    let previous = unsafe { zelf.isolation_level.swap(isolation_level.0.clone()) };
+                    if previous.as_ref().is_some_and(|previous| {
+                        !isolation_level
+                            .0
+                            .as_ref()
+                            .is_some_and(|requested| previous.is(requested))
+                    }) {
+                        // A str subclass can finalize too. Keep the requested
+                        // value alive, but retire other values before publication.
+                        drop(current);
+                        with_deferred_drops(|| drop(previous));
+                        continue;
+                    }
+
+                    // No destructor-capable owner remains to release here.
+                    // Publish the settings and database under the same lock.
+                    zelf.statement_cache.lock().capacity = cached_statements.max(0) as usize;
+                    zelf.detect_types.store(detect_types, Ordering::Relaxed);
+                    zelf.check_same_thread
+                        .store(check_same_thread, Ordering::Relaxed);
+                    *zelf.autocommit.lock() = autocommit;
+                    *zelf.thread_ident.lock() = std::thread::current().id();
+                    *current = Some(db);
+                    Radium::store(&zelf.initialized, true, Ordering::Release);
+                    return Ok(());
+                }
+                drop(current);
+
+                // A concurrent initializer or finalizer installed another database.
+                // Finish its teardown outside the lock before trying to publish.
+                with_deferred_drops(|| {
+                    Radium::store(&zelf.initialized, false, Ordering::Release);
+                    zelf.reset_factories(vm);
+                    zelf.drop_db(vm)
+                })?;
+            }
         }
     }
 
 
@@ -964,3 +964,54 @@ def close(resource, finished):
 
 
 blob_close_during_cursor_reset()
+
+
+def reinit_during_isolation_cleanup():
+    import threading
+
+    failures = []
+    armed = False
+
+    class Callback:
+        def __call__(self, sql):
+            pass
+
+        def __del__(self):
+            try:
+                connection.close()
+            except Exception as exc:
+                failures.append(exc)
+
+    def replace():
+        try:
+            connection.__init__(":memory:", check_same_thread=False, autocommit=True)
+            connection.set_trace_callback(Callback())
+        except Exception as exc:
+            failures.append(exc)
+
+    class Level(str):
+        def __del__(self):
+            # CPython releases this subclass during argument conversion;
+            # RustPython retains it until the isolation level is replaced.
+            if armed:
+                worker = threading.Thread(target=replace, daemon=True)
+                worker.start()
+                worker.join(5)
+                if worker.is_alive():
+                    failures.append("reentrant initializer blocked")
+
+    connection = sqlite.connect(
+        ":memory:", isolation_level=Level(""), check_same_thread=False
+    )
+    armed = True
+    try:
+        connection.__init__(":memory:", check_same_thread=False, autocommit=False)
+        assert not failures, failures
+        assert connection.execute("SELECT 1").fetchone() == (1,)
+        assert connection.autocommit is False
+        assert connection.in_transaction
+    finally:
+        connection.close()
+
+
+reinit_during_isolation_cleanup()