diff --git a/tests/conftest.py b/tests/conftest.py
index 1f7f6d9325..bac7ae2da1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -272,8 +272,15 @@ def reset_console():
 
 
 @pytest.fixture
-def duck_conn() -> duckdb.DuckDBPyConnection:
-    return duckdb.connect()
+def duck_conn() -> t.Generator[duckdb.DuckDBPyConnection, None, None]:
+    conn = duckdb.connect()
+    yield conn
+    # Explicitly close so that any extension DLLs loaded during the test are
+    # released before teardown.  On Windows an open DLL handle prevents
+    # install_extension() from moving the newly-downloaded file into place,
+    # causing an "Access is denied" IOException when subsequent tests in the
+    # same xdist worker call install_extension() for the same extension.
+    conn.close()
 
 
 def push_plan(context: Context, plan: Plan) -> None:
diff --git a/tests/core/engine_adapter/test_duckdb.py b/tests/core/engine_adapter/test_duckdb.py
index 696689219b..18caca8547 100644
--- a/tests/core/engine_adapter/test_duckdb.py
+++ b/tests/core/engine_adapter/test_duckdb.py
@@ -126,17 +126,35 @@ def test_drop_catalog_motherduck(make_mocked_engine_adapter: t.Callable) -> None
     assert to_sql_calls(adapter) == ['DROP DATABASE IF EXISTS "foo" CASCADE']
 
 
-def test_ducklake_partitioning(adapter: EngineAdapter, duck_conn, tmp_path):
+@pytest.fixture
+def ducklake_conn(duck_conn):
+    """Load DuckLake without reinstalling it when a previous test already did.
+
+    On Windows, install_extension moves a file into the extension directory.
+    A ducklake.dll that is still mapped from an earlier test in this process
+    makes that move fail with "Access is denied".  Installing only when
+    duckdb_extensions() says it is absent, then calling load_extension, avoids
+    the second install while the DLL is still mapped.
+    """
+    installed = duck_conn.execute(
+        "SELECT installed FROM duckdb_extensions() WHERE extension_name = 'ducklake'"
+    ).fetchone()
+    if not installed or not installed[0]:
+        duck_conn.install_extension("ducklake")
+    duck_conn.load_extension("ducklake")
+    return duck_conn
+
+
+@pytest.mark.xdist_group("ducklake_extension")
+def test_ducklake_partitioning(adapter: EngineAdapter, ducklake_conn, tmp_path):
     catalog = "a_ducklake_db"
 
-    duck_conn.install_extension("ducklake")
-    duck_conn.load_extension("ducklake")
-    duck_conn.execute(
+    ducklake_conn.execute(
         f"ATTACH 'ducklake:{tmp_path}/{catalog}.ducklake' AS {catalog} (DATA_PATH '{tmp_path}');"
     )
 
     # no partitions on catalog creation
-    partition_info = duck_conn.execute(
+    partition_info = ducklake_conn.execute(
         f"SELECT * FROM __ducklake_metadata_{catalog}.main.ducklake_partition_info"
     ).fetchdf()
     assert partition_info.empty
@@ -150,31 +168,30 @@ def test_ducklake_partitioning(adapter: EngineAdapter, duck_conn, tmp_path):
     )
 
     # 1 partition after table creation
-    partition_info = duck_conn.execute(
+    partition_info = ducklake_conn.execute(
         f"SELECT * FROM __ducklake_metadata_{catalog}.main.ducklake_partition_info"
     ).fetchdf()
     assert partition_info.shape[0] == 1
 
 
-def test_drop_table_ducklake_no_cascade(adapter: EngineAdapter, duck_conn, tmp_path):
+@pytest.mark.xdist_group("ducklake_extension")
+def test_drop_table_ducklake_no_cascade(adapter: EngineAdapter, ducklake_conn, tmp_path):
     # DuckLake does not implement DROP TABLE/VIEW ... CASCADE, so the adapter must
     # omit CASCADE for objects in a DuckLake catalog while keeping it for native catalogs.
     catalog = "a_ducklake_db"
 
-    duck_conn.install_extension("ducklake")
-    duck_conn.load_extension("ducklake")
-    duck_conn.execute(
+    ducklake_conn.execute(
         f"ATTACH 'ducklake:{tmp_path}/{catalog}.ducklake' AS {catalog} (DATA_PATH '{tmp_path}');"
     )
 
-    duck_conn.execute(f"CREATE SCHEMA {catalog}.phys")
-    duck_conn.execute(f"CREATE SCHEMA {catalog}.virt")
-    duck_conn.execute(f"CREATE TABLE {catalog}.phys.t (i INTEGER)")
-    duck_conn.execute(f"CREATE VIEW {catalog}.virt.v AS SELECT * FROM {catalog}.phys.t")
+    ducklake_conn.execute(f"CREATE SCHEMA {catalog}.phys")
+    ducklake_conn.execute(f"CREATE SCHEMA {catalog}.virt")
+    ducklake_conn.execute(f"CREATE TABLE {catalog}.phys.t (i INTEGER)")
+    ducklake_conn.execute(f"CREATE VIEW {catalog}.virt.v AS SELECT * FROM {catalog}.phys.t")
 
     # native catalog, cascade is passed through
-    duck_conn.execute("CREATE TABLE memory.main.native_t (i INTEGER)")
-    duck_conn.execute("CREATE VIEW memory.main.native_v AS SELECT * FROM memory.main.native_t")
+    ducklake_conn.execute("CREATE TABLE memory.main.native_t (i INTEGER)")
+    ducklake_conn.execute("CREATE VIEW memory.main.native_v AS SELECT * FROM memory.main.native_t")
 
     adapter.drop_table(f"{catalog}.phys.t", cascade=True)
     adapter.drop_view(f"{catalog}.virt.v", cascade=True)
@@ -187,19 +204,18 @@ def test_drop_table_ducklake_no_cascade(adapter: EngineAdapter, duck_conn, tmp_p
     assert not adapter.table_exists("memory.main.native_v")
 
 
+@pytest.mark.xdist_group("ducklake_extension")
 def test_drop_table_ducklake_no_cascade_mixed_case_alias(
-    adapter: EngineAdapter, duck_conn, tmp_path
+    adapter: EngineAdapter, ducklake_conn, tmp_path
 ):
     # Catalog aliases keep their case in duckdb_databases(), but SQLMesh normalizes
     # table names to lowercase, so the catalog type lookup must be case-insensitive.
-    duck_conn.install_extension("ducklake")
-    duck_conn.load_extension("ducklake")
-    duck_conn.execute(
+    ducklake_conn.execute(
         f"ATTACH 'ducklake:{tmp_path}/my_lake.ducklake' AS \"MyLake\" (DATA_PATH '{tmp_path}');"
     )
-    duck_conn.execute("CREATE SCHEMA MyLake.phys")
-    duck_conn.execute("CREATE TABLE MyLake.phys.t (i INTEGER)")
-    duck_conn.execute("CREATE VIEW MyLake.phys.v AS SELECT * FROM MyLake.phys.t")
+    ducklake_conn.execute("CREATE SCHEMA MyLake.phys")
+    ducklake_conn.execute("CREATE TABLE MyLake.phys.t (i INTEGER)")
+    ducklake_conn.execute("CREATE VIEW MyLake.phys.v AS SELECT * FROM MyLake.phys.t")
 
     adapter.drop_table("mylake.phys.t", cascade=True)
     adapter.drop_view("mylake.phys.v", cascade=True)
diff --git a/tests/core/linter/test_helpers.py b/tests/core/linter/test_helpers.py
index c3ba46f304..7e0c7027ae 100644
--- a/tests/core/linter/test_helpers.py
+++ b/tests/core/linter/test_helpers.py
@@ -7,8 +7,8 @@
 from sqlmesh.core.model import SqlModel
 
 
-def test_get_position_of_model_block():
-    context = Context(paths=["examples/sushi"])
+def test_get_position_of_model_block(copy_to_temp_path):
+    context = Context(paths=copy_to_temp_path("examples/sushi"))
 
     sql_models = [
         model
@@ -40,8 +40,8 @@ def test_get_position_of_model_block():
         assert read_range.endswith(";")
 
 
-def test_get_range_of_a_key_in_model_block_testing_on_sushi():
-    context = Context(paths=["examples/sushi"])
+def test_get_range_of_a_key_in_model_block_testing_on_sushi(copy_to_temp_path):
+    context = Context(paths=copy_to_temp_path("examples/sushi"))
 
     sql_models = [
         model