diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 4c3c3699..1dde59c3 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -134,6 +134,44 @@ and must fail when a required dependency is absent. Inspect fixture
 prerequisites and use distinct resources/evidence paths; do not run against
 production or reset retained services to make tests pass.
 
+### Run the PostgreSQL integration suite
+
+The integration fixtures give every test its own database on the server named by
+`WEAVE_TEST_DATABASE_URL`:
+
+- A test that needs the current schema clones a template database,
+  `weave_b1_template_<hash>`. The first test of a run finds that template, or
+  builds it by running the migrations when it is missing. Its name is a hash of
+  the migration files, the code that runs them and the Alembic version, so
+  changing a migration builds a new template. The template stays on the server
+  between runs and holds no test data.
+- A build drops other templates only when nobody is connected to them and no
+  run has used them for 6 hours, so checkouts with different migrations can
+  share one server.
+- A test about migrations themselves (the `empty_settings` and
+  `fresh_migration_settings` fixtures) starts from an empty database and runs the
+  real migrations.
+- When a test ends, the databases and login roles it created are dropped. Set
+  `WEAVE_TEST_KEEP_DATABASES=1` to keep them for inspection. Setting
+  `WEAVE_TEST_RESOURCE_MANIFEST` to a file path records each one in that file and
+  keeps them too.
+
+Run the suite on four parallel workers (pytest-xdist). Workers share the
+template, which an advisory lock lets only one of them build:
+
+```sh
+uv run --locked --all-extras pytest tests/integration -n 4 --dist loadfile
+```
+
+`--dist loadfile` keeps each test file on one worker, so a file's module fixture
+(the Kafka broker, for example) starts once. Without `-n` the suite runs serially.
+
+Turning off durability on the test server is optional: with cloned and dropped
+databases it made no measurable difference. If you try it, pass
+`-c fsync=off -c synchronous_commit=off -c full_page_writes=off` only to a
+throwaway server; an unclean stop can corrupt its data, so never to a server
+whose data you keep, such as the Compose development stack.
+
 [Local runtime setup](docs/reference/local-runtime.md) explains provisioning.
 End-to-end fixtures exercise installed wheels, admitted images, identities and
 external-effect receivers; they are acceptance tooling rather than the introductory
diff --git a/pyproject.toml b/pyproject.toml
index 56449261..2d3fc6d7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -92,6 +92,7 @@ dev = [
   "types-jsonschema>=4.26.0",
   "hatchling>=1.28.0",
   "types-regex>=2026.9.10.20260911",
+  "pytest-xdist>=3.8.0,<4",
 ]
 docs = [
     "mkdocs==1.6.1",
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 394b1540..df0d90c5 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -14,10 +14,19 @@
 # Author: Firefly Software Foundation
 # SPDX-License-Identifier: Apache-2.0
 
-"""Explicitly guarded, real PostgreSQL fixtures. Databases are retained, never dropped."""
+"""Explicitly guarded, real PostgreSQL fixtures.
 
+Each test gets its own database. Tests that need the current schema clone it from a template database migrated
+once per schema; tests that exercise migrations start from an empty database and run them for real. Databases
+and login roles a test creates are dropped when it ends, unless WEAVE_TEST_KEEP_DATABASES=1 or
+WEAVE_TEST_RESOURCE_MANIFEST asks to keep them.
+"""
+
+import hashlib
 import os
 import runpy
+import time
+import warnings
 from pathlib import Path
 from uuid import uuid4
 
@@ -39,7 +48,34 @@ def validate_release_receipt(release_backends):
         release_backends["keycloak_endpoint"]()
 
 
+# Databases and roles created by the running test, dropped at teardown by drop_owned_resources.
+_OWNED: list[tuple[str, str]] = []
+# Name of the migrated template database once this process has found or built it, and when it last marked it used.
+_TEMPLATE: list[str] = []
+_TOUCHED: list[float] = []
+TEMPLATE_LOCK = 7260912902
+# Ruling: a build drops another checkout's template only when no session is connected to it and its last
+# recorded use is older than 6 hours — a running process marks its template used when it starts and again at
+# most every 10 minutes while it clones, and the longest serial integration run is about 35 minutes, so a
+# template in use by another checkout is never dropped — cost if wrong: a process idle for over 6 hours while
+# another checkout builds a different schema sees its next clone fail as an unavailable backend (a rerun
+# rebuilds it), and templates from successive migration edits stay on the server for up to 6 hours.
+TEMPLATE_STALE_AFTER = 6 * 60 * 60
+TEMPLATE_TOUCH_EVERY = 10 * 60
+TEMPLATE_USED = "weave-template-used-at="
+
+
+class TemplateBuildError(Exception):
+    """Migrating the template database failed; reported as itself, not as an unavailable backend."""
+
+
+def keep_resources() -> bool:
+    return os.environ.get("WEAVE_TEST_KEEP_DATABASES") == "1" or bool(os.environ.get("WEAVE_TEST_RESOURCE_MANIFEST"))
+
+
 def retain_resource(kind: str, name: str) -> None:
+    """Record a database or role the test created; it is dropped at teardown unless resources are kept."""
+    _OWNED.append((kind, name))
     path = os.environ.get("WEAVE_TEST_RESOURCE_MANIFEST")
     if path:
         import json
@@ -50,6 +86,188 @@ def retain_resource(kind: str, name: str) -> None:
             handle.write(json.dumps({"kind": kind, "name": name, "observed_at": datetime.now(UTC).isoformat()}) + "\n")
 
 
+@pytest.fixture
+def retain():
+    """Record a database or role the test created itself, so it is dropped at teardown like the fixtures' own.
+
+    Tests take this fixture rather than importing ``retain_resource``: ``from conftest import`` resolves to
+    whichever conftest pytest imported last, which is not this one when ``pytest`` or ``pytest tests`` runs.
+    """
+    return retain_resource
+
+
+def control_url():
+    value = os.environ.get("WEAVE_TEST_DATABASE_URL")
+    if not value:
+        pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL using the task-owned Compose backend")
+    url = make_url(value)
+    if (
+        url.drivername != "postgresql+asyncpg"
+        or url.host not in {"127.0.0.1", "localhost"}
+        or url.username != "weave_b1_owner"
+        or url.database != "weave_b1_control"
+        or url.port in {None, 5432, 5434}
+    ):
+        pytest.fail("Refusing unguarded database: use the dedicated Weave integration backend")
+    return url
+
+
+def template_name() -> str:
+    """Name the migrated template after everything that shapes its schema, so any change to it builds a new one.
+
+    The hash covers the migration files, the schema version, the default operations policy the migrations record,
+    the module that defines ``migrate()`` and ``check_schema()``, and the installed Alembic version.
+    """
+    import alembic
+
+    import firefly_weave.persistence.migrations as module
+    from firefly_weave.contracts.operational_policy import OperationsPolicy
+
+    source = Path(module.__file__).resolve().parents[3] / "migrations"
+    source = source if source.is_dir() else Path(module.__file__).with_name("alembic")
+    digest = hashlib.sha256(module.SCHEMA_VERSION.encode())
+    for path in sorted(source.rglob("*.py")):
+        digest.update(path.relative_to(source).as_posix().encode() + b"\0" + path.read_bytes() + b"\0")
+    digest.update(OperationsPolicy().model_dump_json().encode() + b"\0")
+    digest.update(Path(module.__file__).read_bytes() + b"\0")
+    digest.update(alembic.__version__.encode())
+    return "weave_b1_template_" + digest.hexdigest()[:16]
+
+
+async def mark_template_used(connection, name: str) -> None:
+    """Record, in the template's comment, the server time a process last used it."""
+    now = int(await connection.scalar(text("SELECT floor(extract(epoch FROM now()))")))
+    await connection.execute(text(f"COMMENT ON DATABASE \"{name}\" IS '{TEMPLATE_USED}{now}'"))
+    _TOUCHED[:] = [time.monotonic()]
+
+
+async def drop_stale_templates(connection, keep: str) -> None:
+    """Drop other templates nobody is connected to and nobody has used recently; see TEMPLATE_STALE_AFTER."""
+    stale = await connection.scalars(
+        text(
+            "SELECT d.datname FROM pg_database d WHERE d.datname LIKE 'weave\\_b1\\_template\\_%' AND d.datname<>:keep"
+            " AND NOT EXISTS (SELECT 1 FROM pg_stat_activity a WHERE a.datname=d.datname)"
+            " AND coalesce(substring(shobj_description(d.oid,'pg_database') FROM :pattern)::bigint, 0)"
+            " < extract(epoch FROM now()) - :age"
+        ),
+        {"keep": keep, "pattern": f"^{TEMPLATE_USED}([0-9]+)$", "age": TEMPLATE_STALE_AFTER},
+    )
+    for old in stale.all():
+        await connection.execute(text(f'ALTER DATABASE "{old}" IS_TEMPLATE false'))
+        await connection.execute(text(f'DROP DATABASE IF EXISTS "{old}" WITH (FORCE)'))
+
+
+@pytest.fixture
+def template_sweep():
+    """The stale-template sweep a template build runs, with the comment prefix that records a template's last use."""
+    return drop_stale_templates, TEMPLATE_USED, TEMPLATE_STALE_AFTER
+
+
+async def migrated_template(connection, url) -> str:
+    """Find or build the migrated template once; the advisory lock makes parallel workers build it only once."""
+    if _TEMPLATE:
+        if time.monotonic() - _TOUCHED[0] > TEMPLATE_TOUCH_EVERY:
+            await mark_template_used(connection, _TEMPLATE[0])
+        return _TEMPLATE[0]
+    from firefly_weave.persistence.migrations import migrate
+    from firefly_weave.settings import Settings
+
+    name = template_name()
+    await connection.execute(text("SELECT pg_advisory_lock(:key)"), {"key": TEMPLATE_LOCK})
+    try:
+        ready = await connection.scalar(
+            text("SELECT datistemplate FROM pg_database WHERE datname=:name"), {"name": name}
+        )
+        if not ready:
+            # A template is marked only after a complete migration; anything else is a failed build.
+            await connection.execute(text(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)'))
+            await drop_stale_templates(connection, name)
+            await connection.execute(text(f'CREATE DATABASE "{name}"'))
+            settings = Settings(
+                scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False)
+            )
+            try:
+                await migrate(settings)
+            except Exception as error:
+                raise TemplateBuildError("Migrating the template database failed") from error
+            # Clones need a template nobody is connected to.
+            await connection.execute(text(f'ALTER DATABASE "{name}" IS_TEMPLATE true ALLOW_CONNECTIONS false'))
+        await mark_template_used(connection, name)
+    finally:
+        await connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": TEMPLATE_LOCK})
+    _TEMPLATE.append(name)
+    return name
+
+
+async def create_database(*, migrated: bool):
+    from firefly_weave.settings import Settings
+
+    url = control_url()
+    admin = create_async_engine(url, isolation_level="AUTOCOMMIT")
+    name = f"weave_b1_test_{uuid4().hex}"
+    try:
+        async with admin.connect() as connection:
+            marker = await connection.scalar(text("SELECT identity FROM weave_test_backend_guard"))
+            if marker != "firefly-weave-b1-local-integration":
+                pytest.fail("Refusing database without the task-owned backend marker")
+            template = f' TEMPLATE "{await migrated_template(connection, url)}"' if migrated else ""
+            await connection.execute(text(f'CREATE DATABASE "{name}"{template}'))
+            retain_resource("database", name)
+    except TemplateBuildError:
+        raise
+    except Exception:
+        pytest.fail("Real task-owned PostgreSQL is unavailable or its backend marker is missing", pytrace=False)
+    finally:
+        await admin.dispose()
+    return Settings(scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False))
+
+
+async def drop_resources(connection, owned: list[tuple[str, str]]) -> list[str]:
+    """Drop the databases, then the roles; a failed drop is reported, never a reason to skip the rest.
+
+    Databases go first so their roles no longer hold privileges in them. Returns one line per failed drop.
+    """
+    failures = []
+    for kind in ("database", "role"):
+        for item, name in owned:
+            if item != kind:
+                continue
+            try:
+                await connection.execute(
+                    text(f'DROP {kind.upper()} IF EXISTS "{name}"' + " WITH (FORCE)" * (kind == "database"))
+                )
+            except Exception as error:
+                failures.append(f"{kind} {name}: {type(error).__name__}")
+    return failures
+
+
+@pytest.fixture
+def resource_dropper():
+    """The teardown drop step, for tests of the cleanup itself."""
+    return drop_resources
+
+
+@pytest.fixture(autouse=True)
+async def drop_owned_resources():
+    """Drop what the test created and warn about anything left; a leak is reported, never turned into a failure."""
+    start = len(_OWNED)
+    yield
+    owned = _OWNED[start:]
+    del _OWNED[start:]
+    if not owned or keep_resources():
+        return
+    admin = create_async_engine(control_url(), isolation_level="AUTOCOMMIT")
+    try:
+        async with admin.connect() as connection:
+            failures = await drop_resources(connection, owned)
+    except Exception as error:
+        failures = [f"{kind} {name}: {type(error).__name__}" for kind, name in owned]
+    finally:
+        await admin.dispose()
+    if failures:
+        warnings.warn("Test resources were not dropped: " + "; ".join(failures), stacklevel=1)
+
+
 @pytest.fixture
 def services():
     """Resolve transaction tests through the production service scan without owning a pool."""
@@ -98,44 +316,25 @@ def snapshot(self):
 
 @pytest.fixture
 async def empty_settings():
-    value = os.environ.get("WEAVE_TEST_DATABASE_URL")
-    if not value:
-        pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL using the task-owned Compose backend")
-    url = make_url(value)
-    if (
-        url.drivername != "postgresql+asyncpg"
-        or url.host not in {"127.0.0.1", "localhost"}
-        or url.username != "weave_b1_owner"
-        or url.database != "weave_b1_control"
-        or url.port in {None, 5432, 5434}
-    ):
-        pytest.fail("Refusing unguarded database: use the dedicated Weave integration backend")
-    admin = create_async_engine(url, isolation_level="AUTOCOMMIT")
-    name = f"weave_b1_test_{uuid4().hex}"
-    try:
-        async with admin.connect() as connection:
-            marker = await connection.scalar(text("SELECT identity FROM weave_test_backend_guard"))
-            if marker != "firefly-weave-b1-local-integration":
-                pytest.fail("Refusing database without the task-owned backend marker")
-            await connection.execute(text(f'CREATE DATABASE "{name}"'))
-            retain_resource("database", name)
-    except Exception:
-        pytest.fail("Real task-owned PostgreSQL is unavailable or its backend marker is missing", pytrace=False)
-    finally:
-        await admin.dispose()
-    from firefly_weave.settings import Settings
-
-    return Settings(scheduler_enabled=False, database_url=url.set(database=name).render_as_string(hide_password=False))
+    """An empty database, for tests that run migrations themselves."""
+    return await create_database(migrated=False)
 
 
 @pytest.fixture
-async def migration_settings(empty_settings):
+async def fresh_migration_settings(empty_settings):
+    """An empty database migrated by this test, for tests about migrate() itself."""
     from firefly_weave.persistence.migrations import migrate
 
     await migrate(empty_settings)
     return empty_settings
 
 
+@pytest.fixture
+async def migration_settings():
+    """A database at the current schema, cloned from the migrated template, with owner credentials."""
+    return await create_database(migrated=True)
+
+
 @pytest.fixture
 async def db(settings):
     engine = create_async_engine(settings.database_url.get_secret_value())
diff --git a/tests/integration/connectors/test_postgresql.py b/tests/integration/connectors/test_postgresql.py
index 006f3552..49d263b8 100644
--- a/tests/integration/connectors/test_postgresql.py
+++ b/tests/integration/connectors/test_postgresql.py
@@ -35,7 +35,7 @@
 
 
 @pytest.fixture
-async def external_database(release_backends):
+async def external_database(release_backends, retain):
     value = os.environ.get("WEAVE_TEST_DATABASE_URL")
     if not value:
         pytest.fail("Real PostgreSQL required: set WEAVE_TEST_DATABASE_URL for colima-weave-tests", pytrace=False)
@@ -63,11 +63,13 @@ async def external_database(release_backends):
     roles = {kind: f"d1_{kind}_{suffix}" for kind in ("read", "write")}
     try:
         await admin.execute(f'CREATE DATABASE "{name}"')
+        retain("database", name)
         for kind, role in roles.items():
             await admin.execute(
                 f"CREATE ROLE {role} LOGIN PASSWORD '{password}' NOSUPERUSER NOCREATEDB "
                 f"NOCREATEROLE NOINHERIT NOBYPASSRLS"
             )
+            retain("role", role)
             if kind == "read":
                 await admin.execute(f"ALTER ROLE {role} SET default_transaction_read_only=on")
     finally:
diff --git a/tests/integration/test_boot.py b/tests/integration/test_boot.py
index 2482756e..ec24b237 100644
--- a/tests/integration/test_boot.py
+++ b/tests/integration/test_boot.py
@@ -268,13 +268,13 @@ async def test_readiness_real_connection_failure_is_redacted(settings, migration
         await admin.dispose()
 
 
-async def test_explicit_migration_is_idempotent(migration_settings):
+async def test_explicit_migration_is_idempotent(fresh_migration_settings):
     from sqlalchemy.ext.asyncio import create_async_engine
 
     from firefly_weave.persistence.migrations import migrate
 
-    await migrate(migration_settings)
-    engine = create_async_engine(migration_settings.database_url.get_secret_value())
+    await migrate(fresh_migration_settings)
+    engine = create_async_engine(fresh_migration_settings.database_url.get_secret_value())
     try:
         async with engine.connect() as connection:
             assert await connection.scalar(text("SELECT count(*) FROM weave_schema_version")) == 1
diff --git a/tests/integration/test_incident_migration.py b/tests/integration/test_incident_migration.py
index eb98c792..c31e99af 100644
--- a/tests/integration/test_incident_migration.py
+++ b/tests/integration/test_incident_migration.py
@@ -39,7 +39,7 @@ def apply(sync):
     await connection.run_sync(apply)
 
 
-async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settings):
+async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settings, retain):
     engine = create_async_engine(empty_settings.database_url.get_secret_value())
     role = "weave_c2_migration_" + uuid4().hex
     app = "weave_c2_app_" + uuid4().hex
@@ -49,6 +49,8 @@ async def test_forward_backfill_nonsuperuser_force_rls_and_rollback(empty_settin
             await upgrade(connection, "0009_triggers")
             await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS"))
             await connection.execute(text(f"CREATE ROLE {app} NOLOGIN NOSUPERUSER NOBYPASSRLS IN ROLE weave_app"))
+            retain("role", role)
+            retain("role", app)
             await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}"))
             await connection.execute(text(f"GRANT SELECT ON runs,principals TO {role}"))
             await connection.execute(text(f"GRANT REFERENCES ON runs,principals TO {role}"))
diff --git a/tests/integration/test_operations.py b/tests/integration/test_operations.py
index 7ecf7825..e438d448 100644
--- a/tests/integration/test_operations.py
+++ b/tests/integration/test_operations.py
@@ -729,11 +729,10 @@ async def test_inventory_checks_its_own_role_and_endpoint_with_scheduler_disable
         await service.close()
 
 
-async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access_db, scheduler_url):
+async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access_db, scheduler_url, retain):
     import secrets
     from types import SimpleNamespace
 
-    from conftest import retain_resource
     from sqlalchemy import make_url
 
     from firefly_weave.connections.registry import ConnectorRegistry
@@ -743,7 +742,7 @@ async def test_inventory_rejects_column_only_privilege_on_new_owned_login(access
     username, password = "weave_d4_column_" + uuid4().hex, secrets.token_urlsafe(32)
     async with access_db[1].begin() as owner:
         await owner.execute(text(f"CREATE ROLE {username} LOGIN PASSWORD '{password}' IN ROLE weave_scheduler"))
-        retain_resource("role", username)
+        retain("role", username)
         await owner.execute(text(f"GRANT SELECT(state) ON runs TO {username}"))
         assert not await owner.scalar(text("SELECT has_table_privilege(:role,'runs','SELECT')"), {"role": username})
         assert await owner.scalar(text("SELECT has_any_column_privilege(:role,'runs','SELECT')"), {"role": username})
diff --git a/tests/integration/test_resource_cleanup.py b/tests/integration/test_resource_cleanup.py
new file mode 100644
index 00000000..a126c115
--- /dev/null
+++ b/tests/integration/test_resource_cleanup.py
@@ -0,0 +1,51 @@
+# Copyright 2026 Firefly Software Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# Author: Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
+
+"""The integration fixtures' teardown drops everything a test created, even when one drop fails."""
+
+
+class Connection:
+    def __init__(self, failing: set[str]):
+        self.failing = failing
+        self.statements: list[str] = []
+
+    async def execute(self, statement):
+        sql = str(statement)
+        self.statements.append(sql)
+        if any(name in sql for name in self.failing):
+            raise RuntimeError("dependent objects still exist")
+
+
+async def test_every_drop_is_attempted_and_every_failure_reported(resource_dropper):
+    owned = [("role", "role_a"), ("database", "db_a"), ("database", "db_b"), ("role", "role_b")]
+    connection = Connection({"db_a", "role_a"})
+
+    failures = await resource_dropper(connection, owned)
+
+    assert connection.statements == [
+        'DROP DATABASE IF EXISTS "db_a" WITH (FORCE)',
+        'DROP DATABASE IF EXISTS "db_b" WITH (FORCE)',
+        'DROP ROLE IF EXISTS "role_a"',
+        'DROP ROLE IF EXISTS "role_b"',
+    ]
+    assert failures == ["database db_a: RuntimeError", "role role_a: RuntimeError"]
+
+
+async def test_no_failure_reports_nothing(resource_dropper):
+    connection = Connection(set())
+
+    assert await resource_dropper(connection, [("database", "db_a"), ("role", "role_a")]) == []
+    assert len(connection.statements) == 2
diff --git a/tests/integration/test_schedule_migration.py b/tests/integration/test_schedule_migration.py
index ea24d60f..84901998 100644
--- a/tests/integration/test_schedule_migration.py
+++ b/tests/integration/test_schedule_migration.py
@@ -26,13 +26,14 @@
 pytestmark = pytest.mark.integration
 
 
-async def test_schedule_migration_requires_more_than_catalog_execute(empty_settings):
+async def test_schedule_migration_requires_more_than_catalog_execute(empty_settings, retain):
     engine = create_async_engine(empty_settings.database_url.get_secret_value())
     role = "weave_c3_migration_" + uuid4().hex
     try:
         async with engine.begin() as connection:
             await upgrade(connection, "0010_incidents")
             await connection.execute(text(f"CREATE ROLE {role} NOLOGIN NOSUPERUSER NOBYPASSRLS"))
+            retain("role", role)
             await connection.execute(text(f"GRANT USAGE,CREATE ON SCHEMA public TO {role}"))
             await connection.execute(text(f"GRANT SELECT,UPDATE ON weave_schema_version,alembic_version TO {role}"))
             await connection.execute(text(f"GRANT EXECUTE ON FUNCTION weave_tenant_ids() TO {role}"))
diff --git a/tests/integration/test_template_cleanup.py b/tests/integration/test_template_cleanup.py
new file mode 100644
index 00000000..9935a656
--- /dev/null
+++ b/tests/integration/test_template_cleanup.py
@@ -0,0 +1,58 @@
+# Copyright 2026 Firefly Software Foundation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# Author: Firefly Software Foundation
+# SPDX-License-Identifier: Apache-2.0
+
+"""A template build drops only the other templates nobody is using."""
+
+import os
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import make_url, text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+pytestmark = pytest.mark.integration
+
+
+async def test_build_keeps_templates_in_use_or_recently_used(empty_settings, retain, template_sweep):
+    sweep, used, stale_after = template_sweep
+    control = make_url(os.environ["WEAVE_TEST_DATABASE_URL"])
+    prefix = "weave_b1_template_" + uuid4().hex[:8]
+    names = {kind: f"{prefix}_{kind}" for kind in ("recent", "connected", "old", "unmarked")}
+    admin = create_async_engine(control, isolation_level="AUTOCOMMIT")
+    try:
+        async with admin.connect() as connection:
+            now = int(await connection.scalar(text("SELECT floor(extract(epoch FROM now()))")))
+            for kind, name in names.items():
+                await connection.execute(text(f'CREATE DATABASE "{name}"'))
+                retain("database", name)
+                if kind != "unmarked":
+                    seen = now - 60 if kind == "recent" else now - stale_after - 60
+                    await connection.execute(text(f"COMMENT ON DATABASE \"{name}\" IS '{used}{seen}'"))
+            holder = create_async_engine(control.set(database=names["connected"]))
+            try:
+                async with holder.connect() as held:
+                    await held.execute(text("SELECT 1"))
+                    await sweep(connection, "weave_b1_template_none")
+            finally:
+                await holder.dispose()
+            left = set(
+                await connection.scalars(
+                    text("SELECT datname FROM pg_database WHERE datname LIKE :prefix"), {"prefix": prefix + "%"}
+                )
+            )
+    finally:
+        await admin.dispose()
+    assert left == {names["recent"], names["connected"]}
diff --git a/uv.lock b/uv.lock
index 8b0f5556..a35c9646 100644
--- a/uv.lock
+++ b/uv.lock
@@ -704,6 +704,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/99/89/87ef49ffe383ef4e147d27b7bf2088fb0b54ea409dd87b5a89442e5828a5/cryptography-50.0.1-cp39-abi3-win_amd64.whl", hash = "sha256:55d16b1ef3ee0958d893a977b19777887e546c9954ea81b200c3301a864013f2", size = 3875429, upload-time = "2026-08-25T19:45:24.418Z" },
 ]
 
+[[package]]
+name = "execnet"
+version = "2.1.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" },
+]
+
 [[package]]
 name = "firefly-weave"
 version = "0.1.0a15"
@@ -768,6 +777,7 @@ dev = [
     { name = "mypy" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
+    { name = "pytest-xdist" },
     { name = "ruff" },
     { name = "types-jsonschema" },
     { name = "types-pyyaml" },
@@ -822,6 +832,7 @@ dev = [
     { name = "mypy", specifier = ">=1.20.2,<2" },
     { name = "pytest", specifier = ">=9.0.3,<10" },
     { name = "pytest-asyncio", specifier = ">=1.3.0,<2" },
+    { name = "pytest-xdist", specifier = ">=3.8.0,<4" },
     { name = "ruff", specifier = ">=0.15.12,<1" },
     { name = "types-jsonschema", specifier = ">=4.26.0" },
     { name = "types-pyyaml", specifier = ">=6.0.12" },
@@ -2538,6 +2549,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" },
 ]
 
+[[package]]
+name = "pytest-xdist"
+version = "3.8.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "execnet" },
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
diff --git a/workers/agentic/uv.lock b/workers/agentic/uv.lock
index e6b089c7..4442df69 100644
--- a/workers/agentic/uv.lock
+++ b/workers/agentic/uv.lock
@@ -731,6 +731,7 @@ dev = [
     { name = "mypy", specifier = ">=1.20.2,<2" },
     { name = "pytest", specifier = ">=9.0.3,<10" },
     { name = "pytest-asyncio", specifier = ">=1.3.0,<2" },
+    { name = "pytest-xdist", specifier = ">=3.8.0,<4" },
     { name = "ruff", specifier = ">=0.15.12,<1" },
     { name = "types-jsonschema", specifier = ">=4.26.0" },
     { name = "types-pyyaml", specifier = ">=6.0.12" },
diff --git a/workers/files/uv.lock b/workers/files/uv.lock
index c4334fca..b1d6d5f6 100644
--- a/workers/files/uv.lock
+++ b/workers/files/uv.lock
@@ -291,6 +291,7 @@ dev = [
     { name = "mypy", specifier = ">=1.20.2,<2" },
     { name = "pytest", specifier = ">=9.0.3,<10" },
     { name = "pytest-asyncio", specifier = ">=1.3.0,<2" },
+    { name = "pytest-xdist", specifier = ">=3.8.0,<4" },
     { name = "ruff", specifier = ">=0.15.12,<1" },
     { name = "types-jsonschema", specifier = ">=4.26.0" },
     { name = "types-pyyaml", specifier = ">=6.0.12" },