dmux · Oct 28, 2026 · Oct 26, 2026 · Oct 26, 2026 · Oct 28, 2026
diff --git a/src/minilake/admin.py b/src/minilake/admin.py
@@ -62,7 +62,7 @@ async def ready() -> ReadyResponse:
     return ReadyResponse(ready=True, message="minilake is ready")
 
 
-# Directories under data_dir that `full=True` wipes.
+# Data directories under data_dir that `full=True` wipes.
 #
 # `catalogs`, `volumes`, `warehouses` and `scratch` are the ones that make `full`
 # mean anything: no service `reset()` touches them, so a plain reset clears the
@@ -80,8 +80,8 @@ def _wipe_data_dirs() -> list[str]:
     """Delete the on-disk trees holding user content. Returns what was removed.
 
     Only safe once every service `reset()` has detached its catalogs and the pool
-    has closed its warehouse connections — deleting a database file out from under
-    an open DuckDB handle is what this ordering avoids.
+    has closed its DuckDB connections — deleting database files out from under an
+    open DuckDB handle is what this ordering avoids.
     """
     removed = []
     for name in _WIPED_ON_FULL_RESET:
@@ -122,9 +122,10 @@ async def reset_state(full: bool = False) -> dict[str, str]:
             except Exception as e:
                 logger.warning(f"Failed to reset {service_name}: {e}")
 
-        # Reset warehouse connections
+        # Reset warehouse connections and, for a full reset, remove the shared
+        # Unity Catalog database after service reset hooks have detached catalogs.
         if _duckdb_pool:
-            await _duckdb_pool.reset_all()
+            await _duckdb_pool.reset_all(full=full)
 
         if full:
             removed = _wipe_data_dirs()

diff --git a/src/minilake/duckdb_pool.py b/src/minilake/duckdb_pool.py
@@ -201,11 +201,27 @@ async def close_all(self) -> None:
 
         logger.info("Closed all DuckDB connections")
 
-    async def reset_all(self) -> None:
-        """Reset all warehouse connections (clear in-memory data, keep UC catalog)."""
+    async def reset_all(self, *, full: bool = False) -> None:
+        """Reset warehouse connections and optionally clear the shared UC database.
+
+        A plain state reset preserves Unity Catalog's default database. A full reset
+        closes its connection and removes both the database and its WAL so the next
+        query starts with a fresh database.
+        """
         async with self._warehouse_dict_lock:
             for conn in self._warehouse_connections.values():
                 conn.close()
             self._warehouse_connections.clear()
             self._warehouse_locks.clear()
+
+        if full:
+            async with self._uc_lock:
+                if self._uc_connection is not None:
+                    self._uc_connection.close()
+                    self._uc_connection = None
+
+                uc_path = self.data_dir / "unity_catalog.duckdb"
+                uc_path.unlink(missing_ok=True)
+                Path(str(uc_path) + ".wal").unlink(missing_ok=True)
+
         logger.info("Reset all warehouse connections")
diff --git a/tests/test_admin.py b/tests/test_admin.py
@@ -1,6 +1,7 @@
 """Admin endpoints tests (/health, /ready, /reset, /services)."""
 
 import urllib.request
+from uuid import uuid4
 
 import pytest
 from databricks.sdk import WorkspaceClient
@@ -73,6 +74,14 @@ def test_reset_without_full_keeps_files_on_disk(workspace_client: WorkspaceClien
     without `full`, a volume's directory survives and can be listed again after the
     volume is recreated.
     """
+    warehouse = workspace_client.warehouses.create(name="keep_wh")
+    table_name = f"survives_plain_reset_{uuid4().hex}"
+    created = workspace_client.statement_execution.execute_statement(
+        warehouse_id=warehouse.id,
+        statement=f"CREATE TABLE {table_name} AS SELECT 777 AS value",
+    )
+    assert created.status.state.value == "SUCCEEDED"
+
     workspace_client.catalogs.create(name="keep_cat")
     workspace_client.schemas.create(name="keep_schema", catalog_name="keep_cat")
     workspace_client.volumes.create(
@@ -97,7 +106,15 @@ def test_reset_without_full_keeps_files_on_disk(workspace_client: WorkspaceClien
             volume_type=catalog.VolumeType.MANAGED,
         )
 
-    print("✓ Plain reset left the volume directory on disk")
+    warehouse = workspace_client.warehouses.create(name="keep_wh_after_reset")
+    result = workspace_client.statement_execution.execute_statement(
+        warehouse_id=warehouse.id,
+        statement=f"SELECT value FROM {table_name}",
+    )
+    assert result.status.state.value == "SUCCEEDED"
+    assert result.result.data_array == [[777]]
+
+    print("✓ Plain reset left volume files and the shared SQL database intact")
 
 
 @pytest.mark.crud

diff --git a/tests/test_full_reset_persistence.py b/tests/test_full_reset_persistence.py
@@ -0,0 +1,64 @@
+"""Persistence round-trip test.
+
+Regression test for a real bug found while auditing the project: the
+MINILAKE_PERSIST env var and persistence.py's save_state/load_state existed,
+but were never actually invoked from app.py's lifespan — so setting
+MINILAKE_PERSIST=1 silently did nothing. This test starts an isolated
+minilake subprocess (not the shared session-scoped server used by other
+tests, since this specifically needs a real process restart), creates real
+state, restarts the process against the same data dir, and verifies the
+state survived — using the real databricks-sdk client throughout.
+"""
+
+import shutil
+import tempfile
+import urllib.request
+import uuid
+
+import pytest
+from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors import DatabricksError
+from persistence_helpers import _free_port, _start_minilake, _stop_minilake
+
+
+@pytest.mark.serial
+def test_full_reset_removes_default_database_across_restart():
+    """A full reset must remove unqualified SQL tables from the shared database,
+    including after a process restart against the same data directory."""
+    data_dir = tempfile.mkdtemp(prefix="minilake_full_reset_")
+    port = _free_port()
+    proc = None
+    table_name = f"reset_survivor_{uuid.uuid4().hex}"
+    try:
+        proc = _start_minilake(port, data_dir)
+        client = WorkspaceClient(host=f"http://127.0.0.1:{port}", token="dev", auth_type="pat")
+        warehouse = client.warehouses.create(name="full_reset_before")
+        created = client.statement_execution.execute_statement(
+            warehouse_id=warehouse.id,
+            statement=f"CREATE TABLE {table_name} AS SELECT 777 AS value",
+        )
+        assert created.status.state.value == "SUCCEEDED"
+
+        request = urllib.request.Request(f"http://127.0.0.1:{port}/_minilake/reset?full=true", method="POST")
+        assert urllib.request.urlopen(request, timeout=10).status == 200
+
+        def assert_table_is_gone(client: WorkspaceClient, warehouse_name: str) -> None:
+            warehouse = client.warehouses.create(name=warehouse_name)
+            with pytest.raises(DatabricksError) as exc:
+                client.statement_execution.execute_statement(
+                    warehouse_id=warehouse.id,
+                    statement=f"SELECT value FROM {table_name}",
+                )
+            assert "does not exist" in str(exc.value).lower() or "not found" in str(exc.value).lower()
+
+        assert_table_is_gone(client, "full_reset_immediate")
+
+        _stop_minilake(proc)
+        proc = None
+        proc = _start_minilake(port, data_dir)
+        client = WorkspaceClient(host=f"http://127.0.0.1:{port}", token="dev", auth_type="pat")
+        assert_table_is_gone(client, "full_reset_after_restart")
+    finally:
+        if proc is not None:
+            _stop_minilake(proc)
+        shutil.rmtree(data_dir, ignore_errors=True)