Dispose replaced migration database handles

src/govoplan_core/db/migrations.py

@@ -215,7 +215,10 @@ def _registered_module_registry(
     server_config = get_server_config()
     active_database_url = database_url or settings.database_url
     if active_database_url:
-        configure_database(active_database_url)
+        # Registry planning may target a different database than the currently
+        # configured handle. The global handle is replaced either way, so
+        # dispose its superseded pool instead of leaking open DBAPI connections.
+        configure_database(active_database_url, dispose_previous=True)
     active_manifest_factories = manifest_factories or tuple(server_config.manifest_factories)
     raw_enabled_modules = tuple(enabled_modules) if enabled_modules is not None else load_startup_enabled_modules(server_config.enabled_modules)
     candidate_modules = startup_candidate_module_ids(server_config.enabled_modules, raw_enabled_modules)

tests/test_module_system.py

@@ -2040,6 +2040,25 @@ finally:
         self.assertEqual(["check", "verify"], [record["task_id"] for record in records])
         self.assertEqual(["warning", "warning"], [record["status"] for record in records])
 
+    def test_registered_module_migration_tasks_dispose_replaced_database(self) -> None:
+        root = Path(tempfile.mkdtemp(prefix="govoplan-module-migration-dispose-", dir=_TEST_ROOT))
+        previous = configure_database(f"sqlite:///{root / 'previous.db'}")
+        previous_pool = previous.engine.pool
+        manifest = ModuleManifest(id="taskmod", name="Task Module", version="1.0.0")
+
+        try:
+            run_registered_module_migration_tasks(
+                database_url=f"sqlite:///{root / 'target.db'}",
+                enabled_modules=("taskmod",),
+                phases=(),
+                manifest_factories=(lambda: manifest,),
+            )
+
+            self.assertIsNot(previous.engine.pool, previous_pool)
+            self.assertEqual(f"sqlite:///{root / 'target.db'}", get_database().database_url)
+        finally:
+            reset_database(dispose=True)
+
     def test_module_installer_preflight_blocks_provider_update_that_breaks_installed_consumer(self) -> None:
         available = {
             "files": ModuleManifest(