dmux · Oct 28, 2026 · Oct 25, 2026 · Oct 25, 2026 · Oct 25, 2026 · Oct 28, 2026
diff --git a/src/minilake/mcp/tools/composite.py b/src/minilake/mcp/tools/composite.py
@@ -5,11 +5,14 @@
 """
 
 from typing import Any
+from uuid import uuid4
 
 from mcp.server.fastmcp.exceptions import ToolError
 
+from minilake.errors import DatabricksError
 from minilake.mcp.client import MinilakeClient
 from minilake.mcp.tools.sql import ensure_warehouse
+from minilake.uc_types import validate_identifier
 
 _UC = "/api/2.1/unity-catalog"
 _SQL = "/api/2.0/sql"
@@ -88,6 +91,19 @@ async def seed_table(
         await _ensure_schema(client, catalog, schema)
 
         if recreate:
+            # The staging table has a generated, valid name, so its creation cannot
+            # validate the requested target name. Validate that name before dropping
+            # the existing table; SQL-created tables can have names UC rejects.
+            try:
+                validate_identifier(table, "table")
+            except DatabricksError as e:
+                raise ToolError(f"{e.error_code}: {e.message}") from e
+
+            # Validate both the replacement schema and every value before dropping the
+            # current table. Creating/populating a separate managed table exercises the
+            # same UC type validation and DuckDB conversions without risking user data.
+            await _validate_seed_replacement(client, warehouse_id, catalog, schema, columns, rows)
+
             # Drop the physical table too: deleting only the UC metadata would leave the
             # DuckDB table (with its old column list) in place.
             await _run_sql(client, warehouse_id, f"DROP TABLE IF EXISTS {quoted}")
@@ -115,11 +131,12 @@ async def seed_table(
             if "already exists" not in str(e).lower():
                 raise
 
-        await _run_sql(client, warehouse_id, f"DELETE FROM {quoted}")
-
+        statements = ["BEGIN TRANSACTION", f"DELETE FROM {quoted}"]
         if rows:
             values = ", ".join(f"({', '.join(_literal(v) for v in row)})" for row in rows)
-            await _run_sql(client, warehouse_id, f"INSERT INTO {quoted} VALUES {values}")
+            statements.append(f"INSERT INTO {quoted} VALUES {values}")
+        statements.append("COMMIT")
+        await _run_sql(client, warehouse_id, "; ".join(statements))
 
         return {"full_name": full_name, "rows_inserted": len(rows)}
 
@@ -152,6 +169,46 @@ async def _run_sql(client: MinilakeClient, warehouse_id: str, statement: str) ->
     return await client.post(f"{_SQL}/statements", json={"statement": statement, "warehouse_id": warehouse_id})
 
 
+async def _validate_seed_replacement(
+    client: MinilakeClient,
+    warehouse_id: str,
+    catalog: str,
+    schema: str,
+    columns: list[dict[str, str]],
+    rows: list[list[Any]],
+) -> None:
+    """Validate replacement columns and rows in a disposable managed table."""
+    stage_name = f"_minilake_seed_validation_{uuid4().hex}"
+    stage_full_name = f"{catalog}.{schema}.{stage_name}"
+    quoted_stage = f'"{catalog}"."{schema}"."{stage_name}"'
+
+    try:
+        await client.post(
+            f"{_UC}/tables",
+            json={
+                "name": stage_name,
+                "catalog_name": catalog,
+                "schema_name": schema,
+                "columns": columns,
+  
8000
              "table_type": "MANAGED",
+            },
+        )
+        if rows:
+            values = ", ".join(f"({', '.join(_literal(value) for value in row)})" for row in rows)
+            await _run_sql(client, warehouse_id, f"INSERT INTO {quoted_stage} VALUES {values}")
+    except BaseException:
+        # Preserve the validation error while making a best effort to remove any
+        # partially-created staging table. The target has not been touched yet.
+        try:
+            await client.delete(f"{_UC}/tables/{stage_full_name}")
+        except Exception:
+            pass
+        raise
+
+    # If cleanup fails, abort before the target table is modified.
+    await client.delete(f"{_UC}/tables/{stage_full_name}")
+
+
 async def _ensure_catalog(client: MinilakeClient, catalog: str) -> None:
     """Create a catalog unless it already exists."""
     try:

diff --git a/src/minilake/services/sql_statements.py b/src/minilake/services/sql_statements.py
@@ -1269,6 +1269,14 @@ async def _execute_sql_real_typed(
             return columns, column_types, formatted_rows, arrow_table
 
         except Exception as e:
+            # A failed statement inside an explicit transaction leaves DuckDB's shared
+            # connection in an aborted state. Roll it back before releasing the lock so
+            # the failed request cannot poison later statements or strand partial writes.
+            try:
+                conn.execute("ROLLBACK")
+            except Exception:
+                # No active transaction is the normal case for non-transactional errors.
+                pass
             logger.error(f"SQL execution failed: {e}")
             raise DatabricksError(
                 error_code="INVALID_REQUEST",

diff --git a/src/minilake/uc_types.py b/src/minilake/uc_types.py
@@ -60,7 +60,7 @@
     "NULL": ("VARCHAR", "NULL", "void", "void"),
 }
 
-_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
 
 
 @dataclass(frozen=True)
@@ -117,7 +117,7 @@ def validate_identifier(name: str, kind: str) -> str:
        name additionally becomes a filename (`catalogs/{name}.duckdb` in duckdb_pool). A
     name containing a quote or a path separator escapes both.
     """
-    if not name or not _IDENTIFIER_RE.match(name):
+    if not name or not _IDENTIFIER_RE.fullmatch(name):
         raise DatabricksError(
             error_code="INVALID_REQUEST",
             message=(

diff --git a/tests/mcp_server/test_uc_tools.py b/tests/mcp_server/test_uc_tools.py
@@ -122,6 +122,105 @@ async def test_seed_table_surfaces_a_bad_column_type(mcp):
     assert "does not exist" not in message
 
 
+@pytest.mark.parametrize(
+    ("case", "columns", "rows", "recreate", "error_fragment"),
+    [
+        ("invalid_value", [{"name": "id", "type_text": "INTEGER"}], [["not an integer"]], False, "not an integer"),
+        ("invalid_recreate_type", [{"name": "id", "type_text": "NOT_A_TYPE"}], [[43]], True, "not_a_type"),
+        (
+            "invalid_recreate_value",
+            [{"name": "id", "type_text": "INTEGER"}],
+            [["not an integer"]],
+            True,
+            "not an integer",
+        ),
+    ],
+)
+async def test_seed_table_failed_replacement_preserves_existing_table(
+    mcp, case, columns, rows, recreate, error_fragment
+):
+    """A failed replacement must leave both the existing table and its rows intact."""
+    full_name = f"seedpreserve.s.{case}"
+    original_columns = [{"name": "id", "type_text": "INTEGER"}]
+    await mcp.call(
+        "seed_table",
+        {"full_name": full_name, "columns": original_columns, "rows": [[41], [42]]},
+    )
+
+    error = await mcp.call_expecting_error(
+        "seed_table",
+        {"full_name": full_name, "columns": columns, "rows": rows, "recreate": recreate},
+    )
+
+    assert error_fragment in error.lower()
+    result = await mcp.call("run_sql", {"statement": f'SELECT id FROM "seedpreserve"."s"."{case}" ORDER BY id'})
+    assert result["rows"] == [[41], [42]]
+    table = await mcp.call("describe_table", {"full_name": full_name})
+    assert [column["name"] for column in table["columns"]] == ["id"]
+
+
+async def test_seed_table_recreate_invalid_target_preserves_sql_created_table(mcp):
+    """Reject a UC-invalid target name before recreate drops a SQL-created table."""
+    full_name = "seedinvalidname.s.daily-sales"
+    await mcp.call("setup_fixture", {"catalog": "seedinvalidname", "schema": "s"})
+    await mcp.call(
+        "run_sql",
+        {"statement": 'CREATE TABLE "seedinvalidname"."s"."daily-sales" (id INTEGER)'},
+    )
+    await mcp.call(
+        "run_sql",
+        {"statement": 'INSERT INTO "seedinvalidname"."s"."daily-sales" VALUES (41), (42)'},
+    )
+
+    error = await mcp.call_expecting_error(
+        "seed_table",
+        {
+            "full_name": full_name,
+            "columns": [{"name": "id", "type_text": "INTEGER"}],
+            "rows": [[43]],
+            "recreate": True,
+        },
+    )
+
+    assert "invalid table name 'daily-sales'" in error.lower()
+    result = await mcp.call(
+        "run_sql",
+        {"statement": 'SELECT id FROM "seedinvalidname"."s"."daily-sales" ORDER BY id'},
+    )
+    assert result["rows"] == [[41], [42]]
+
+
+async def test_seed_table_recreate_trailing_newline_preserves_sql_created_table(mcp):
+    """Reject a trailing newline before recreate can drop a SQL-created table."""
+    full_name = "seednewline.s.orders\n"
+    await mcp.call("setup_fixture", {"catalog": "seednewline", "schema": "s"})
+    await mcp.call(
+        "run_sql",
+        {"statement": 'CREATE TABLE "seednewline"."s"."orders\n" (id INTEGER)'},
+    )
+    await mcp.call(
+        "run_sql",
+        {"statement": 'INSERT INTO "seednewline"."s"."orders\n" VALUES (41), (42)'},
+    )
+
+    error = await mcp.call_expecting_error(
+        "seed_table",
+        {
+            "full_name": full_name,
+            "columns": [{"name": "id", "type_text": "INTEGER"}],
+            "rows": [[43]],
+            "recreate": True,
+        },
+    )
+
+    assert "invalid table name" in error.lower()
+    result = await mcp.call(
+        "run_sql",
+        {"statement": 'SELECT id FROM "seednewline"."s"."orders\n" ORDER BY id'},
+    )
+    assert result["rows"] == [[41], [42]]
+
+
 async def test_seed_table_accepts_databricks_types(mcp):
     """Databricks type names are what a model will reach for first."""
     result = await mcp.call(