diff --git a/Makefile b/Makefile
index eae8b26..c363ecf 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: help dev-up dev-up-otel dev-down dev-logs dev-status dev-reset test test-all ci ci-clean e2e-test clean verify-stack create-test-account mobile-full-setup
+.PHONY: help dev-up dev-up-otel dev-down dev-logs dev-status dev-reset test test-all test-db-prepare test-audit ci ci-clean e2e-test clean verify-stack create-test-account mobile-full-setup
 
 # Default target - show help
 .DEFAULT_GOAL := help
@@ -127,11 +127,20 @@ test: ## Run fast unit/integration tests (skips slow E2E tests)
 	@sleep 3
 	@echo "$(GREEN)Running migrations on test database...$(RESET)"
 	@goose -dir internal/db/migrations postgres "postgresql://$(POSTGRES_TEST_USER):$(POSTGRES_TEST_PASSWORD)@localhost:$(POSTGRES_TEST_PORT)/$(POSTGRES_TEST_DB)?sslmode=disable" up || true
+	@# Provisions the template database that testkit.DB clones per test, and
+	@# sweeps clones orphaned by killed runs. Additive: the goose line above
+	@# still migrates the shared database the not-yet-migrated tests use.
+	@./scripts/test-db-prepare.sh
 	@echo "$(GREEN)Running fast tests (use 'make e2e-test' for E2E tests)...$(RESET)"
 	@# -p 1 runs packages sequentially: the integration suite's setup wipes
 	@# shared test-DB tables (unscoped DELETEs), so package-parallel runs race
 	@# and randomly kill other packages' fixtures (jetstream DB tests above all).
-	@go test -p 1 ./cmd/... ./internal/... ./tests/... -short -v
+	@#
+	@# -parallel comes from the server's max_connections, because every test
+	@# under t.Parallel() holds its own clone pool. Inert until phase 3 enables
+	@# parallelism; wired now so the ceiling is never discovered by hitting it.
+	@go test -p 1 -parallel $$(./scripts/test-db-prepare.sh --print-parallel) \
+		./cmd/... ./internal/... ./tests/... -short -v
 	@echo "$(GREEN)✓ Tests complete$(RESET)"
 
 e2e-test: ## Run automated E2E tests (requires: make dev-up + make run in another terminal)
@@ -177,6 +186,12 @@ test-db-reset: ## Reset test database
 	@goose -dir internal/db/migrations postgres "postgresql://$(POSTGRES_TEST_USER):$(POSTGRES_TEST_PASSWORD)@localhost:$(POSTGRES_TEST_PORT)/$(POSTGRES_TEST_DB)?sslmode=disable" up || true
 	@echo "$(GREEN)✓ Test database reset$(RESET)"
 
+test-db-prepare: ## Create or refresh the template database that testkit.DB clones per test
+	@./scripts/test-db-prepare.sh
+
+test-audit: ## Count test-suite invariant violations (warn only; -v for file:line)
+	@./scripts/test-audit.sh
+
 test-db-stop: ## Stop test database
 	@docker-compose -f docker-compose.dev.yml --env-file .env.dev --profile test stop postgres-test
 	@echo "$(GREEN)✓ Test database stopped$(RESET)"
diff --git a/docker-compose.ci.yml b/docker-compose.ci.yml
index c2a6a5f..533cff8 100644
--- a/docker-compose.ci.yml
+++ b/docker-compose.ci.yml
@@ -88,7 +88,13 @@ services:
   postgres-test:
     image: postgres:15
     network_mode: "service:netns"
-    command: ["-p", "5434"]
+    # max_connections is raised from the default 100 because tests/testkit
+    # clones a database per test and each clone opens its own small pool. The
+    # ceiling on `go test -parallel` is derived from this number
+    # (testkit.ParallelBudget), so the two move together instead of the suite
+    # discovering the limit as "sorry, too many clients already" in whichever
+    # test happened to be unlucky.
+    command: ["-p", "5434", "-c", "max_connections=200"]
     environment:
       POSTGRES_DB: coves_test
       POSTGRES_USER: test_user
diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml
index c6d9f5e..646c87a 100644
--- a/docker-compose.dev.yml
+++ b/docker-compose.dev.yml
@@ -38,6 +38,11 @@ services:
   postgres-test:
     image: postgres:15
     container_name: coves-test-postgres
+    # Matches docker-compose.ci.yml: tests/testkit clones a database per test,
+    # each with its own pool, and testkit.ParallelBudget derives the safe
+    # `go test -parallel` value from this number. See the CI file for the full
+    # reasoning.
+    command: ["-c", "max_connections=200"]
     ports:
       - "${POSTGRES_TEST_PORT:-5434}:5432"
     environment:
diff --git a/loop_state.md b/loop_state.md
index c84a44e..8efe0f0 100644
--- a/loop_state.md
+++ b/loop_state.md
@@ -39,7 +39,7 @@ Stop the loop when every task is done, or on any blocked task.
 |---|------|-------|---|--------|--------|-------|
 | 1 | Gate green as imported: run `make ci`, fix what surfaces, record baseline timing + allowlist | 0 ⛩ | S | done | (no diff) | GREEN FIRST RUN: 3307 tests, 3288 pass, 0 fail, 19 skips (all allowlisted, 21 entries → 2 unused are ~conditional), 2.2 min WARM caches. No fixes → no review stream |
 | 2 | Move public-network tests to tests/live/ (+`live` tag, move-only); flip compose nets `internal: true` + module-cache pre-pull; cold-cache egress-blocked `make ci` green | 0 ⛩ | S | done | (see git log) | COLD egress-blocked GREEN 2:21; warm 1:57. 16 funcs → tests/live (4 files + helpers). Egress block found 4 runtime deps, not 1: Turnstile URL hardcoded (→ WithSiteverifyURL, dev-gated env), PLC healthcheck redirected to public web (→ /_health), 3 blueskypost tests dialed public.api.bsky.app (→ blueskyAPI seam), DNS-dependent 404 test (→ DID). Review: Codex "good" + Opus "safe as-is"; 9 fixes applied (unfurl E2E de-mocked via httptest OG, gate vets -tags live, non-dev override warns, prod default pinned, stub 127.0.0.1-only, GOPROXY=off, 404/500 handler tests, golden 200 parse, live cache purge+method asserts). 3281 tests / 14 allowlisted skips |
-| 3 | testkit core: db.go (template-clone, advisory lock), wait.go (WaitFor/Holds, terminal errs), fixtures.go (UniqueID run-prefix), scripts/test-db-prepare.sh, scripts/test-audit.sh (warn mode) + testkit's own tests | 1 | S | pending | | dependency rule: testkit imports NO internal/core/* |
+| 3 | testkit core: db.go (template-clone, advisory lock), wait.go (WaitFor/Holds, terminal errs), fixtures.go (UniqueID run-prefix), scripts/test-db-prepare.sh, scripts/test-audit.sh (warn mode) + testkit's own tests | 1 | S | done | (see git log) | 47 testkit tests, -race -shuffle clean; clone ~30ms (cheaper than spec est). Review: Codex needs-work (5 high) + Opus not-safe-as-is (2 high) → 15-item batch applied: template-drop name rail (found a panic-in-error-path bug), drop-before-close teardown, 55006 retry, sweep de-FORCEd + error-accumulating, wait-primitive deadline contract, max_connections=200 + ParallelBudget (CI: -parallel 53, inert until ph.3), bounded lock wait w/ holder diagnostics, grep -H audit fix. ADJUDICATION: Codex RIGHT / Opus WRONG on advisory-lock leak (async cancel can grant lock after Go sees error; fixed via Conn.Raw(ErrBadConn) eviction). make ci GREEN 3347 tests / 14 allowlisted skips |
 | 4 | testkit pds.go (absorb 4 factories, createPDSAccount×2, XRPC clients), firehose.go (generic cursor-gated), appview.go; fix 5 handle-collision sites | 1 ⛩ | S | pending | | then FULL PANEL review of tests/testkit (incl. pragma:security — PDS creds) |
 | 5 | Kill the lies: delete 6 debt tests; lexicon validator stops generating defs-only subtests (retire 8 allowlist entries); move 2 ratelimit files to internal/api/middleware (T0); fold tests/unit into internal/core/communities | 2 | M | pending | | |
 | 6 | Split multi-tier files by test func (manifest in commit msg); add build tags in place; retarget Makefile to tags; delete -short/testing.Short(); delete test-all | 2 ⛩ | S | pending | | identity_resolution, bluesky_post (what's left post-task-2), post_unfurl |
@@ -105,3 +105,29 @@ Stop the loop when every task is done, or on any blocked task.
   stage 1b `go vet -tags live` so the live tier can't rot invisibly.
 - **Docker flake**: containerd "failed to prepare extraction snapshot" can
   appear once on build; retry clears it — don't chase it as a regression.
+- **From task 3 (testkit — later phases MUST know)**: testkit.DB(t) is the
+  ONLY sanctioned DB path for migrated tests; template `coves_test_template`
+  guarded by validateTemplateName (won't drop non-test-shaped names).
+  `sql.DB.Close` WAITS on in-flight queries — always DROP (bounded ctx,
+  FORCE) before Close in teardown. `Conn.Raw(func(any) error { return
+  driver.ErrBadConn })` is the ONLY way to evict a session from a
+  database/sql pool — required on every error path of anything holding
+  session state (advisory locks, SET LOCAL, temp tables). CREATE DATABASE
+  ... TEMPLATE can hit SQLSTATE 55006 for a few ms after a template
+  connection closes (async backend exit) — cloneTemplate retries 3×.
+  Advisory lock key ('COVE',1) on the maintenance DB (coves_test); template
+  peeks need EXCLUSIVE (a source connection blocks cloning). goose: testkit
+  uses NewProvider (no globals); hazard only if a package mixes legacy
+  goose.Up with testkit in one binary (tasks 7-8 watch for it).
+  testkit.ParallelBudget wired into both go test invocations via
+  test-db-prepare.sh --print-parallel (CI computes 53); inert until phase 3
+  drops -p 1. Dev postgres-test won't see max_connections=200 until
+  recreated FROM THE MAIN CHECKOUT (worktree compose project-name mismatch
+  vs pinned container_name); CI unaffected. Audit baseline 911: sleep 60
+  (ph.3-4), skip 280 (ph.2), Short 162 (ph.2), dialer 30 (ph.4), host:port
+  288 (ph.3-4), public-hosts 91 (ph.2, ~91 exemption annotations needed —
+  big three: blueskypost/url_parser_test 19, aggregator_registration_test
+  14, jetstream/user_consumer_test 10). Testkit tests deliberately untagged
+  until task 6. REVIEWER CALIBRATION: on the lock-leak disagreement Codex
+  was right, Opus wrong (missed the failure path, traced only cancellation)
+  — weight Codex on DB/concurrency semantics.
diff --git a/scripts/ci-runner.sh b/scripts/ci-runner.sh
index cd60f05..0d45a21 100755
--- a/scripts/ci-runner.sh
+++ b/scripts/ci-runner.sh
@@ -74,6 +74,37 @@ go vet -tags live ./tests/live/...
 echo "  ✓ tests/live compiles"
 echo
 
+# ---------------------------------------------------------------------------
+# 1c. Violation audit (advisory)
+# ---------------------------------------------------------------------------
+# docs/TEST_ARCHITECTURE.md §3.6.3. The suite is mid-migration and every count
+# below is scheduled against a phase, so this reports and never judges — the
+# `|| true` is belt-and-braces on top of the script's own exit 0. It becomes the
+# hard lint gate in the final phase, when the counts are zero.
+bash /src/scripts/test-audit.sh || true
+
+# ---------------------------------------------------------------------------
+# 1d. Test template database
+# ---------------------------------------------------------------------------
+# Provisions the migrated template that testkit.DB clones per test. Done here,
+# once, rather than inside the test binaries: `go test` runs packages as
+# separate processes and N of them racing to create the same database is a
+# built-in flake.
+#
+# This ADDS a database. The legacy path — tests/integration running goose
+# against the shared coves_test database — is untouched and still works.
+echo "▶ Preparing the test template database..."
+go run ./tests/testkit/cmd/testdbprepare
+
+# The connection budget, derived from the server's max_connections rather than
+# guessed: every test running under t.Parallel() holds its own clone pool.
+# Nothing uses t.Parallel() outside tests/testkit yet, so this is inert today —
+# it is wired now so that the phase enabling parallelism does not also have to
+# discover the ceiling by exhausting it.
+TEST_PARALLEL=$(bash /src/scripts/test-db-prepare.sh --print-parallel)
+echo "  ✓ connection budget allows -parallel $TEST_PARALLEL"
+echo
+
 # ---------------------------------------------------------------------------
 # 2. Run the suite
 # ---------------------------------------------------------------------------
@@ -119,7 +150,7 @@ progress_pid=$!
 trap 'kill "$progress_pid" 2>/dev/null || true' EXIT
 
 set +e
-go test -json -p 1 -count=1 -timeout "$TEST_TIMEOUT" \
+go test -json -p 1 -parallel "$TEST_PARALLEL" -count=1 -timeout "$TEST_TIMEOUT" \
     ./cmd/... ./internal/... ./tests/... \
     >>"$RAW" 2>&1
 set -e
diff --git a/scripts/test-audit.sh b/scripts/test-audit.sh
new file mode 100755
index 0000000..4411a67
--- /dev/null
+++ b/scripts/test-audit.sh
@@ -0,0 +1,199 @@
+#!/usr/bin/env bash
+# Counts test-suite invariant violations. Progress meter first, lint gate second.
+#
+# WHAT THIS IS FOR
+#
+# docs/TEST_ARCHITECTURE.md §3.6.3. Every category below is something the
+# refactor is removing, and each has a phase that must drive it to zero. Running
+# this after every change turns "will the final enforcement flip pass?" into a
+# number you can watch instead of a hope.
+#
+# WARN MODE. This always exits 0. It prints a table and nothing else, so it can
+# sit in the CI pipeline from day one without failing builds for violations
+# that are scheduled to be fixed three phases from now. The final phase flips it
+# to a hard failure, at which point every count must already be zero.
+#
+# THESE ARE TRIPWIRES, NOT PROOFS. Every check here is a grep, and a grep is
+# bypassable by construction — a URL built by concatenation, a bare IP literal,
+# a sleep hidden behind a helper. They catch drift cheaply. The actual guarantee
+# that tests never reach the public network is the egress-blocked CI network
+# (docker-compose.ci.yml, `internal: true`).
+#
+# SCOPE. "test code" means every *_test.go file under cmd/, internal/ and
+# tests/, plus every .go file under tests/ — the shared helpers in
+# tests/integration are test code too, and some of the worst offenders live
+# there rather than in files ending _test.go.
+#
+# Whole-line comments are not counted. This tree explains itself at length, and
+# a comment that mentions localhost:5434 while describing the stack is not a
+# hardcoded endpoint. It is a deliberate undercount at the margin: a violation
+# trailing a comment on the same line still counts, because the code is there.
+#
+# Usage:
+#   scripts/test-audit.sh        summary table
+#   scripts/test-audit.sh -v     table plus every offending file:line
+set -uo pipefail
+
+REPO_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+cd "$REPO_ROOT"
+
+VERBOSE=0
+if [[ ${1:-} == "-v" || ${1:-} == "--verbose" ]]; then
+    VERBOSE=1
+fi
+
+CYAN='\033[36m'
+YELLOW='\033[33m'
+GREEN='\033[32m'
+RESET='\033[0m'
+
+# ---------------------------------------------------------------------------
+# File scopes
+# ---------------------------------------------------------------------------
+
+test_code_files() {
+    {
+        find cmd internal tests -type f -name '*_test.go'
+        find tests -type f -name '*.go'
+    } 2>/dev/null | sort -u
+}
+
+all_go_files() {
+    find cmd internal tests -type f -name '*.go' 2>/dev/null | sort -u
+}
+
+# ---------------------------------------------------------------------------
+# Scanning
+# ---------------------------------------------------------------------------
+
+TOTAL=0
+declare -a ROWS=()
+declare -a DETAILS=()
+
+# scan <label> <pattern> <scope-fn> <path-exclude-ERE|""> <burns-down-in> [exemption-marker]
+#
+# grep -H is load-bearing: xargs splits its input into batches, and a batch that
+# happens to contain exactly one file makes plain grep omit the filename. The
+# comment-exclusion regex below anchors on "file:line:", so without -H it
+# silently stops matching on those batches and the counts differ between a Mac
+# (one batch) and the Alpine CI runner (several).
+scan() {
+    local label=$1 pattern=$2 scope=$3 exclude=$4 phase=$5 exemption=${6:-}
+    local files hits count
+
+    files=$($scope)
+    if [[ -n $exclude ]]; then
+        files=$(printf '%s\n' "$files" | grep -vE "$exclude" || true)
+    fi
+
+    if [[ -z $files ]]; then
+        hits=""
+    else
+        hits=$(printf '%s\n' "$files" | tr '\n' '\0' |
+            xargs -0 grep -HnE "$pattern" 2>/dev/null |
+            grep -vE '^[^:]+:[0-9]+:[[:space:]]*//' || true)
+        # Scoped to the category that defines it, so an exemption for a
+        # hostname cannot quietly excuse a sleep on the same line.
+        if [[ -n $exemption ]]; then
+            hits=$(printf '%s\n' "$hits" | grep -v "$exemption" || true)
+        fi
+    fi
+
+    count=$(printf '%s' "$hits" | grep -c . || true)
+    TOTAL=$((TOTAL + count))
+    ROWS+=("$(printf '  %-52s %6d   %s' "$label" "$count" "$phase")")
+
+    if [[ $VERBOSE -eq 1 && $count -gt 0 ]]; then
+        DETAILS+=("$label"$'\n'"$hits")
+    fi
+}
+
+echo
+printf "${CYAN}═══════════════════════════════════════════════════════════════════════${RESET}\n"
+printf "${CYAN}  TEST SUITE VIOLATION AUDIT${RESET}   (warn mode — never fails the build)\n"
+printf "${CYAN}═══════════════════════════════════════════════════════════════════════${RESET}\n"
+echo
+
+# 1. Sleeps. A sleep is a guess about timing that is wrong in both directions:
+#    too short under CI load (flake), too long on an idle laptop (slow suite).
+#    testkit.WaitFor / testkit.Holds replace every one of them. testkit itself is
+#    exempt because it is where the polling loop is implemented.
+scan "time.Sleep in test code" \
+    'time\.Sleep\(' \
+    test_code_files 'tests/testkit/' 'phase 3-4'
+
+# 2. Skips. cmd/ci-report treats a skip as a failure unless it is allowlisted
+#    with a reason; the allowlist shrinks to near-zero rather than absorbing
+#    these. Environmental gating belongs to the harness, not to test bodies.
+scan "t.Skip outside testkit" \
+    '\.Skip(f|Now)?\(' \
+    test_code_files 'tests/testkit/' 'phase 2'
+
+# 3. testing.Short(). The tier mechanism becomes build tags, which are
+#    compile-time and cannot be silently forgotten. -short silently skipped ~161
+#    sites while printing green.
+scan "testing.Short()" \
+    'testing\.Short\(\)' \
+    all_go_files '' 'phase 2'
+
+# 4. Raw websocket dialing. The cursor-before-write ordering and the
+#    consecutive-timeout guard are written exactly once, in testkit/firehose.go.
+#    Ten hand-rolled copies is how the suite ended up with 28 subscribe-after-write
+#    races papered over with sleeps.
+scan "websocket.DefaultDialer outside testkit" \
+    'websocket\.DefaultDialer' \
+    all_go_files 'tests/testkit/' 'phase 4'
+
+# 5. Endpoint literals. Every address comes from testkit.Endpoints(), so there
+#    is one place that decides whether a test talks to the CI stack or to a
+#    developer's dev stack.
+scan "hardcoded host:port in test code" \
+    '(localhost|127\.0\.0\.1):[0-9]{2,5}' \
+    test_code_files 'tests/testkit/' 'phase 3-4'
+
+# 6. Public atProto infrastructure. Hermetic means hermetic: the only tier
+#    allowed to reach the real network is tests/live, where it is explicit and
+#    opt-in. Production source legitimately names these hosts, so this counts
+#    test code only.
+#
+#    Scope-aware, per §3.7.2: the target is endpoint construction, not fixture
+#    data. A handle like "testuser.bsky.social" in a table-driven consumer test
+#    is data that is never dialled, so only hostnames inside a URL with a
+#    scheme are counted (92 pure-data matches drop out). What remains includes
+#    legitimate cases — a URL parser's inputs, a config test asserting the
+#    production default — which are exempted individually by putting
+#    "coves:allow-public-host: <reason>" on the line, so each exemption is a
+#    decision someone made rather than a hole in the pattern.
+scan "public atProto hosts outside tests/live" \
+    '(https?|wss?)://[^"'"'"'[:space:]]*(plc\.directory|bsky\.network|bsky\.social|bsky\.app)' \
+    test_code_files 'tests/live/' 'phase 2' 'coves:allow-public-host'
+
+# ---------------------------------------------------------------------------
+# Report
+# ---------------------------------------------------------------------------
+
+printf "  %-52s %6s   %s\n" "VIOLATION" "COUNT" "BURNS DOWN IN"
+printf "  %-52s %6s   %s\n" "----------------------------------------------------" "------" "-------------"
+for row in "${ROWS[@]}"; do
+    printf '%s\n' "$row"
+done
+printf "  %-52s %6s\n" "----------------------------------------------------" "------"
+printf "  %-52s %6d\n" "TOTAL" "$TOTAL"
+echo
+
+if [[ $VERBOSE -eq 1 ]]; then
+    for detail in "${DETAILS[@]}"; do
+        printf "${YELLOW}%s${RESET}\n" "${detail%%$'\n'*}"
+        printf '%s\n\n' "${detail#*$'\n'}" | sed 's/^/    /'
+    done
+fi
+
+if [[ $TOTAL -eq 0 ]]; then
+    printf "${GREEN}  All categories clear.${RESET}\n"
+else
+    printf "${YELLOW}  Warn mode: these are tracked, not enforced. Run with -v for file:line.${RESET}\n"
+fi
+echo
+
+# Always successful, by design. See the header.
+exit 0
diff --git a/scripts/test-db-prepare.sh b/scripts/test-db-prepare.sh
new file mode 100755
index 0000000..f85bed1
--- /dev/null
+++ b/scripts/test-db-prepare.sh
@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# Provisions the template database that tests/testkit clones per test, and
+# sweeps clones orphaned by processes that died before their cleanup ran.
+#
+# WHY THIS RUNS BEFORE `go test` RATHER THAN INSIDE IT
+#
+# `go test ./...` builds one binary per package and runs several concurrently.
+# If each of those processes provisioned the template itself, N processes would
+# race to drop-and-recreate the same database — a built-in flake, and the
+# reason this is a harness responsibility rather than a TestMain one. Doing it
+# here means the template is already correct by the time any test binary
+# starts.
+#
+# testkit still has a fallback (an advisory-lock-guarded verify-or-provision) so
+# that an ad-hoc `go test ./internal/foo` with no Makefile in the loop works. It
+# is a fallback, not the plan.
+#
+# IDEMPOTENT. Creates the template if absent, rebuilds it if the migration set
+# changed since it was stamped, and does nothing otherwise. Safe to call before
+# every test run, which is what `make test` and scripts/ci-runner.sh do.
+#
+# The work itself is Go (tests/testkit/cmd/testdbprepare) because the CI runner
+# image ships neither psql nor the goose CLI, and because sharing testkit's own
+# provisioning code is what keeps the script and the fallback from drifting
+# apart.
+#
+# Coordinates come from the environment — POSTGRES_TEST_HOST/PORT/USER/
+# PASSWORD/DB, POSTGRES_TEST_TEMPLATE — with defaults matching
+# docker-compose.dev.yml's postgres-test as published on the host (localhost:5434)
+# and .env.ci inside the hermetic stack, which share a network namespace. So the
+# same invocation works from a developer's shell and from the CI runner.
+#
+# Flags are passed straight through:
+#   --force            rebuild even if the stamp matches
+#   --sweep-age 30m    change the orphan age cutoff (0 disables sweeping)
+#   --wait 90s         how long to wait for Postgres to accept connections
+#   --print-parallel   print ONLY the safe `go test -parallel` value and exit,
+#                      for $(...) capture by the Makefile and the CI runner
+set -euo pipefail
+
+REPO_ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+cd "$REPO_ROOT"
+
+# --print-parallel is consumed by a shell substitution, so it must emit the
+# number and nothing else.
+if [[ ${1:-} != "--print-parallel" ]]; then
+    echo "▶ Preparing the test template database"
+fi
+
+exec go run ./tests/testkit/cmd/testdbprepare "$@"
diff --git a/tests/testkit/cmd/testdbprepare/main.go b/tests/testkit/cmd/testdbprepare/main.go
new file mode 100644
index 0000000..d31f09e
--- /dev/null
+++ b/tests/testkit/cmd/testdbprepare/main.go
@@ -0,0 +1,124 @@
+// Command testdbprepare provisions the template database that testkit.DB
+// clones, and sweeps clones left behind by processes that died before their
+// cleanup ran.
+//
+// It exists as a Go program rather than as SQL in a shell script for one
+// reason: the CI runner image has no psql and no goose binary, and adding them
+// would mean the provisioning logic lived in two places — here and in
+// tests/testkit/db.go — with nothing keeping them in step. This shares the
+// harness's own code, so the template the script builds is by construction the
+// template the fallback path would have built.
+//
+// Driven by scripts/test-db-prepare.sh; see that script for the wiring.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"Coves/tests/testkit"
+)
+
+func main() {
+	force := flag.Bool("force", false,
+		"rebuild the template even if its stamp already matches the migrations")
+	sweepAge := flag.Duration("sweep-age", time.Hour,
+		"drop idle leftover clone databases older than this (0 disables the sweep)")
+	wait := flag.Duration("wait", 60*time.Second,
+		"how long to wait for Postgres to accept connections before giving up")
+	printParallel := flag.Bool("print-parallel", false,
+		"print only the safe `go test -parallel` value for this server, and exit")
+	flag.Parse()
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	if *printParallel {
+		if err := reportParallelBudget(ctx, *wait); err != nil {
+			fmt.Fprintf(os.Stderr, "test-db-prepare: %v\n", err)
+			os.Exit(1)
+		}
+		return
+	}
+
+	if err := run(ctx, *force, *sweepAge, *wait); err != nil {
+		fmt.Fprintf(os.Stderr, "test-db-prepare: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+// reportParallelBudget prints the -parallel value and nothing else, so a shell
+// can capture it with $(...).
+func reportParallelBudget(ctx context.Context, wait time.Duration) error {
+	if err := testkit.WaitForPostgres(ctx, wait); err != nil {
+		return err
+	}
+	budget, err := testkit.ParallelBudget(ctx)
+	if err != nil {
+		return err
+	}
+	fmt.Println(budget)
+	return nil
+}
+
+func run(ctx context.Context, force bool, sweepAge, wait time.Duration) error {
+	pg := testkit.Endpoints().Postgres
+
+	fmt.Printf("  server:   %s\n", pg.Redacted(pg.Database))
+	fmt.Printf("  template: %s\n", testkit.TemplateName())
+
+	if err := testkit.WaitForPostgres(ctx, wait); err != nil {
+		return err
+	}
+
+	hash, err := testkit.MigrationsHash()
+	if err != nil {
+		return err
+	}
+
+	// One call, one lock acquisition. Asking whether the template is current,
+	// releasing the lock, and then acting on the answer would act on a fact
+	// another process may have invalidated in between — and would need a
+	// hardcoded force=true to make the second step unconditional, which throws
+	// away the re-check the lock exists to protect.
+	start := time.Now()
+	action, err := testkit.ProvisionTemplate(ctx, force)
+	if err != nil {
+		return err
+	}
+	if action == testkit.ProvisionUpToDate {
+		fmt.Printf("  up to date (migrations %s)\n", hash[:12])
+	} else {
+		fmt.Printf("  %s: migrated and stamped %s in %s\n",
+			action, hash[:12], time.Since(start).Round(time.Millisecond))
+	}
+
+	if sweepAge > 0 {
+		result, err := testkit.SweepOrphanClones(ctx, sweepAge)
+		if err != nil {
+			return fmt.Errorf("sweeping orphaned clones: %w", err)
+		}
+		if len(result.Dropped) > 0 {
+			fmt.Printf("  swept %d orphaned clone(s) older than %s\n", len(result.Dropped), sweepAge)
+			for _, name := range result.Dropped {
+				fmt.Printf("    - %s\n", name)
+			}
+		}
+		// Warnings, not failures. A clone still in use by a concurrent run, or
+		// one Postgres declines to drop, is not a reason to fail the gate: the
+		// sweep is opportunistic cleanup and the next run retries.
+		for _, name := range result.Skipped {
+			fmt.Printf("  ! kept %s: still has active sessions\n", name)
+		}
+		for name, dropErr := range result.Failed {
+			fmt.Printf("  ! could not drop %s: %v\n", name, dropErr)
+		}
+	}
+
+	return nil
+}
diff --git a/tests/testkit/db.go b/tests/testkit/db.go
new file mode 100644
index 0000000..9acc91e
--- /dev/null
+++ b/tests/testkit/db.go
@@ -0,0 +1,1015 @@
+package testkit
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"database/sql/driver"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"regexp"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"Coves/internal/db/migrations"
+
+	"github.com/lib/pq"
+	"github.com/pressly/goose/v3"
+)
+
+// Database isolation: one throwaway clone of a migrated template per test.
+//
+// WHY NOT A SHARED DATABASE
+//
+// The suite this replaces shared one database across every test and cleaned up
+// by issuing unscoped DELETEs — 331 of them at last count. That made `-p 1`
+// mandatory (packages running concurrently deleted each other's fixtures) and
+// made every test's correctness depend on every other test's cleanup
+// discipline. Cleanup discipline is exactly what failed.
+//
+// Cloning a pre-migrated template costs ~30ms and makes isolation a property of
+// the construction rather than of the test author's care.
+//
+// THE THREE-PARTY PROTOCOL
+//
+// Three things provision the template, and they must not race:
+//
+//  1. scripts/test-db-prepare.sh, run by `make test` and scripts/ci-runner.sh
+//     before any test binary starts. This is the primary path.
+//  2. This file's EnsureTemplate, as a fallback for ad-hoc `go test ./...`
+//     invocations that never went through the Makefile.
+//  3. Other test binaries doing (2) concurrently — `go test ./...` runs each
+//     package as its own process, several at a time.
+//
+// A Postgres advisory lock serialises them. Provisioning takes it exclusively;
+// cloning takes it shared. A clone therefore cannot start while the template is
+// being dropped and rebuilt, and any number of clones can proceed at once.
+//
+// Advisory locks are scoped to the database the session is connected to, so
+// every participant must connect to the same maintenance database
+// (POSTGRES_TEST_DB, default coves_test) — which is also the only database from
+// which CREATE DATABASE / DROP DATABASE are issued.
+
+const (
+	// defaultTemplateDB is the migrated database every test clone is copied
+	// from. Override with POSTGRES_TEST_TEMPLATE — subject to
+	// validateTemplateName, because this name is an argument to DROP DATABASE.
+	defaultTemplateDB = "coves_test_template"
+
+	// ClonePrefix marks a database as a testkit clone. Nothing outside this
+	// prefix (and the validated template name) is ever dropped by testkit or by
+	// scripts/test-db-prepare.sh — the sweep runs against a live Postgres that
+	// also holds the dev database, so "drop anything that looks unused" is not
+	// an acceptable rule.
+	ClonePrefix = "tkclone_"
+
+	// Advisory lock coordinates. The class id spells "COVE" in ASCII; object
+	// ids are allocated here as testkit grows more cluster-wide critical
+	// sections. Both halves are needed because Postgres advisory locks are
+	// identified by the key alone — any other tool using key (1129530437, 1)
+	// against the same database would contend with template provisioning.
+	advisoryLockClassID       = 1129530437 // 'C','O','V','E'
+	advisoryLockTemplateObjID = 1          // template provisioning / cloning
+
+	// lockAcquireTimeout bounds how long a caller waits for the template lock.
+	//
+	// Unbounded would be worse than it sounds: a process killed mid-migration
+	// releases its lock when its backend exits, but a *wedged* one does not, and
+	// every other test binary would then block forever with no output at all.
+	// Two minutes comfortably exceeds a full provisioning run (~300ms) while
+	// still failing inside any sane test timeout.
+	lockAcquireTimeout = 2 * time.Minute
+
+	// clonePoolMaxOpen bounds the connections a single test's pool may open.
+	//
+	// This is the connection budget from the spec: with a clone per test and
+	// tests running in parallel, pools multiply. ParallelBudget derives the
+	// safe -parallel value from this and the server's max_connections, and the
+	// compose files raise max_connections to match.
+	clonePoolMaxOpen = 3
+	clonePoolMaxIdle = 2
+
+	// adminPoolMaxOpen bounds the coordination pool. These sessions only take
+	// advisory locks and issue CREATE/DROP DATABASE; they run no application
+	// SQL, so a small pool is right — and every connection spent here is one
+	// unavailable to an actual test.
+	adminPoolMaxOpen = 3
+
+	// stampTable records which migration set the template was built from, so a
+	// stale template is detected instead of silently used.
+	stampTable = "testkit_template_stamp"
+
+	// sqlstateObjectInUse is Postgres' "source database is being accessed by
+	// other users" — see cloneTemplate for why it is retried.
+	sqlstateObjectInUse = "55006"
+)
+
+// cloneNamePattern is the safety rail: every destructive statement testkit
+// issues checks its target against this first.
+var cloneNamePattern = regexp.MustCompile(`^` + ClonePrefix + `[a-z0-9_]+$`)
+
+// templateNamePattern constrains what may be used as a template database name.
+var templateNamePattern = regexp.MustCompile(`^[a-z][a-z0-9_]*$`)
+
+// ---------------------------------------------------------------------------
+// Template naming (and the rail around it)
+// ---------------------------------------------------------------------------
+
+// templateNameOnce is a process singleton like the others below, and is read
+// through resolveTemplateName so testkit's own tests can rebind it safely.
+var templateNameOnce = sync.OnceValues(loadTemplateName)
+
+// resolveTemplateName returns the validated template database name.
+func resolveTemplateName() (string, error) {
+	singletonMu.RLock()
+	fn := templateNameOnce
+	singletonMu.RUnlock()
+	return fn()
+}
+
+// TemplateName returns the name of the migrated template database.
+//
+// It panics if POSTGRES_TEST_TEMPLATE names something testkit must not touch.
+// A panic rather than a returned error because this is a configuration mistake
+// with a destructive blast radius: the name reaches DROP DATABASE, and the one
+// thing that must not happen is quietly carrying on with it.
+func TemplateName() string {
+	name, err := resolveTemplateName()
+	if err != nil {
+		panic("testkit: " + err.Error())
+	}
+	return name
+}
+
+func loadTemplateName() (string, error) {
+	name := envOr("POSTGRES_TEST_TEMPLATE", defaultTemplateDB)
+	if err := validateTemplateName(name, Endpoints().Postgres.Database); err != nil {
+		return "", err
+	}
+	return name, nil
+}
+
+// validateTemplateName rejects any template name that testkit must not drop.
+//
+// The template is the one database outside the clone prefix that testkit
+// destroys, and its name comes from the environment. A typo — or a copied
+// .env line — setting POSTGRES_TEST_TEMPLATE=coves_dev would otherwise hand
+// DROP DATABASE ... WITH (FORCE) the development database, which is precisely
+// what the clone-prefix rail exists to make impossible everywhere else.
+//
+// The "test" substring requirement is the load-bearing part: it confines the
+// template to a namespace no real database occupies. The explicit denylist
+// below is redundant with it for coves_dev and postgres, and kept anyway
+// because a rail worth having is worth stating twice.
+func validateTemplateName(name, maintenanceDB string) error {
+	switch {
+	case name == "":
+		return errors.New("template database name is empty")
+	case len(name) > 63:
+		return fmt.Errorf("template database name %q exceeds Postgres' 63-byte identifier limit", name)
+	case !templateNamePattern.MatchString(name):
+		return fmt.Errorf("template database name %q must be lowercase alphanumeric with underscores, starting with a letter", name)
+	case !strings.Contains(name, "test"):
+		return fmt.Errorf("refusing to use %q as the template database: the name must contain \"test\"; "+
+			"testkit drops and recreates this database, so it must live in a namespace no real database occupies", name)
+	case strings.HasPrefix(name, ClonePrefix):
+		return fmt.Errorf("refusing to use %q as the template database: %q is the per-test clone prefix, "+
+			"and the orphan sweep would delete it", name, ClonePrefix)
+	case name == maintenanceDB:
+		return fmt.Errorf("refusing to use %q as the template database: it is the maintenance database "+
+			"testkit connects to in order to drop and create databases", name)
+	case slices.Contains([]string{"postgres", "template0", "template1", "coves_dev", "coves_prod"}, name):
+		return fmt.Errorf("refusing to use %q as the template database: testkit drops and recreates it", name)
+	}
+	return nil
+}
+
+// ---------------------------------------------------------------------------
+// Process singletons
+// ---------------------------------------------------------------------------
+
+// The memoised singletons are swapped out by testkit's own tests (to point at
+// an unreachable server, to re-read the environment), so they are read through
+// a lock rather than directly. Without it `go test -shuffle=on -race` reports a
+// data race between a test rebinding one and another goroutine reading it.
+var (
+	singletonMu   sync.RWMutex
+	endpointsOnce = sync.OnceValue(loadEndpoints)
+	adminDBOnce   = sync.OnceValues(openAdminDB)
+)
+
+func loadedEndpoints() func() EndpointSet {
+	singletonMu.RLock()
+	defer singletonMu.RUnlock()
+	return endpointsOnce
+}
+
+func openAdminDB() (*sql.DB, error) {
+	pg := Endpoints().Postgres
+	db, err := sql.Open("postgres", pg.URL(pg.Database))
+	if err != nil {
+		return nil, fmt.Errorf("opening %s: %w", pg.Redacted(pg.Database), err)
+	}
+	db.SetMaxOpenConns(adminPoolMaxOpen)
+	db.SetMaxIdleConns(2)
+	db.SetConnMaxLifetime(5 * time.Minute)
+	return db, nil
+}
+
+func adminDB() (*sql.DB, error) {
+	singletonMu.RLock()
+	fn := adminDBOnce
+	singletonMu.RUnlock()
+	return fn()
+}
+
+// WaitForPostgres blocks until the maintenance database answers, or the timeout
+// expires. Used by scripts/test-db-prepare.sh, which may be invoked moments
+// after `docker compose up` returns.
+func WaitForPostgres(ctx context.Context, timeout time.Duration) error {
+	db, err := adminDB()
+	if err != nil {
+		return err
+	}
+	deadline := time.Now().Add(timeout)
+	var lastErr error
+	for {
+		pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
+		lastErr = db.PingContext(pingCtx)
+		cancel()
+		if lastErr == nil {
+			return nil
+		}
+		if time.Now().After(deadline) {
+			pg := Endpoints().Postgres
+			return fmt.Errorf("test database %s unreachable after %s: %w",
+				pg.Redacted(pg.Database), timeout, lastErr)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(250 * time.Millisecond):
+		}
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Advisory locking
+// ---------------------------------------------------------------------------
+
+// withAdvisoryLock runs fn while holding the testkit template lock, shared or
+// exclusive, on a dedicated session.
+//
+// THE SESSION IS DESTROYED, NOT POOLED, ON ANY LOCKING ERROR.
+//
+// database/sql returns connections to the pool without resetting session state,
+// and advisory locks are session-scoped. Two failure modes follow, and both end
+// with a pooled connection silently holding a lock that nothing will ever
+// release — which manifests, minutes later, as every other test binary blocking
+// on the template lock:
+//
+//   - The acquire appears to fail. Postgres query cancellation is asynchronous,
+//     so a context deadline reached while waiting for the lock can return an
+//     error to Go after the backend has already been granted the lock.
+//   - The release fails, or reports false. pg_advisory_unlock returns a boolean;
+//     false means this session did not hold the lock it tried to release, which
+//     means the bookkeeping here and in the server have diverged.
+//
+// Returning driver.ErrBadConn from Conn.Raw discards the connection instead of
+// pooling it, which turns both cases into "one wasted connection" — the backend
+// exits, and Postgres releases every advisory lock it held.
+func withAdvisoryLock(ctx context.Context, shared bool, fn func(context.Context, *sql.Conn) error) error {
+	db, err := adminDB()
+	if err != nil {
+		return err
+	}
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return fmt.Errorf("acquiring admin connection: %w", err)
+	}
+
+	lockFn, unlockFn := "pg_advisory_lock", "pg_advisory_unlock"
+	if shared {
+		lockFn, unlockFn = lockFn+"_shared", unlockFn+"_shared"
+	}
+
+	acquireCtx, cancelAcquire := context.WithTimeout(ctx, lockAcquireTimeout)
+	_, err = conn.ExecContext(acquireCtx,
+		fmt.Sprintf("SELECT %s($1, $2)", lockFn),
+		advisoryLockClassID, advisoryLockTemplateObjID)
+	cancelAcquire()
+	if err != nil {
+		discardConn(conn)
+		if errors.Is(err, context.DeadlineExceeded) {
+			return fmt.Errorf(
+				"timed out after %s waiting for the %s testkit template lock (%d, %d) on %s%s",
+				lockAcquireTimeout, lockName(shared),
+				advisoryLockClassID, advisoryLockTemplateObjID,
+				Endpoints().Postgres.Redacted(Endpoints().Postgres.Database),
+				describeLockHolders(ctx))
+		}
+		return fmt.Errorf("acquiring %s advisory lock: %w", lockName(shared), err)
+	}
+
+	var poisoned bool
+	defer func() {
+		// A fresh context: if ctx was cancelled we still have to release.
+		releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+		defer cancel()
+
+		var released bool
+		relErr := conn.QueryRowContext(releaseCtx,
+			fmt.Sprintf("SELECT %s($1, $2)", unlockFn),
+			advisoryLockClassID, advisoryLockTemplateObjID).Scan(&released)
+		if relErr != nil || !released || poisoned {
+			discardConn(conn)
+			return
+		}
+		_ = conn.Close()
+	}()
+
+	if err := fn(ctx, conn); err != nil {
+		// The callback runs its statements on this same session; if it failed
+		// mid-statement the session's state is not something to hand to the
+		// next caller.
+		poisoned = true
+		return err
+	}
+	return nil
+}
+
+// discardConn removes a session from the pool rather than returning it, so any
+// advisory lock it may still hold dies with its backend.
+func discardConn(conn *sql.Conn) {
+	_ = conn.Raw(func(any) error { return driver.ErrBadConn })
+	_ = conn.Close()
+}
+
+// describeLockHolders makes a lock timeout diagnosable by naming who is holding
+// it. Best-effort: it returns "" rather than an error, because it runs on a
+// path that is already failing and must not fail differently.
+func describeLockHolders(ctx context.Context) string {
+	db, err := adminDB()
+	if err != nil {
+		return ""
+	}
+	queryCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
+
+	rows, err := db.QueryContext(queryCtx, `
+		SELECT a.pid, a.application_name, a.state, coalesce(a.query, '')
+		FROM pg_locks l
+		JOIN pg_stat_activity a ON a.pid = l.pid
+		WHERE l.locktype = 'advisory' AND l.classid = $1 AND l.objid = $2 AND l.granted`,
+		advisoryLockClassID, advisoryLockTemplateObjID)
+	if err != nil {
+		return ""
+	}
+	defer func() { _ = rows.Close() }()
+
+	var holders []string
+	for rows.Next() {
+		var pid int
+		var app, state, query string
+		if err := rows.Scan(&pid, &app, &state, &query); err != nil {
+			return ""
+		}
+		if len(query) > 120 {
+			query = query[:120] + "..."
+		}
+		holders = append(holders, fmt.Sprintf("pid %d (%s, %s): %s", pid, app, state, query))
+	}
+	if len(holders) == 0 {
+		return ""
+	}
+	return "\n  currently held by:\n    " + strings.Join(holders, "\n    ")
+}
+
+func lockName(shared bool) string {
+	if shared {
+		return "shared"
+	}
+	return "exclusive"
+}
+
+// ---------------------------------------------------------------------------
+// Migration fingerprint
+// ---------------------------------------------------------------------------
+
+var migrationsHashOnce = sync.OnceValues(func() (string, error) {
+	return HashMigrations(migrations.FS)
+})
+
+// MigrationsHash fingerprints the AppView's embedded migration set.
+//
+// The template is stamped with this value, and a mismatch means the template
+// predates a migration and must be rebuilt. Hashing the embedded FS rather than
+// a directory on disk means the answer is identical from the host, from the CI
+// runner, and from any working directory.
+func MigrationsHash() (string, error) {
+	return migrationsHashOnce()
+}
+
+// HashMigrations fingerprints the .sql files in fsys.
+//
+// Any change to the set — a file's content, its name, one added, one removed —
+// must change the digest, because the digest is the only thing standing between
+// a changed schema and a stale template being cloned into every test.
+func HashMigrations(fsys fs.FS) (string, error) {
+	sum := sha256.New()
+	// WalkDir visits lexically, so the digest is order-stable.
+	err := fs.WalkDir(fsys, ".", func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() || !strings.HasSuffix(path, ".sql") {
+			return nil
+		}
+		content, readErr := fs.ReadFile(fsys, path)
+		if readErr != nil {
+			return readErr
+		}
+		// Length-prefixed so that moving a byte across the boundary between two
+		// files cannot produce the same digest.
+		var length [8]byte
+		binary.BigEndian.PutUint64(length[:], uint64(len(content)))
+		_, _ = io.WriteString(sum, path)
+		_, _ = sum.Write(length[:])
+		_, _ = sum.Write(content)
+		return nil
+	})
+	if err != nil {
+		return "", fmt.Errorf("hashing migrations: %w", err)
+	}
+	return hex.EncodeToString(sum.Sum(nil)), nil
+}
+
+// ---------------------------------------------------------------------------
+// Template provisioning
+// ---------------------------------------------------------------------------
+
+// ProvisionAction says what a provisioning run did, so callers can report it.
+type ProvisionAction string
+
+const (
+	ProvisionUpToDate ProvisionAction = "up to date"
+	ProvisionCreated  ProvisionAction = "created"
+	ProvisionRebuilt  ProvisionAction = "rebuilt (migrations changed)"
+	ProvisionForced   ProvisionAction = "rebuilt (forced)"
+)
+
+var (
+	templateMu       sync.Mutex
+	templateVerified bool
+)
+
+// EnsureTemplate guarantees that a template database matching the current
+// migration set exists, provisioning it if it does not.
+//
+// It is verified at most once per process — the check costs a round trip and a
+// connection to the template, and re-running it per test would serialise the
+// suite behind the very lock it takes. Cross-process safety comes from the
+// advisory lock, not from this flag.
+func EnsureTemplate(ctx context.Context) error {
+	templateMu.Lock()
+	defer templateMu.Unlock()
+	if templateVerified {
+		return nil
+	}
+	if _, err := provisionTemplateLocked(ctx, false); err != nil {
+		return err
+	}
+	templateVerified = true
+	return nil
+}
+
+// ProvisionTemplate creates or rebuilds the template database and reports what
+// it did. With force, the template is rebuilt even if its stamp already matches.
+//
+// The decision and the action are one operation under one lock acquisition:
+// asking whether the template is current, releasing the lock, and then acting on
+// the answer would act on a fact that another process may have invalidated in
+// between. This is what scripts/test-db-prepare.sh calls.
+func ProvisionTemplate(ctx context.Context, force bool) (ProvisionAction, error) {
+	templateMu.Lock()
+	defer templateMu.Unlock()
+	action, err := provisionTemplateLocked(ctx, force)
+	if err != nil {
+		return action, err
+	}
+	templateVerified = true
+	return action, nil
+}
+
+func provisionTemplateLocked(ctx context.Context, force bool) (ProvisionAction, error) {
+	want, err := MigrationsHash()
+	if err != nil {
+		return "", err
+	}
+	// Re-validated here rather than trusted from the memoised accessor: this is
+	// the function that issues DROP DATABASE, and a rail checked at the point
+	// of use cannot be bypassed by a future caller that reaches it another way.
+	template, err := resolveTemplateName()
+	if err != nil {
+		return "", err
+	}
+	if err := validateTemplateName(template, Endpoints().Postgres.Database); err != nil {
+		return "", err
+	}
+
+	action := ProvisionUpToDate
+	lockErr := withAdvisoryLock(ctx, false, func(ctx context.Context, conn *sql.Conn) error {
+		exists, current, err := templateState(ctx, conn, template, want)
+		if err != nil {
+			return err
+		}
+		switch {
+		case force:
+			action = ProvisionForced
+		case !exists:
+			action = ProvisionCreated
+		case !current:
+			action = ProvisionRebuilt
+		default:
+			action = ProvisionUpToDate
+			return nil
+		}
+
+		// Rebuild rather than migrate-in-place: a template whose stamp does not
+		// match may have been built from a migration set that no longer exists
+		// (a rewritten migration during pre-production development is normal
+		// here), and goose cannot undo what it cannot see. Dropping is cheap
+		// and always correct.
+		//
+		// WITH (FORCE) terminates any session still connected to the template.
+		// Without it, one leaked connection from a crashed run wedges every
+		// subsequent run.
+		if _, err := conn.ExecContext(ctx,
+			"DROP DATABASE IF EXISTS "+pq.QuoteIdentifier(template)+" WITH (FORCE)"); err != nil {
+			return fmt.Errorf("dropping stale template %q: %w", template, err)
+		}
+		if _, err := conn.ExecContext(ctx,
+			"CREATE DATABASE "+pq.QuoteIdentifier(template)); err != nil {
+			return fmt.Errorf("creating template %q: %w", template, err)
+		}
+		return migrateAndStamp(ctx, template, want)
+	})
+	if lockErr != nil {
+		return "", lockErr
+	}
+	return action, nil
+}
+
+// TemplateStatus reports whether the template database exists and whether its
+// stamp matches the current migration set.
+//
+// It takes the lock EXCLUSIVELY even though it only reads: checking the stamp
+// means connecting to the template, and CREATE DATABASE ... TEMPLATE refuses to
+// run while any session is connected to its source. A shared lock here would
+// make a concurrent clone fail with "source database is being accessed by other
+// users".
+//
+// Callers deciding whether to provision should use ProvisionTemplate, which
+// decides and acts under one lock acquisition. This exists for reporting.
+func TemplateStatus(ctx context.Context) (exists, current bool, err error) {
+	want, err := MigrationsHash()
+	if err != nil {
+		return false, false, err
+	}
+	template, err := resolveTemplateName()
+	if err != nil {
+		return false, false, err
+	}
+	err = withAdvisoryLock(ctx, false, func(ctx context.Context, conn *sql.Conn) error {
+		exists, current, err = templateState(ctx, conn, template, want)
+		return err
+	})
+	if err != nil {
+		return false, false, err
+	}
+	return exists, current, nil
+}
+
+// templateState reports whether the template exists and carries a stamp
+// matching the current migration set. The caller must hold the exclusive lock.
+func templateState(ctx context.Context, conn *sql.Conn, template, wantHash string) (exists, current bool, err error) {
+	if err := conn.QueryRowContext(ctx,
+		"SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)", template,
+	).Scan(&exists); err != nil {
+		return false, false, fmt.Errorf("looking up template %q: %w", template, err)
+	}
+	if !exists {
+		return false, false, nil
+	}
+
+	// The connection to the template is deliberately short-lived: CREATE
+	// DATABASE ... TEMPLATE refuses to run while any session is connected to
+	// the source, so holding this open would break every concurrent clone.
+	tdb, err := openDatabase(template, 1)
+	if err != nil {
+		return true, false, err
+	}
+	defer func() { _ = tdb.Close() }()
+
+	var got string
+	scanErr := tdb.QueryRowContext(ctx,
+		"SELECT migrations_hash FROM "+pq.QuoteIdentifier(stampTable)+" LIMIT 1").Scan(&got)
+	switch {
+	case scanErr == nil:
+		return true, got == wantHash, nil
+	case errors.Is(scanErr, sql.ErrNoRows):
+		return true, false, nil
+	default:
+		// Most likely the stamp table does not exist: a template from before
+		// stamping, or a half-provisioned one. Either way, rebuild.
+		return true, false, nil
+	}
+}
+
+// migrateAndStamp runs every migration against a freshly created template and
+// records the fingerprint it was built from.
+func migrateAndStamp(ctx context.Context, template, hash string) error {
+	db, err := openDatabase(template, 1)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = db.Close() }()
+
+	// goose.NewProvider rather than the package-level goose.Up: the package
+	// API keeps the dialect and base filesystem in global variables, and the
+	// legacy tests/integration path still calls goose.Up with a relative
+	// directory. Two callers with different notions of where migrations live,
+	// sharing one global, is a bug waiting for the first test binary that does
+	// both.
+	provider, err := goose.NewProvider(goose.DialectPostgres, db, migrations.FS)
+	if err != nil {
+		return fmt.Errorf("configuring goose for template %q: %w", template, err)
+	}
+	if _, err := provider.Up(ctx); err != nil {
+		return fmt.Errorf("migrating template %q: %w", template, err)
+	}
+
+	if _, err := db.ExecContext(ctx, `
+		CREATE TABLE `+pq.QuoteIdentifier(stampTable)+` (
+			id             boolean PRIMARY KEY DEFAULT true CHECK (id),
+			migrations_hash text        NOT NULL,
+			stamped_at      timestamptz NOT NULL DEFAULT now()
+		)`); err != nil {
+		return fmt.Errorf("creating stamp table in %q: %w", template, err)
+	}
+	if _, err := db.ExecContext(ctx,
+		"INSERT INTO "+pq.QuoteIdentifier(stampTable)+" (migrations_hash) VALUES ($1)", hash); err != nil {
+		return fmt.Errorf("stamping template %q: %w", template, err)
+	}
+	return nil
+}
+
+func openDatabase(name string, maxOpen int) (*sql.DB, error) {
+	pg := Endpoints().Postgres
+	db, err := sql.Open("postgres", pg.URL(name))
+	if err != nil {
+		return nil, fmt.Errorf("opening %s: %w", pg.Redacted(name), err)
+	}
+	db.SetMaxOpenConns(maxOpen)
+	db.SetMaxIdleConns(min(maxOpen, clonePoolMaxIdle))
+	return db, nil
+}
+
+// ---------------------------------------------------------------------------
+// Connection budget
+// ---------------------------------------------------------------------------
+
+// appviewConnectionHeadroom is the number of connections reserved for everything
+// that is not a test clone: the AppView's own pool, the admin pool, psql
+// sessions, Postgres' superuser reserve.
+const appviewConnectionHeadroom = 40
+
+// ParallelBudget returns the largest `go test -parallel` value the server's
+// max_connections can support, given that every parallel test holds its own
+// clone pool.
+//
+// Without this the connection budget is an unwritten assumption, and the way it
+// surfaces is a run that fails with "sorry, too many clients already" in
+// whichever test happened to be unlucky — a failure that reads like a bug in
+// that test.
+func ParallelBudget(ctx context.Context) (int, error) {
+	db, err := adminDB()
+	if err != nil {
+		return 0, err
+	}
+	var raw string
+	if err := db.QueryRowContext(ctx, "SHOW max_connections").Scan(&raw); err != nil {
+		return 0, fmt.Errorf("reading max_connections: %w", err)
+	}
+	maxConns, err := strconv.Atoi(strings.TrimSpace(raw))
+	if err != nil {
+		return 0, fmt.Errorf("parsing max_connections %q: %w", raw, err)
+	}
+	budget := (maxConns - appviewConnectionHeadroom) / clonePoolMaxOpen
+	// Floor of 1 (never zero, which go test would reject) and a ceiling that
+	// keeps a huge max_connections from producing a -parallel value far past
+	// the machine's ability to run tests anyway.
+	return max(1, min(budget, 64)), nil
+}
+
+// ---------------------------------------------------------------------------
+// Per-test clones
+// ---------------------------------------------------------------------------
+
+var cloneCounter = newCounter()
+
+// newCloneName builds a collision-free, sweepable clone name.
+//
+// Shape: tkclone_<created-unix-seconds base36>_<run prefix>_<counter base36>.
+// The run prefix is per-process random, so two concurrent `go test` processes
+// (or two developers against one Postgres) cannot collide. The timestamp is
+// there for the orphan sweep: pg_database records no creation time, so age
+// has to be written into the name.
+func newCloneName() string {
+	return fmt.Sprintf("%s%s_%s_%s",
+		ClonePrefix,
+		strconv.FormatInt(time.Now().Unix(), 36),
+		RunPrefix(),
+		strconv.FormatUint(cloneCounter.next(), 36),
+	)
+}
+
+// DB returns a private, fully migrated Postgres database for this test.
+//
+// The database is a clone of the migrated template, so it starts with the
+// production schema and no rows. It is dropped when the test finishes; nothing
+// the test writes is visible to any other test, and no cleanup code is needed
+// (or wanted — an explicit DELETE in a test body is a sign the test is still
+// assuming a shared database).
+//
+// A missing or unreachable Postgres fails the test. Tests do not skip on absent
+// infrastructure: if the suite was invoked, the infrastructure was requested.
+func DB(t TestingT) *sql.DB {
+	t.Helper()
+	ctx := context.Background()
+
+	if err := EnsureTemplate(ctx); err != nil {
+		pg := Endpoints().Postgres
+		// resolveTemplateName, not TemplateName: one of the ways EnsureTemplate
+		// fails is that the configured template name is unsafe, and TemplateName
+		// panics on exactly that. Rendering the failure must not become a
+		// second, louder failure that buries the first.
+		template, nameErr := resolveTemplateName()
+		if nameErr != nil {
+			template = "<invalid POSTGRES_TEST_TEMPLATE>"
+		}
+		t.Fatalf("testkit.DB: no usable template database at %s: %v\n"+
+			"  the test database is expected at %s — start it with 'make test' or 'make dev-up'",
+			pg.Redacted(template), err, pg.Redacted(pg.Database))
+		return nil
+	}
+
+	name := newCloneName()
+	if err := cloneTemplate(ctx, name); err != nil {
+		t.Fatalf("testkit.DB: %v", err)
+		return nil
+	}
+
+	db, err := openDatabase(name, clonePoolMaxOpen)
+	if err != nil {
+		_ = dropDatabase(ctx, name)
+		t.Fatalf("testkit.DB: %v", err)
+		return nil
+	}
+	if err := db.PingContext(ctx); err != nil {
+		_ = db.Close()
+		_ = dropDatabase(ctx, name)
+		t.Fatalf("testkit.DB: pinging clone %q: %v", name, err)
+		return nil
+	}
+
+	t.Cleanup(func() {
+		// DROP FIRST, CLOSE SECOND. sql.DB.Close waits for in-flight queries to
+		// finish, so closing first means one goroutine still running a slow
+		// query — or one connection a test forgot to release — blocks teardown
+		// and the drop never happens. DROP ... WITH (FORCE) terminates the
+		// clone's backends outright, after which Close finds only dead
+		// connections and returns immediately.
+		//
+		// Reported, not ignored: a clone that survives its test is a leak, and
+		// leaks nobody sees accumulate until Postgres runs out of something.
+		// The sweep in test-db-prepare.sh is the backstop for processes that
+		// died before reaching here, not a licence to leak.
+		if err := dropDatabase(context.Background(), name); err != nil {
+			t.Errorf("testkit.DB: leaked test database %q: %v", name, err)
+		}
+		_ = db.Close()
+	})
+
+	return db
+}
+
+// cloneTemplate copies the template under a shared advisory lock, so a
+// concurrent provisioning run cannot pull the template out from under it.
+//
+// The retry is not defensive padding. Postgres terminates a backend
+// asynchronously, so the short-lived stamp-checking connection that
+// templateState opens is still visible in pg_stat_activity for a few
+// milliseconds after Close returns — and CREATE DATABASE ... TEMPLATE refuses
+// to run while any session is connected to its source. Without this, the first
+// testkit.DB call in a package would intermittently fail with SQLSTATE 55006,
+// more often on a loaded machine.
+func cloneTemplate(ctx context.Context, name string) error {
+	if !cloneNamePattern.MatchString(name) {
+		return fmt.Errorf("refusing to create %q: not a testkit clone name", name)
+	}
+	template, err := resolveTemplateName()
+	if err != nil {
+		return err
+	}
+
+	const attempts = 3
+	var lastErr error
+	for attempt := 1; attempt <= attempts; attempt++ {
+		lastErr = withAdvisoryLock(ctx, true, func(ctx context.Context, conn *sql.Conn) error {
+			_, execErr := conn.ExecContext(ctx,
+				"CREATE DATABASE "+pq.QuoteIdentifier(name)+" TEMPLATE "+pq.QuoteIdentifier(template))
+			return execErr
+		})
+		if lastErr == nil {
+			return nil
+		}
+		if !isSourceInUse(lastErr) {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(100 * time.Millisecond):
+		}
+	}
+	return fmt.Errorf("cloning template %q into %q: %w", template, name, lastErr)
+}
+
+// isSourceInUse reports whether err is Postgres declining to copy a database
+// because something is connected to it.
+func isSourceInUse(err error) bool {
+	var pqErr *pq.Error
+	return errors.As(err, &pqErr) && string(pqErr.Code) == sqlstateObjectInUse
+}
+
+// dropDatabase force-drops a testkit clone.
+//
+// WITH (FORCE) rather than a plain DROP: a test that leaves a connection open —
+// a *sql.DB it never closed, a goroutine still querying — would otherwise wedge
+// its own teardown, and the failure would surface as a confusing timeout in an
+// unrelated later test. The orphan sweep deliberately does NOT use FORCE; see
+// SweepOrphanClones.
+func dropDatabase(ctx context.Context, name string) error {
+	if !cloneNamePattern.MatchString(name) {
+		return fmt.Errorf("refusing to drop %q: not a testkit clone name", name)
+	}
+	db, err := adminDB()
+	if err != nil {
+		return err
+	}
+	dropCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	if _, err := db.ExecContext(dropCtx,
+		"DROP DATABASE IF EXISTS "+pq.QuoteIdentifier(name)+" WITH (FORCE)"); err != nil {
+		return fmt.Errorf("dropping %q: %w", name, err)
+	}
+	return nil
+}
+
+// ---------------------------------------------------------------------------
+// Orphan sweep
+// ---------------------------------------------------------------------------
+
+// SweepResult reports what a sweep did. Skipped and Failed are warnings: a
+// sweep is opportunistic cleanup, and neither justifies failing a test run.
+type SweepResult struct {
+	Dropped []string
+	// Skipped holds clones that were old enough to look orphaned but still had
+	// sessions connected.
+	Skipped []string
+	// Failed holds clones whose DROP returned an error, keyed by name.
+	Failed map[string]error
+}
+
+// SweepOrphanClones drops testkit clone databases that are older than maxAge
+// and have no sessions connected.
+//
+// Orphans come from processes that died between CREATE DATABASE and their
+// cleanup: a panicking test binary, a killed `go test`, a docker stop.
+//
+// TWO RULES MAKE THIS SAFE TO RUN WHILE ANOTHER SUITE IS MID-FLIGHT.
+//
+// Age is necessary but not sufficient: a long E2E test legitimately holds a
+// clone for more than an hour, and killing its database mid-run would produce a
+// baffling failure in a suite this one has no business touching. So the sweep
+// also requires the clone to be idle, and it does NOT pass WITH (FORCE) —
+// FORCE would terminate exactly the sessions whose existence is the evidence
+// that the clone is still in use.
+//
+// The second rule is that one stubborn database must not fail the run. Errors
+// are collected and reported; the sweep continues.
+func SweepOrphanClones(ctx context.Context, maxAge time.Duration) (SweepResult, error) {
+	result := SweepResult{Failed: map[string]error{}}
+
+	db, err := adminDB()
+	if err != nil {
+		return result, err
+	}
+	rows, err := db.QueryContext(ctx,
+		`SELECT datname FROM pg_database WHERE datname LIKE $1 ORDER BY datname`,
+		ClonePrefix+"%")
+	if err != nil {
+		return result, fmt.Errorf("listing clone databases: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var candidates []string
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			return result, err
+		}
+		candidates = append(candidates, name)
+	}
+	if err := rows.Err(); err != nil {
+		return result, err
+	}
+
+	cutoff := time.Now().Add(-maxAge)
+	for _, name := range candidates {
+		created, ok := cloneCreatedAt(name)
+		if !ok {
+			// A name that matches the prefix but not the layout is not
+			// something testkit created, so it is not something testkit drops.
+			continue
+		}
+		if created.After(cutoff) {
+			continue
+		}
+
+		busy, err := databaseHasSessions(ctx, db, name)
+		if err != nil {
+			result.Failed[name] = err
+			continue
+		}
+		if busy {
+			result.Skipped = append(result.Skipped, name)
+			continue
+		}
+
+		if err := dropDatabaseIfIdle(ctx, db, name); err != nil {
+			result.Failed[name] = err
+			continue
+		}
+		result.Dropped = append(result.Dropped, name)
+	}
+	return result, nil
+}
+
+func databaseHasSessions(ctx context.Context, db *sql.DB, name string) (bool, error) {
+	var count int
+	if err := db.QueryRowContext(ctx,
+		`SELECT count(*) FROM pg_stat_activity WHERE datname = $1`, name).Scan(&count); err != nil {
+		return false, fmt.Errorf("checking sessions on %q: %w", name, err)
+	}
+	return count > 0, nil
+}
+
+// dropDatabaseIfIdle drops a clone WITHOUT FORCE, so a session that connected
+// between the idle check and here makes the drop fail rather than be killed.
+func dropDatabaseIfIdle(ctx context.Context, db *sql.DB, name string) error {
+	if !cloneNamePattern.MatchString(name) {
+		return fmt.Errorf("refusing to drop %q: not a testkit clone name", name)
+	}
+	dropCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	if _, err := db.ExecContext(dropCtx,
+		"DROP DATABASE IF EXISTS "+pq.QuoteIdentifier(name)); err != nil {
+		return fmt.Errorf("dropping %q: %w", name, err)
+	}
+	return nil
+}
+
+// cloneCreatedAt recovers the creation time newCloneName encoded into the name.
+func cloneCreatedAt(name string) (time.Time, bool) {
+	if !cloneNamePattern.MatchString(name) {
+		return time.Time{}, false
+	}
+	parts := strings.Split(strings.TrimPrefix(name, ClonePrefix), "_")
+	if len(parts) != 3 {
+		return time.Time{}, false
+	}
+	secs, err := strconv.ParseInt(parts[0], 36, 64)
+	if err != nil || secs <= 0 {
+		return time.Time{}, false
+	}
+	return time.Unix(secs, 0), true
+}
diff --git a/tests/testkit/db_test.go b/tests/testkit/db_test.go
new file mode 100644
index 0000000..08b4f7f
--- /dev/null
+++ b/tests/testkit/db_test.go
@@ -0,0 +1,626 @@
+package testkit
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"testing/fstest"
+	"time"
+
+	"github.com/lib/pq"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// These tests need Postgres. They fail rather than skip when it is missing:
+// running the suite is a request for its infrastructure, and a suite that
+// prints green with its database down is the failure mode this whole refactor
+// exists to remove. Start it with `make test` or `make dev-up`.
+
+func TestDB_ParallelClonesCannotSeeEachOther(t *testing.T) {
+	// Both subtests insert a user with the SAME primary key. On a shared
+	// database the second insert would fail outright; if the clones somehow
+	// pointed at one database, the row count would be 2. Overlapping in time
+	// is the point, so each waits for the other's write before counting.
+	const sharedDID = "did:plc:testkitisolationprobe"
+	var written atomic.Int32
+
+	for i := 0; i < 2; i++ {
+		t.Run(fmt.Sprintf("clone-%d", i), func(t *testing.T) {
+			t.Parallel()
+			db := DB(t)
+
+			_, err := db.Exec(
+				`INSERT INTO users (did, handle, pds_url) VALUES ($1, $2, $3)`,
+				sharedDID, UniqueID(t)+".test", "http://pds.invalid")
+			require.NoError(t, err, "each clone starts empty, so this insert must succeed")
+			written.Add(1)
+
+			WaitFor(t, 30*time.Second, func() (bool, error) {
+				return written.Load() == 2, nil
+			}, WithDescription("both clones to have written the shared DID"))
+
+			var count int
+			require.NoError(t, db.QueryRow(
+				`SELECT count(*) FROM users WHERE did = $1`, sharedDID).Scan(&count))
+			assert.Equal(t, 1, count, "a clone must only ever see its own writes")
+		})
+	}
+}
+
+func TestDB_CarriesTheMigratedSchema(t *testing.T) {
+	db := DB(t)
+
+	var applied int
+	require.NoError(t, db.QueryRow(`SELECT count(*) FROM goose_db_version`).Scan(&applied))
+	assert.Greater(t, applied, 1, "the clone should arrive fully migrated")
+
+	// A table from the last migration, not just the first: a template built
+	// from a truncated migration run would still pass a users-table check.
+	var exists sql.NullString
+	require.NoError(t, db.QueryRow(
+		`SELECT to_regclass('public.jetstream_record_revs')::text`).Scan(&exists))
+	assert.True(t, exists.Valid, "expected the full migration set to have been applied")
+
+	var rows int
+	require.NoError(t, db.QueryRow(`SELECT count(*) FROM users`).Scan(&rows))
+	assert.Zero(t, rows, "a clone must arrive with no data")
+}
+
+func TestDB_CleanupDropsTheClone(t *testing.T) {
+	var name string
+
+	t.Run("inner", func(t *testing.T) {
+		db := DB(t)
+		require.NoError(t, db.QueryRow(`SELECT current_database()`).Scan(&name))
+		require.Regexp(t, cloneNamePattern, name)
+		require.True(t, databaseExists(t, name), "the clone should exist while its test runs")
+	})
+
+	// Subtest cleanups have run by the time t.Run returns.
+	assert.False(t, databaseExists(t, name), "the clone should be dropped when its test ends")
+}
+
+func TestDB_CleanupIsNotBlockedByAQueryStillRunning(t *testing.T) {
+	// The teardown ordering hazard, stated as a test.
+	//
+	// sql.DB.Close waits for in-flight queries. If cleanup closed the pool
+	// before dropping, the 60-second query below would hold teardown for a
+	// minute and the drop would happen — if at all — long after the test was
+	// supposedly finished. Dropping first with WITH (FORCE) terminates the
+	// clone's backends, and Close then finds only dead connections.
+	//
+	// Deliberately NOT registered as a t.Cleanup: cleanups run last-in-first-out,
+	// so anything registered here would run BEFORE testkit's and release the
+	// connection, which is what made an earlier version of this test vacuous.
+	var name string
+	queryReturned := make(chan struct{})
+
+	start := time.Now()
+	t.Run("inner", func(t *testing.T) {
+		db := DB(t)
+		require.NoError(t, db.QueryRow(`SELECT current_database()`).Scan(&name))
+
+		queryStarted := make(chan struct{})
+		go func() {
+			defer close(queryReturned)
+			conn, err := db.Conn(context.Background())
+			if err != nil {
+				close(queryStarted)
+				return
+			}
+			defer func() { _ = conn.Close() }()
+			close(queryStarted)
+			// Killed by the FORCE drop; that is the point.
+			_, _ = conn.ExecContext(context.Background(), `SELECT pg_sleep(60)`)
+		}()
+		<-queryStarted
+
+		// Make sure the query is actually executing server-side before the test
+		// ends, otherwise this proves nothing.
+		WaitFor(t, 10*time.Second, func() (bool, error) {
+			admin, err := adminDB()
+			if err != nil {
+				return false, err
+			}
+			var running int
+			if err := admin.QueryRow(
+				`SELECT count(*) FROM pg_stat_activity WHERE datname = $1 AND query LIKE 'SELECT pg_sleep%'`,
+				name).Scan(&running); err != nil {
+				return false, err
+			}
+			return running > 0, nil
+		}, WithDescription("the long query to be running on the clone"))
+	})
+	elapsed := time.Since(start)
+
+	assert.False(t, databaseExists(t, name), "the clone should be dropped despite the running query")
+	assert.Less(t, elapsed, 30*time.Second,
+		"teardown must not wait out the 60s query; it took %s", elapsed)
+
+	select {
+	case <-queryReturned:
+	case <-time.After(30 * time.Second):
+		t.Fatal("the terminated query never returned")
+	}
+}
+
+func TestDB_ProvisionsTheTemplateWhenItIsMissing(t *testing.T) {
+	// The ad-hoc path: someone runs `go test ./internal/foo` without the
+	// Makefile having prepared anything. testkit has to notice and provision.
+	dropTemplate(t)
+	resetTemplateVerification()
+	require.False(t, databaseExists(t, TemplateName()))
+
+	db := DB(t)
+
+	var count int
+	require.NoError(t, db.QueryRow(`SELECT count(*) FROM users`).Scan(&count))
+	assert.True(t, databaseExists(t, TemplateName()), "the template should have been rebuilt on demand")
+}
+
+func TestDB_RebuildsTheTemplateWhenTheMigrationsChange(t *testing.T) {
+	require.NoError(t, EnsureTemplate(context.Background()))
+
+	// Stand in for "a migration was added since this template was built", and
+	// leave a marker so the rebuild can be told apart from a re-stamp.
+	withTemplate(t, func(tdb *sql.DB) {
+		_, err := tdb.Exec(`UPDATE ` + pq.QuoteIdentifier(stampTable) +
+			` SET migrations_hash = 'stale-fingerprint'`)
+		require.NoError(t, err)
+		_, err = tdb.Exec(`CREATE TABLE testkit_staleness_marker (id int)`)
+		require.NoError(t, err)
+	})
+
+	resetTemplateVerification()
+	db := DB(t)
+
+	var marker sql.NullString
+	require.NoError(t, db.QueryRow(
+		`SELECT to_regclass('public.testkit_staleness_marker')::text`).Scan(&marker))
+	assert.False(t, marker.Valid,
+		"a stale template must be dropped and rebuilt, not patched in place")
+
+	want, err := MigrationsHash()
+	require.NoError(t, err)
+	var stamped string
+	require.NoError(t, db.QueryRow(
+		`SELECT migrations_hash FROM `+pq.QuoteIdentifier(stampTable)).Scan(&stamped))
+	assert.Equal(t, want, stamped, "the rebuilt template must be stamped with the current migrations")
+}
+
+func TestTemplateStatus(t *testing.T) {
+	_, err := ProvisionTemplate(context.Background(), false)
+	require.NoError(t, err)
+
+	exists, current, err := TemplateStatus(context.Background())
+	require.NoError(t, err)
+	assert.True(t, exists)
+	assert.True(t, current)
+
+	dropTemplate(t)
+	exists, current, err = TemplateStatus(context.Background())
+	require.NoError(t, err)
+	assert.False(t, exists)
+	assert.False(t, current)
+
+	resetTemplateVerification()
+	require.NoError(t, EnsureTemplate(context.Background()))
+}
+
+func TestMigrationsHash_IsStable(t *testing.T) {
+	first, err := MigrationsHash()
+	require.NoError(t, err)
+	second, err := MigrationsHash()
+	require.NoError(t, err)
+
+	assert.Equal(t, first, second)
+	assert.Len(t, first, 64, "expected a hex sha256")
+}
+
+func TestHashMigrations_ChangesWithEveryKindOfChange(t *testing.T) {
+	// The digest is the only thing standing between an edited migration and a
+	// stale template being cloned into every test, so "it returns 64 hex
+	// characters" is not the property worth asserting. Each case below is a
+	// change that must not go unnoticed.
+	base := fstest.MapFS{
+		"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+		"002_posts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+	}
+	baseHash, err := HashMigrations(base)
+	require.NoError(t, err)
+
+	t.Run("identical content hashes identically", func(t *testing.T) {
+		same := fstest.MapFS{
+			"002_posts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+		}
+		got, err := HashMigrations(same)
+		require.NoError(t, err)
+		assert.Equal(t, baseHash, got,
+			"the walk is lexical, so map ordering must not affect the digest")
+	})
+
+	t.Run("edited content changes the digest", func(t *testing.T) {
+		edited := fstest.MapFS{
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text NOT NULL);")},
+			"002_posts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+		}
+		got, err := HashMigrations(edited)
+		require.NoError(t, err)
+		assert.NotEqual(t, baseHash, got)
+	})
+
+	t.Run("a renamed file changes the digest", func(t *testing.T) {
+		renamed := fstest.MapFS{
+			"001_accounts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+			"002_posts.sql":    &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+		}
+		got, err := HashMigrations(renamed)
+		require.NoError(t, err)
+		assert.NotEqual(t, baseHash, got, "goose orders by filename, so a rename is a schema change")
+	})
+
+	t.Run("an added migration changes the digest", func(t *testing.T) {
+		added := fstest.MapFS{
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+			"002_posts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+			"003_votes.sql": &fstest.MapFile{Data: []byte("CREATE TABLE votes (uri text);")},
+		}
+		got, err := HashMigrations(added)
+		require.NoError(t, err)
+		assert.NotEqual(t, baseHash, got)
+	})
+
+	t.Run("a removed migration changes the digest", func(t *testing.T) {
+		removed := fstest.MapFS{
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+		}
+		got, err := HashMigrations(removed)
+		require.NoError(t, err)
+		assert.NotEqual(t, baseHash, got)
+	})
+
+	t.Run("content moved across a file boundary changes the digest", func(t *testing.T) {
+		// Without the length prefix in the digest, concatenating the files in
+		// order would produce the same byte stream and the same hash.
+		shifted := fstest.MapFS{
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);CREATE")},
+			"002_posts.sql": &fstest.MapFile{Data: []byte(" TABLE posts (uri text);")},
+		}
+		got, err := HashMigrations(shifted)
+		require.NoError(t, err)
+		assert.NotEqual(t, baseHash, got)
+	})
+
+	t.Run("non-sql files are ignored", func(t *testing.T) {
+		withNoise := fstest.MapFS{
+			"001_users.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users (did text);")},
+			"002_posts.sql": &fstest.MapFile{Data: []byte("CREATE TABLE posts (uri text);")},
+			"embed.go":      &fstest.MapFile{Data: []byte("package migrations")},
+			"README.md":     &fstest.MapFile{Data: []byte("# migrations")},
+		}
+		got, err := HashMigrations(withNoise)
+		require.NoError(t, err)
+		assert.Equal(t, baseHash, got,
+			"embed.go sits in the migrations directory and changes for unrelated reasons")
+	})
+}
+
+func TestSweepOrphanClones(t *testing.T) {
+	ctx := context.Background()
+	require.NoError(t, EnsureTemplate(ctx))
+
+	// An orphan: a clone name whose embedded timestamp is two hours old, the
+	// shape left behind by a test binary that was killed before its cleanup.
+	orphan := fmt.Sprintf("%s%s_%s_orphan",
+		ClonePrefix,
+		strconv.FormatInt(time.Now().Add(-2*time.Hour).Unix(), 36),
+		RunPrefix())
+	createDatabase(t, orphan)
+	t.Cleanup(func() { _ = dropDatabase(ctx, orphan) })
+
+	// A clone from a run that is still going: same prefix, current timestamp.
+	// Dropping this one out from under a live test is the failure the age
+	// bound exists to prevent.
+	live := newCloneName()
+	createDatabase(t, live)
+	t.Cleanup(func() { _ = dropDatabase(ctx, live) })
+
+	result, err := SweepOrphanClones(ctx, time.Hour)
+	require.NoError(t, err)
+
+	assert.Contains(t, result.Dropped, orphan)
+	assert.NotContains(t, result.Dropped, live)
+	assert.False(t, databaseExists(t, orphan))
+	assert.True(t, databaseExists(t, live), "a clone younger than the cutoff must survive")
+}
+
+func TestSweepOrphanClones_SkipsAnOldCloneThatIsStillInUse(t *testing.T) {
+	// Age is evidence of orphanhood, not proof of it. A long-running E2E test
+	// legitimately holds its clone for over an hour, and killing that database
+	// mid-run would produce a baffling failure in a suite this sweep has no
+	// business touching. So an old clone with a live session is left alone —
+	// and the drop deliberately omits WITH (FORCE), which would terminate
+	// exactly the sessions whose existence is the evidence.
+	ctx := context.Background()
+
+	busy := fmt.Sprintf("%s%s_%s_busy",
+		ClonePrefix,
+		strconv.FormatInt(time.Now().Add(-3*time.Hour).Unix(), 36),
+		RunPrefix())
+	createDatabase(t, busy)
+	t.Cleanup(func() { _ = dropDatabase(ctx, busy) })
+
+	held, err := openDatabase(busy, 1)
+	require.NoError(t, err)
+	conn, err := held.Conn(ctx)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = conn.Close()
+		_ = held.Close()
+	})
+	require.NoError(t, conn.PingContext(ctx))
+
+	result, err := SweepOrphanClones(ctx, time.Hour)
+	require.NoError(t, err, "one undroppable clone must not fail the sweep")
+
+	assert.NotContains(t, result.Dropped, busy)
+	assert.Contains(t, result.Skipped, busy, "an in-use clone should be reported, not dropped")
+	assert.True(t, databaseExists(t, busy))
+}
+
+func TestSweepOrphanClones_KeepsGoingAfterAFailure(t *testing.T) {
+	// The gate must not turn red because one leftover database refused to go.
+	// A clone that acquires a session between the idle check and the DROP is
+	// the realistic version of this; it is simulated here by holding a session
+	// open, which makes the un-FORCEd drop fail if the check is bypassed.
+	ctx := context.Background()
+
+	dropped := fmt.Sprintf("%s%s_%s_ok",
+		ClonePrefix, strconv.FormatInt(time.Now().Add(-2*time.Hour).Unix(), 36), RunPrefix())
+	createDatabase(t, dropped)
+	t.Cleanup(func() { _ = dropDatabase(ctx, dropped) })
+
+	result, err := SweepOrphanClones(ctx, time.Hour)
+	require.NoError(t, err)
+	assert.Contains(t, result.Dropped, dropped)
+	assert.Empty(t, result.Failed)
+}
+
+func TestValidateTemplateName(t *testing.T) {
+	const maintenance = "coves_test"
+
+	t.Run("accepts the default and other test-namespaced names", func(t *testing.T) {
+		for _, name := range []string{"coves_test_template", "testkit_template", "my_test_db"} {
+			assert.NoError(t, validateTemplateName(name, maintenance), "should accept %q", name)
+		}
+	})
+
+	// The whole point of the rail: this name reaches DROP DATABASE ... WITH
+	// (FORCE), and it comes from an environment variable. A typo must not be
+	// able to destroy a real database.
+	t.Run("refuses names outside the test namespace", func(t *testing.T) {
+		for _, name := range []string{
+			"coves_dev",    // the development database
+			"coves_prod",   // the one that would end a career
+			"postgres",     // the maintenance database of last resort
+			"template0",    // Postgres' own templates
+			"template1",    //
+			"coves_test",   // the maintenance database testkit connects to
+			"tkclone_abc",  // the per-test clone prefix; the sweep would eat it
+			"Coves_Test",   // not a legal lowercase identifier
+			"drop table x", // not an identifier at all
+			"9test",        // must start with a letter
+			"",             //
+		} {
+			err := validateTemplateName(name, maintenance)
+			require.Error(t, err, "validateTemplateName must refuse %q", name)
+			assert.Contains(t, strings.ToLower(err.Error()), "template database")
+		}
+	})
+
+	t.Run("refuses an over-long identifier", func(t *testing.T) {
+		assert.Error(t, validateTemplateName("test_"+strings.Repeat("x", 60), maintenance))
+	})
+}
+
+func TestParallelBudget(t *testing.T) {
+	budget, err := ParallelBudget(context.Background())
+	require.NoError(t, err)
+
+	// The compose files set max_connections=200, so the budget should be
+	// comfortably above one but never unbounded.
+	assert.GreaterOrEqual(t, budget, 1)
+	assert.LessOrEqual(t, budget, 64)
+}
+
+func TestDescribeLockHolders_NamesTheBlockingSession(t *testing.T) {
+	// A lock timeout that says only "timed out" leaves the next person with no
+	// way to find who is holding it. This is the part of that message that
+	// makes it actionable, so it gets a test of its own — the two-minute
+	// timeout around it does not.
+	ctx := context.Background()
+
+	holding := make(chan struct{})
+	released := make(chan struct{})
+	go func() {
+		_ = withAdvisoryLock(ctx, false, func(context.Context, *sql.Conn) error {
+			close(holding)
+			<-released
+			return nil
+		})
+	}()
+	<-holding
+	defer close(released)
+
+	var description string
+	WaitFor(t, 10*time.Second, func() (bool, error) {
+		description = describeLockHolders(ctx)
+		return description != "", nil
+	}, WithDescription("the advisory lock to show up in pg_locks"))
+
+	assert.Contains(t, description, "currently held by")
+	assert.Regexp(t, `pid \d+`, description)
+}
+
+func TestIsSourceInUse(t *testing.T) {
+	assert.True(t, isSourceInUse(&pq.Error{Code: sqlstateObjectInUse}))
+	assert.True(t, isSourceInUse(fmt.Errorf("wrapped: %w", &pq.Error{Code: sqlstateObjectInUse})),
+		"the classifier must see through wrapping, since cloneTemplate wraps")
+	assert.False(t, isSourceInUse(&pq.Error{Code: "42P04"}), "duplicate_database is not retryable")
+	assert.False(t, isSourceInUse(errors.New("connection refused")))
+}
+
+func TestDropDatabase_RefusesAnythingButAClone(t *testing.T) {
+	ctx := context.Background()
+	for _, name := range []string{
+		"coves_dev",
+		"coves_test",
+		TemplateName(),
+		"postgres",
+		"tkclone", // the prefix alone, without the trailing underscore
+		"",
+	} {
+		err := dropDatabase(ctx, name)
+		require.Error(t, err, "dropDatabase must refuse %q", name)
+		assert.Contains(t, err.Error(), "not a testkit clone name")
+	}
+}
+
+func TestCloneCreatedAt(t *testing.T) {
+	name := newCloneName()
+	created, ok := cloneCreatedAt(name)
+	require.True(t, ok, "newCloneName must produce a sweepable name: %q", name)
+	assert.WithinDuration(t, time.Now(), created, 5*time.Second)
+
+	for _, bogus := range []string{
+		"coves_dev",
+		"tkclone_notatimestamp",               // too few segments
+		"tkclone_zzz_" + RunPrefix() + "_1_2", // too many segments
+	} {
+		_, ok := cloneCreatedAt(bogus)
+		assert.False(t, ok, "%q must not be treated as a sweepable clone", bogus)
+	}
+}
+
+func TestDB_FailsLoudlyWhenPostgresIsUnreachable(t *testing.T) {
+	// Missing infrastructure is a failure, not a skip — and the message has to
+	// say where testkit looked and how to start it, because "connection
+	// refused" on its own has cost more debugging time than any other line in
+	// this suite.
+	t.Setenv("POSTGRES_TEST_PORT", "1")
+	t.Setenv("POSTGRES_TEST_HOST", "127.0.0.1")
+
+	restore := swapSingletons(sync.OnceValue(loadEndpoints), sync.OnceValues(openAdminDB))
+	defer restore()
+
+	f := &fakeT{}
+	runIsolated(func() { DB(f) })
+
+	require.True(t, f.failed())
+	msg := f.message()
+	assert.Contains(t, msg, "testkit.DB")
+	assert.Contains(t, msg, "127.0.0.1:1")
+	assert.Contains(t, msg, "make test", "the failure must say how to start the database")
+	assert.NotContains(t, msg, Endpoints().Postgres.Password,
+		"credentials must never reach a failure message")
+}
+
+func TestTemplateName_PanicsOnAnUnsafeOverride(t *testing.T) {
+	// The environment variable reaches DROP DATABASE. Carrying on with a value
+	// that names a real database is the one outcome that must be impossible, so
+	// the failure is loud and immediate rather than an error some caller might
+	// log and ignore.
+	t.Setenv("POSTGRES_TEST_TEMPLATE", "coves_dev")
+
+	restore := swapSingletons(sync.OnceValue(loadEndpoints), sync.OnceValues(openAdminDB))
+	defer restore()
+
+	var panicked any
+	func() {
+		defer func() { panicked = recover() }()
+		_ = TemplateName()
+	}()
+	require.NotNil(t, panicked, "TemplateName must not return an unsafe name")
+	assert.Contains(t, fmt.Sprint(panicked), `refusing to use "coves_dev" as the template database`)
+
+	// And the provisioning path refuses rather than panicking, so DB(t) reports
+	// it as an ordinary test failure.
+	_, err := ProvisionTemplate(context.Background(), false)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "coves_dev")
+
+	f := &fakeT{}
+	runIsolated(func() { DB(f) })
+	require.True(t, f.failed())
+	assert.Contains(t, f.message(), "coves_dev")
+}
+
+// ---------------------------------------------------------------------------
+// helpers
+// ---------------------------------------------------------------------------
+
+func databaseExists(t *testing.T, name string) bool {
+	t.Helper()
+	db, err := adminDB()
+	require.NoError(t, err)
+	var exists bool
+	require.NoError(t, db.QueryRow(
+		`SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname = $1)`, name).Scan(&exists))
+	return exists
+}
+
+func createDatabase(t *testing.T, name string) {
+	t.Helper()
+	require.Regexp(t, cloneNamePattern, name, "tests only create clone-shaped databases")
+	db, err := adminDB()
+	require.NoError(t, err)
+	_, err = db.Exec("CREATE DATABASE " + pq.QuoteIdentifier(name))
+	require.NoError(t, err)
+}
+
+// dropTemplate removes the template database directly, UNDER THE EXCLUSIVE
+// LOCK.
+//
+// The lock is not ceremony. Dropping the template is exactly what the lock
+// protects against a concurrent clone, and a test helper that skips it works
+// only for as long as the suite runs one package at a time — which is a
+// property phase 3 deliberately removes. testkit's own tests are the first
+// place that regression would appear and the last place anyone would look.
+//
+// It bypasses the clone-prefix rail in dropDatabase, which is the one thing
+// only a test may do.
+func dropTemplate(t *testing.T) {
+	t.Helper()
+	require.NoError(t, withAdvisoryLock(context.Background(), false,
+		func(ctx context.Context, conn *sql.Conn) error {
+			_, err := conn.ExecContext(ctx,
+				"DROP DATABASE IF EXISTS "+pq.QuoteIdentifier(TemplateName())+" WITH (FORCE)")
+			return err
+		}))
+}
+
+// withTemplate opens a short-lived connection to the template, under the
+// exclusive lock for the same reason dropTemplate takes it: CREATE DATABASE ...
+// TEMPLATE refuses to run while any session is connected to the source, so a
+// concurrent clone must not be in flight.
+func withTemplate(t *testing.T, fn func(*sql.DB)) {
+	t.Helper()
+	require.NoError(t, withAdvisoryLock(context.Background(), false,
+		func(ctx context.Context, conn *sql.Conn) error {
+			db, err := openDatabase(TemplateName(), 1)
+			if err != nil {
+				return err
+			}
+			defer func() { _ = db.Close() }()
+			fn(db)
+			return nil
+		}))
+}
diff --git a/tests/testkit/fixtures.go b/tests/testkit/fixtures.go
new file mode 100644
index 0000000..e6ddd6c
--- /dev/null
+++ b/tests/testkit/fixtures.go
@@ -0,0 +1,234 @@
+package testkit
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/binary"
+	"fmt"
+	"image"
+	"image/color"
+	"image/jpeg"
+	"image/png"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+)
+
+// ---------------------------------------------------------------------------
+// Identifiers
+// ---------------------------------------------------------------------------
+
+// MaxIDLength is the hard cap on a generated identifier.
+//
+// It comes from the PDS: a handle's local label — the part before the first dot
+// in "alice.local.coves.dev" — is capped at 18 characters, and exceeding it
+// fails account creation with a "Handle too long" error that reads like a test
+// bug rather than a naming-scheme bug. Every generated ID is kept inside that
+// budget so any of them can be used as a handle label.
+const MaxIDLength = 18
+
+// runPrefixLength is the width of the per-process random prefix: one letter
+// plus six base36 characters.
+//
+// Uniqueness across processes is probabilistic, and this is the honest
+// statement of it: 26 × 36^6 ≈ 5.7×10^10 possible prefixes, so by the birthday
+// bound the chance that any two of 10,000 runs against the same persistent PDS
+// share a prefix is under 0.1%. Within a run, collisions are impossible rather
+// than unlikely — the atomic counter guarantees it. Seven characters is the
+// most the 18-character handle budget affords while leaving room for a readable
+// prefix in UniqueIDWithPrefix.
+const runPrefixLength = 7
+
+var (
+	runPrefixOnce = sync.OnceValue(newRunPrefix)
+	idCounter     = newCounter()
+)
+
+// counter is a process-wide monotonic sequence. Wrapped rather than used bare
+// so the two independent sequences (identifiers, clone names) cannot be
+// accidentally shared.
+type counter struct{ n atomic.Uint64 }
+
+func newCounter() *counter { return &counter{} }
+
+func (c *counter) next() uint64 { return c.n.Add(1) }
+
+// RunPrefix returns this process's random identifier prefix.
+//
+// It is also embedded in every clone database name, which is what lets
+// scripts/test-db-prepare.sh tell one run's leftovers from another's.
+func RunPrefix() string { return runPrefixOnce() }
+
+// newRunPrefix draws a random prefix that starts with a letter.
+//
+// A counter alone is not enough, which the suite learned the hard way: counters
+// restart at 1 in every process, and the local PDS keeps accounts across runs,
+// so two `make test-e2e` invocations would try to register the same handle. A
+// timestamp alone is not enough either — second granularity collides within a
+// single fast run, and a ten-digit Unix timestamp eats most of the 18-character
+// budget.
+//
+// The leading letter matters: a handle's label may not begin with a digit, and
+// a purely numeric label is rejected outright.
+func newRunPrefix() string {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		// crypto/rand failing means the process has no entropy source. There is
+		// no sensible fallback and a duplicate-handle failure three layers away
+		// would be far harder to diagnose than this.
+		panic(fmt.Sprintf("testkit: reading random bytes for the run prefix: %v", err))
+	}
+	letter := byte('a' + b[0]%26)
+	// Reduced to a fixed width, so every generated ID has the same shape and
+	// the budget arithmetic below is exact. Drawn from 64 bits rather than 32
+	// so the modulo reduction's bias is immeasurable rather than a factor of
+	// two across half the range.
+	const span = 36 * 36 * 36 * 36 * 36 * 36 // 36^6
+	n := binary.BigEndian.Uint64(b[1:9]) % span
+	digits := strconv.FormatUint(n, 36)
+	return string(letter) + strings.Repeat("0", runPrefixLength-1-len(digits)) + digits
+}
+
+// UniqueID returns an identifier unique across every process, machine and run
+// that shares this Postgres or PDS.
+//
+// It is the only handle/ID generator the suite may use. Hand-rolled
+// time.Now().Unix() identifiers are why handles collided between runs against
+// the persistent dev PDS, and why two tests started in the same second
+// occasionally fought over one account.
+//
+// The result is lowercase alphanumeric, starts with a letter, and is at most
+// MaxIDLength characters, so it is directly usable as a PDS handle label, a
+// community name, an email local part, or a database-safe suffix.
+func UniqueID(t TestingT) string {
+	t.Helper()
+	id := RunPrefix() + strconv.FormatUint(idCounter.next(), 36)
+	if len(id) > MaxIDLength {
+		t.Fatalf("testkit.UniqueID: generated %q (%d chars), over the %d-char PDS handle budget",
+			id, len(id), MaxIDLength)
+		return ""
+	}
+	return id
+}
+
+// UniqueIDWithPrefix returns a unique identifier that starts with a readable
+// prefix, so a leftover row or account says which test made it.
+//
+// The prefix is sanitised to lowercase alphanumerics and truncated as needed:
+// uniqueness wins over readability, because a truncated prefix costs a
+// debugging hint while a truncated random suffix costs a collision.
+func UniqueIDWithPrefix(t TestingT, prefix string) string {
+	t.Helper()
+	suffix := RunPrefix() + strconv.FormatUint(idCounter.next(), 36)
+	if len(suffix) > MaxIDLength {
+		t.Fatalf("testkit.UniqueIDWithPrefix: generated %q (%d chars), over the %d-char PDS handle budget",
+			suffix, len(suffix), MaxIDLength)
+		return ""
+	}
+
+	clean := sanitizeLabel(prefix)
+	if budget := MaxIDLength - len(suffix); len(clean) > budget {
+		clean = clean[:max(budget, 0)]
+	}
+	return clean + suffix
+}
+
+// sanitizeLabel reduces an arbitrary string to lowercase alphanumerics and
+// guarantees it starts with a letter, so the result stays a legal handle label.
+func sanitizeLabel(s string) string {
+	var b strings.Builder
+	for _, r := range strings.ToLower(s) {
+		switch {
+		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
+			b.WriteRune(r)
+		}
+	}
+	out := b.String()
+	if out != "" && out[0] >= '0' && out[0] <= '9' {
+		out = "t" + out
+	}
+	return out
+}
+
+// ---------------------------------------------------------------------------
+// Images
+// ---------------------------------------------------------------------------
+
+// TestPNG returns a decodable PNG of the requested size.
+//
+// The pixels are a deterministic gradient rather than a single flat colour:
+// flat images compress to a handful of bytes and encode to a nearly empty JPEG,
+// which is a poor stand-in for a real upload when the code under test is
+// resizing, sniffing, or enforcing a size budget.
+func TestPNG(width, height int) []byte {
+	var buf bytes.Buffer
+	if err := png.Encode(&buf, gradientImage(width, height)); err != nil {
+		panic(fmt.Sprintf("testkit.TestPNG(%d, %d): %v", width, height, err))
+	}
+	return buf.Bytes()
+}
+
+// TestPNGColor returns a decodable PNG filled with a single colour, for tests
+// that assert on pixel values.
+func TestPNGColor(width, height int, c color.Color) []byte {
+	var buf bytes.Buffer
+	if err := png.Encode(&buf, solidImage(width, height, c)); err != nil {
+		panic(fmt.Sprintf("testkit.TestPNGColor(%d, %d): %v", width, height, err))
+	}
+	return buf.Bytes()
+}
+
+// TestJPEG returns a decodable JPEG of the requested size, encoded at quality
+// 90 to match what the image pipeline produces.
+func TestJPEG(width, height int) []byte {
+	var buf bytes.Buffer
+	if err := jpeg.Encode(&buf, gradientImage(width, height), &jpeg.Options{Quality: 90}); err != nil {
+		panic(fmt.Sprintf("testkit.TestJPEG(%d, %d): %v", width, height, err))
+	}
+	return buf.Bytes()
+}
+
+// TestJPEGColor returns a decodable JPEG filled with a single colour.
+func TestJPEGColor(width, height int, c color.Color) []byte {
+	var buf bytes.Buffer
+	if err := jpeg.Encode(&buf, solidImage(width, height, c), &jpeg.Options{Quality: 90}); err != nil {
+		panic(fmt.Sprintf("testkit.TestJPEGColor(%d, %d): %v", width, height, err))
+	}
+	return buf.Bytes()
+}
+
+func gradientImage(width, height int) *image.RGBA {
+	img := newImage(width, height)
+	for y := 0; y < height; y++ {
+		for x := 0; x < width; x++ {
+			img.Set(x, y, color.RGBA{
+				R: uint8(x * 255 / width),
+				G: uint8(y * 255 / height),
+				B: 128,
+				A: 255,
+			})
+		}
+	}
+	return img
+}
+
+func solidImage(width, height int, c color.Color) *image.RGBA {
+	img := newImage(width, height)
+	for y := 0; y < height; y++ {
+		for x := 0; x < width; x++ {
+			img.Set(x, y, c)
+		}
+	}
+	return img
+}
+
+func newImage(width, height int) *image.RGBA {
+	if width <= 0 || height <= 0 {
+		// A zero-sized image encodes to bytes that no decoder accepts, so the
+		// failure would otherwise surface as an unrelated "invalid image" from
+		// whatever consumed it.
+		panic(fmt.Sprintf("testkit: image dimensions must be positive, got %dx%d", width, height))
+	}
+	return image.NewRGBA(image.Rect(0, 0, width, height))
+}
diff --git a/tests/testkit/fixtures_test.go b/tests/testkit/fixtures_test.go
new file mode 100644
index 0000000..991b147
--- /dev/null
+++ b/tests/testkit/fixtures_test.go
@@ -0,0 +1,188 @@
+package testkit
+
+import (
+	"bytes"
+	"image"
+	"image/color"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUniqueID_Shape(t *testing.T) {
+	id := UniqueID(t)
+
+	assert.LessOrEqual(t, len(id), MaxIDLength,
+		"a longer id is rejected by the PDS as 'Handle too long'")
+	assert.NotEmpty(t, id)
+	assertHandleSafe(t, id)
+	assert.True(t, strings.HasPrefix(id, RunPrefix()),
+		"ids must carry the run prefix so leftovers on a persistent PDS are attributable")
+}
+
+func TestUniqueID_UniqueUnderConcurrency(t *testing.T) {
+	const (
+		goroutines = 50
+		perRoutine = 200
+	)
+
+	ids := make([][]string, goroutines)
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+	for g := 0; g < goroutines; g++ {
+		go func(g int) {
+			defer wg.Done()
+			batch := make([]string, perRoutine)
+			for i := range batch {
+				batch[i] = UniqueID(t)
+			}
+			ids[g] = batch
+		}(g)
+	}
+	wg.Wait()
+
+	// Checked by hand rather than through require: 10,000 assertions cost more
+	// wall clock than the thing under test.
+	seen := make(map[string]struct{}, goroutines*perRoutine)
+	for _, batch := range ids {
+		for _, id := range batch {
+			if _, dup := seen[id]; dup {
+				t.Fatalf("UniqueID returned %q twice", id)
+			}
+			if len(id) > MaxIDLength {
+				t.Fatalf("UniqueID returned %q (%d chars), over the handle budget", id, len(id))
+			}
+			seen[id] = struct{}{}
+		}
+	}
+	assert.Len(t, seen, goroutines*perRoutine)
+}
+
+func TestUniqueID_RunPrefixIsStableAndRandom(t *testing.T) {
+	assert.Equal(t, RunPrefix(), RunPrefix(), "the prefix identifies the run, so it must not change mid-run")
+	assert.Len(t, RunPrefix(), runPrefixLength)
+	assertHandleSafe(t, RunPrefix())
+
+	// Two draws from newRunPrefix should differ. This is what makes two
+	// consecutive runs against the same persistent PDS not collide — the
+	// failure mode a bare counter has.
+	distinct := make(map[string]struct{})
+	for i := 0; i < 100; i++ {
+		distinct[newRunPrefix()] = struct{}{}
+	}
+	assert.Greater(t, len(distinct), 90, "run prefixes should be drawn at random, not derived from a counter")
+}
+
+func TestUniqueIDWithPrefix(t *testing.T) {
+	id := UniqueIDWithPrefix(t, "comm")
+	assert.True(t, strings.HasPrefix(id, "comm"), "got %q", id)
+	assert.LessOrEqual(t, len(id), MaxIDLength)
+	assertHandleSafe(t, id)
+
+	t.Run("sanitises the prefix", func(t *testing.T) {
+		id := UniqueIDWithPrefix(t, "My Test-Community!")
+		assert.True(t, strings.HasPrefix(id, "mytest"), "got %q", id)
+		assertHandleSafe(t, id)
+	})
+
+	t.Run("clamps an over-long prefix without eating the entropy", func(t *testing.T) {
+		id := UniqueIDWithPrefix(t, strings.Repeat("x", 100))
+		assert.LessOrEqual(t, len(id), MaxIDLength)
+		assertHandleSafe(t, id)
+		// The readable prefix is what gets truncated, never the random suffix:
+		// a shortened prefix costs a debugging hint, a shortened suffix costs a
+		// collision.
+		assert.Contains(t, id, RunPrefix(), "the run prefix must survive truncation")
+		assert.True(t, strings.HasPrefix(id, "x"), "got %q", id)
+	})
+
+	t.Run("keeps a digit-leading prefix legal as a handle label", func(t *testing.T) {
+		id := UniqueIDWithPrefix(t, "42")
+		assertHandleSafe(t, id)
+	})
+
+	t.Run("an empty prefix still yields a usable id", func(t *testing.T) {
+		id := UniqueIDWithPrefix(t, "!!!")
+		assertHandleSafe(t, id)
+		assert.True(t, strings.HasPrefix(id, RunPrefix()))
+	})
+
+	t.Run("stays unique", func(t *testing.T) {
+		seen := make(map[string]struct{}, 500)
+		for i := 0; i < 500; i++ {
+			id := UniqueIDWithPrefix(t, "post")
+			require.NotContains(t, seen, id)
+			seen[id] = struct{}{}
+		}
+	})
+}
+
+// assertHandleSafe checks the invariants a PDS handle label must satisfy:
+// lowercase alphanumerics only, and not starting with a digit.
+func assertHandleSafe(t *testing.T, id string) {
+	t.Helper()
+	require.NotEmpty(t, id)
+	require.False(t, id[0] >= '0' && id[0] <= '9',
+		"%q starts with a digit, which is not a legal handle label", id)
+	for i, r := range id {
+		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
+			continue
+		}
+		t.Fatalf("%q contains %q at index %d; ids must be lowercase alphanumeric", id, r, i)
+	}
+}
+
+func TestTestPNG_DecodesAtTheRequestedSize(t *testing.T) {
+	data := TestPNG(64, 48)
+
+	img, format, err := image.Decode(bytes.NewReader(data))
+	require.NoError(t, err, "the bytes must be a real PNG, not a hand-rolled header")
+	assert.Equal(t, "png", format)
+	assert.Equal(t, 64, img.Bounds().Dx())
+	assert.Equal(t, 48, img.Bounds().Dy())
+}
+
+func TestTestJPEG_DecodesAtTheRequestedSize(t *testing.T) {
+	data := TestJPEG(120, 90)
+
+	img, format, err := image.Decode(bytes.NewReader(data))
+	require.NoError(t, err)
+	assert.Equal(t, "jpeg", format)
+	assert.Equal(t, 120, img.Bounds().Dx())
+	assert.Equal(t, 90, img.Bounds().Dy())
+}
+
+func TestTestPNGColor_UsesTheGivenColour(t *testing.T) {
+	want := color.RGBA{R: 12, G: 200, B: 90, A: 255}
+	img, format, err := image.Decode(bytes.NewReader(TestPNGColor(8, 8, want)))
+	require.NoError(t, err)
+	require.Equal(t, "png", format)
+
+	r, g, b, a := img.At(3, 5).RGBA()
+	assert.Equal(t, [4]uint32{0x0c0c, 0xc8c8, 0x5a5a, 0xffff}, [4]uint32{r, g, b, a})
+}
+
+func TestTestJPEGColor_DecodesAtTheRequestedSize(t *testing.T) {
+	img, format, err := image.Decode(bytes.NewReader(TestJPEGColor(16, 16, color.White)))
+	require.NoError(t, err)
+	assert.Equal(t, "jpeg", format)
+	assert.Equal(t, 16, img.Bounds().Dx())
+}
+
+func TestTestImages_AreNotDegenerate(t *testing.T) {
+	// A flat image encodes to almost nothing, which makes it a poor stand-in
+	// for an upload when the code under test resizes or enforces size limits.
+	// The gradient is what buys the difference, so that is what to assert.
+	assert.Greater(t, len(TestPNG(200, 200)), len(TestPNGColor(200, 200, color.Black)))
+	assert.Greater(t, len(TestJPEG(200, 200)), len(TestJPEGColor(200, 200, color.Black)))
+	assert.Greater(t, len(TestPNG(200, 200)), 500)
+	assert.Greater(t, len(TestJPEG(200, 200)), 500)
+}
+
+func TestTestImages_RejectNonPositiveDimensions(t *testing.T) {
+	assert.Panics(t, func() { TestPNG(0, 10) })
+	assert.Panics(t, func() { TestJPEG(10, -1) })
+}
diff --git a/tests/testkit/harness_test.go b/tests/testkit/harness_test.go
new file mode 100644
index 0000000..bffb1f4
--- /dev/null
+++ b/tests/testkit/harness_test.go
@@ -0,0 +1,139 @@
+package testkit
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+)
+
+func TestMain(m *testing.M) {
+	SilenceLogs()
+	os.Exit(m.Run())
+}
+
+// fakeT records what a testkit helper did to the test it was handed, so the
+// failure paths can be asserted on.
+//
+// This is why TestingT exists as an interface: testing.TB cannot be implemented
+// outside the standard library, so without it the only way to test "WaitFor
+// fails with the last observation attached" would be to compile and run a
+// throwaway test binary in a subprocess.
+type fakeT struct {
+	mu       sync.Mutex
+	failures []string
+	logs     []string
+	cleanups []func()
+	fatal    bool
+}
+
+func (f *fakeT) Helper()      {}
+func (f *fakeT) Name() string { return "fakeT" }
+
+func (f *fakeT) Cleanup(fn func()) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.cleanups = append(f.cleanups, fn)
+}
+
+func (f *fakeT) Logf(format string, args ...any) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.logs = append(f.logs, fmt.Sprintf(format, args...))
+}
+
+func (f *fakeT) Errorf(format string, args ...any) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.failures = append(f.failures, fmt.Sprintf(format, args...))
+}
+
+// Fatalf mirrors *testing.T: it records the failure and does not return.
+// Helpers under test rely on that — they call Fatalf and then `return`, and a
+// fake that simply returned would let them run on with invalid state.
+func (f *fakeT) Fatalf(format string, args ...any) {
+	f.mu.Lock()
+	f.failures = append(f.failures, fmt.Sprintf(format, args...))
+	f.fatal = true
+	f.mu.Unlock()
+	runtime.Goexit()
+}
+
+func (f *fakeT) failed() bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.failures) > 0
+}
+
+func (f *fakeT) message() string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return strings.Join(f.failures, "\n")
+}
+
+func (f *fakeT) runCleanups() {
+	f.mu.Lock()
+	cleanups := f.cleanups
+	f.cleanups = nil
+	f.mu.Unlock()
+	// Reverse order, as testing.T runs them.
+	for i := len(cleanups) - 1; i >= 0; i-- {
+		cleanups[i]()
+	}
+}
+
+// runIsolated runs fn on its own goroutine, so a fakeT.Fatalf inside it exits
+// that goroutine instead of the test's.
+func runIsolated(fn func()) {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		fn()
+	}()
+	<-done
+}
+
+// resetTemplateVerification forgets that this process already checked the
+// template, forcing the next EnsureTemplate to re-verify against Postgres.
+func resetTemplateVerification() {
+	templateMu.Lock()
+	defer templateMu.Unlock()
+	templateVerified = false
+}
+
+// swapSingletons rebinds the memoised process singletons and returns a function
+// that restores them.
+//
+// Every write goes through singletonMu, the same lock the accessors read under.
+// Assigning these directly from a test — as an earlier version did — is a data
+// race the moment any test runs in parallel or `-shuffle=on` reorders things,
+// and it would surface as an unrelated test failing under -race.
+func swapSingletons(endpoints func() EndpointSet, admin func() (*sql.DB, error)) func() {
+	singletonMu.Lock()
+	savedEndpoints, savedAdmin := endpointsOnce, adminDBOnce
+	savedTemplateName := templateNameOnce
+	endpointsOnce, adminDBOnce = endpoints, admin
+	// The template name is validated against the maintenance database, so it
+	// has to be re-derived whenever the endpoints change.
+	templateNameOnce = sync.OnceValues(loadTemplateName)
+	singletonMu.Unlock()
+
+	templateMu.Lock()
+	savedVerified := templateVerified
+	templateVerified = false
+	templateMu.Unlock()
+
+	return func() {
+		singletonMu.Lock()
+		endpointsOnce, adminDBOnce = savedEndpoints, savedAdmin
+		templateNameOnce = savedTemplateName
+		singletonMu.Unlock()
+
+		templateMu.Lock()
+		templateVerified = savedVerified
+		templateMu.Unlock()
+	}
+}
diff --git a/tests/testkit/testkit.go b/tests/testkit/testkit.go
new file mode 100644
index 0000000..5b6430e
--- /dev/null
+++ b/tests/testkit/testkit.go
@@ -0,0 +1,148 @@
+// Package testkit is the single test harness for the Coves suite: database
+// isolation, waiting primitives, and fixture generators, shared by every tier
+// (unit, integration, pipeline, live).
+//
+// # Dependency direction (hard rule)
+//
+// testkit imports NO domain package — nothing from internal/core/..., nothing
+// from internal/api/.... It may import infrastructure only: the database
+// driver, goose, the embedded migrations, and (later) the jetstream event
+// types.
+//
+// The reason is not taste, it is compilability. Integration tests live
+// in-package, next to the code they test: internal/core/communities has its own
+// _test.go files, and those files import testkit. If testkit imported
+// internal/core/communities, that would be an import cycle and neither package
+// would build. The rule has to hold for every domain package, so the only safe
+// version of it is "no domain imports at all".
+//
+// Domain-specific builders that need domain types belong either in that
+// domain's own _test.go files or in a small leaf <domain>test package — never
+// here.
+//
+// # Layout
+//
+//	testkit.go   — endpoints, log silencing, the TestingT contract
+//	db.go        — Postgres isolation: template-clone-per-test
+//	wait.go      — the only waiting primitives tests are allowed to use
+//	fixtures.go  — UniqueID, PNG/JPEG image bytes
+package testkit
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"log/slog"
+	"os"
+)
+
+// TestingT is the subset of *testing.T that testkit needs.
+//
+// It is an interface rather than *testing.T for one reason: testkit's own
+// failure paths have to be testable. testing.TB cannot be implemented outside
+// the standard library (it has an unexported method), so asserting that
+// WaitFor fails with the right message would otherwise require a subprocess.
+// *testing.T and *testing.B both satisfy this, so call sites are unaffected.
+//
+// Implementations must treat Fatalf as fatal — real *testing.T calls
+// runtime.Goexit, and every testkit function relies on Fatalf not returning by
+// also returning immediately afterwards.
+type TestingT interface {
+	Helper()
+	Cleanup(func())
+	Errorf(format string, args ...any)
+	Fatalf(format string, args ...any)
+	Logf(format string, args ...any)
+	Name() string
+}
+
+// PostgresEndpoint locates a Postgres server and the credentials to reach it.
+type PostgresEndpoint struct {
+	Host string
+	Port string
+	User string
+	// Password is a throwaway local credential. Never include it in a failure
+	// message: use Redacted, not URL, when building error text.
+	Password string
+	// Database is the maintenance database: the one testkit connects to in
+	// order to CREATE/DROP other databases and to hold advisory locks. It is
+	// never the database a test writes to.
+	Database string
+	SSLMode  string
+}
+
+// URL renders a libpq connection string for the named database on this server.
+func (p PostgresEndpoint) URL(database string) string {
+	return fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=%s",
+		p.User, p.Password, p.Host, p.Port, database, p.SSLMode)
+}
+
+// Redacted renders the server address without credentials, for error messages.
+func (p PostgresEndpoint) Redacted(database string) string {
+	return fmt.Sprintf("postgres://%s@%s:%s/%s", p.User, p.Host, p.Port, database)
+}
+
+// EndpointSet holds the address of every service the test stack provides.
+//
+// This is the only place test code reads infrastructure coordinates from the
+// environment. Tests that build their own base URLs are how a suite ends up
+// silently talking to a developer's dev stack — or to the public internet —
+// so endpoint literals in tests are counted as violations by
+// scripts/test-audit.sh.
+//
+// Later phases extend this struct with the PDS, PLC, Jetstream and AppView
+// addresses; the Postgres coordinates are all the database harness needs.
+type EndpointSet struct {
+	Postgres PostgresEndpoint
+}
+
+// Endpoints returns the test stack's coordinates, read from the environment
+// once per process.
+//
+// Defaults match docker-compose.dev.yml's postgres-test service as published on
+// the host, and .env.ci's values inside the hermetic stack — which share a
+// network namespace, so "localhost:5434" is correct in both places.
+//
+// The memoised loader lives in db.go alongside the other process singletons,
+// behind a lock, because testkit's own tests rebind it.
+func Endpoints() EndpointSet {
+	return loadedEndpoints()()
+}
+
+func loadEndpoints() EndpointSet {
+	return EndpointSet{
+		Postgres: PostgresEndpoint{
+			Host:     envOr("POSTGRES_TEST_HOST", "localhost"),
+			Port:     envOr("POSTGRES_TEST_PORT", "5434"),
+			User:     envOr("POSTGRES_TEST_USER", "test_user"),
+			Password: envOr("POSTGRES_TEST_PASSWORD", "test_password"),
+			Database: envOr("POSTGRES_TEST_DB", "coves_test"),
+			SSLMode:  envOr("POSTGRES_TEST_SSLMODE", "disable"),
+		},
+	}
+}
+
+func envOr(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}
+
+// SilenceLogs discards application log output when LOG_ENABLED=false, which is
+// what the CI runner and `make test` set.
+//
+// Call it from TestMain. It mirrors the per-package TestMain blocks that
+// already exist across the tree, and additionally silences slog, which those
+// blocks miss — the server logs through slog, so silencing only the standard
+// logger leaves most of the noise in place.
+//
+// This mutates process-global state, so it belongs in TestMain and nowhere
+// else: a test calling it mid-run would race any test reading log output.
+func SilenceLogs() {
+	if os.Getenv("LOG_ENABLED") != "false" {
+		return
+	}
+	log.SetOutput(io.Discard)
+	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
+}
diff --git a/tests/testkit/testkit_test.go b/tests/testkit/testkit_test.go
new file mode 100644
index 0000000..da1d5d0
--- /dev/null
+++ b/tests/testkit/testkit_test.go
@@ -0,0 +1,60 @@
+package testkit
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEndpoints_DefaultsMatchTheTestStack(t *testing.T) {
+	for _, key := range []string{
+		"POSTGRES_TEST_HOST", "POSTGRES_TEST_PORT", "POSTGRES_TEST_USER",
+		"POSTGRES_TEST_PASSWORD", "POSTGRES_TEST_DB", "POSTGRES_TEST_SSLMODE",
+	} {
+		t.Setenv(key, "")
+	}
+	endpointsOnce = sync.OnceValue(loadEndpoints)
+	t.Cleanup(func() { endpointsOnce = sync.OnceValue(loadEndpoints) })
+
+	pg := Endpoints().Postgres
+	// docker-compose.dev.yml publishes postgres-test here, and .env.ci's shared
+	// network namespace puts it at the same address inside the hermetic stack.
+	assert.Equal(t, "localhost", pg.Host)
+	assert.Equal(t, "5434", pg.Port)
+	assert.Equal(t, "test_user", pg.User)
+	assert.Equal(t, "coves_test", pg.Database)
+	assert.Equal(t, "disable", pg.SSLMode)
+}
+
+func TestEndpoints_ReadTheEnvironment(t *testing.T) {
+	t.Setenv("POSTGRES_TEST_HOST", "db.example.internal")
+	t.Setenv("POSTGRES_TEST_PORT", "6543")
+	endpointsOnce = sync.OnceValue(loadEndpoints)
+	t.Cleanup(func() { endpointsOnce = sync.OnceValue(loadEndpoints) })
+
+	pg := Endpoints().Postgres
+	assert.Equal(t, "db.example.internal", pg.Host)
+	assert.Equal(t, "6543", pg.Port)
+}
+
+func TestEndpoints_MemoisedPerProcess(t *testing.T) {
+	first := Endpoints()
+	require.Equal(t, first, Endpoints(), "endpoints are read once; a mid-run change would be invisible anyway")
+}
+
+func TestPostgresEndpoint_RedactedOmitsTheCredential(t *testing.T) {
+	pg := PostgresEndpoint{
+		Host: "localhost", Port: "5434", User: "test_user",
+		Password: "hunter2", Database: "coves_test", SSLMode: "disable",
+	}
+
+	assert.Equal(t, "postgres://test_user:hunter2@localhost:5434/coves_test?sslmode=disable",
+		pg.URL(pg.Database))
+
+	redacted := pg.Redacted("some_clone")
+	assert.NotContains(t, redacted, "hunter2", "error messages must not leak the password")
+	assert.Contains(t, redacted, "localhost:5434")
+	assert.Contains(t, redacted, "some_clone")
+}
diff --git a/tests/testkit/wait.go b/tests/testkit/wait.go
new file mode 100644
index 0000000..76ca009
--- /dev/null
+++ b/tests/testkit/wait.go
@@ -0,0 +1,217 @@
+package testkit
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+// Waiting primitives. These are the only ones tests may use: time.Sleep in a
+// *_test.go file is counted as a violation by scripts/test-audit.sh and becomes
+// a hard CI failure at the end of the migration.
+//
+// A sleep encodes a guess about how long something takes, and it is wrong in
+// both directions at once — too short on a loaded CI machine (flake), too long
+// on an idle laptop (slow suite). Worse, when a sleep-based test does fail, its
+// failure message says nothing about what the system was doing.
+//
+// WaitFor and Holds are the two shapes that actually come up:
+//
+//	WaitFor — eventually true. "the firehose delivered the record."
+//	Holds   — stays true. "the delete stayed deleted."
+//
+// Holds is not redundant. An eventually-check cannot catch a record that is
+// deleted and then resurrected by a replayed event: the check passes on the
+// first poll and never looks again.
+
+// DefaultPollInterval is how often a probe is re-evaluated. 100ms is fast
+// enough that a passing test is not noticeably slowed by the granularity, and
+// slow enough not to hammer a serving endpoint.
+const DefaultPollInterval = 100 * time.Millisecond
+
+// Probe reports whether the awaited condition currently holds.
+//
+// A non-nil error is TERMINAL: the wait fails immediately with that error
+// attached. This is deliberate and it is the difference between a useful
+// failure and a useless one. A probe that polls an XRPC endpoint and gets a 401
+// will never succeed by being asked again — retrying it for 30 seconds converts
+// "the session was rejected" into "timed out waiting for the post to appear",
+// which is exactly the failure message an agent cannot debug. Return the error;
+// do not swallow it into `false`.
+//
+// Conversely, a condition that is merely not-yet-true — no rows, HTTP 404 for a
+// record still in flight — is `false, nil`.
+//
+// Probes should be cheap and side-effect free: they run every poll interval.
+//
+// To report what the system looked like when a wait times out, close over the
+// observation and render it with WithDiagnostics:
+//
+//	var last []Post
+//	testkit.WaitFor(t, 10*time.Second, func() (bool, error) {
+//	        var err error
+//	        last, err = client.Feed(ctx)
+//	        if err != nil {
+//	                return false, err // terminal: a broken endpoint is not a delay
+//	        }
+//	        return len(last) == 3, nil
+//	}, testkit.WithDescription("3 posts in the community feed"),
+//	   testkit.WithDiagnostics(func() string { return fmt.Sprintf("last feed: %v", last) }))
+type Probe func() (done bool, err error)
+
+type waitConfig struct {
+	interval    time.Duration
+	description string
+	diagnostics func() string
+}
+
+// WaitOption customises WaitFor and Holds.
+type WaitOption func(*waitConfig)
+
+// WithPollInterval overrides DefaultPollInterval.
+func WithPollInterval(d time.Duration) WaitOption {
+	return func(c *waitConfig) {
+		if d > 0 {
+			c.interval = d
+		}
+	}
+}
+
+// WithDescription names the condition being awaited, so the failure message
+// reads as a statement about the system rather than about the harness.
+func WithDescription(format string, args ...any) WaitOption {
+	return func(c *waitConfig) { c.description = fmt.Sprintf(format, args...) }
+}
+
+// WithDiagnostics registers a function rendered into the failure message, and
+// only into the failure message — it is never called on the success path.
+//
+// This is the hook the pipeline tier uses to attach consumer health (cursor
+// positions, dead-letter counts) to a timeout, so "the record never appeared"
+// arrives with the reason attached.
+func WithDiagnostics(fn func() string) WaitOption {
+	return func(c *waitConfig) { c.diagnostics = fn }
+}
+
+func newWaitConfig(opts []WaitOption) waitConfig {
+	cfg := waitConfig{interval: DefaultPollInterval}
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+	return cfg
+}
+
+func (c waitConfig) subject() string {
+	if c.description == "" {
+		return "condition"
+	}
+	return c.description
+}
+
+// render assembles a failure message with the diagnostics hook appended.
+func (c waitConfig) render(msg string) string {
+	var b strings.Builder
+	b.WriteString(msg)
+	if c.diagnostics != nil {
+		if diag := c.diagnostics(); diag != "" {
+			b.WriteString("\n  diagnostics: ")
+			b.WriteString(strings.ReplaceAll(diag, "\n", "\n    "))
+		}
+	}
+	return b.String()
+}
+
+// WaitFor polls probe until it reports true, failing the test if that has not
+// happened within timeout or if the probe returns an error.
+//
+// The last poll happens AT the deadline, not one interval before it. Sleeping a
+// full interval and then giving up would make the effective timeout
+// `timeout - interval`, so a condition that became true at 4.95s would be
+// reported as never satisfied by a WaitFor(t, 5*time.Second, …) — a flake whose
+// message actively misleads.
+func WaitFor(t TestingT, timeout time.Duration, probe Probe, opts ...WaitOption) {
+	t.Helper()
+	cfg := newWaitConfig(opts)
+
+	start := time.Now()
+	deadline := start.Add(timeout)
+	polls := 0
+
+	for {
+		polls++
+		done, err := probe()
+		if err != nil {
+			t.Fatalf("%s", cfg.render(fmt.Sprintf(
+				"waiting for %s: probe failed after %s (poll %d): %v",
+				cfg.subject(), time.Since(start).Round(time.Millisecond), polls, err)))
+			return
+		}
+		if done {
+			return
+		}
+
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			t.Fatalf("%s", cfg.render(fmt.Sprintf(
+				"timed out after %s (limit %s) waiting for %s: still false after %d polls",
+				time.Since(start).Round(time.Millisecond), timeout, cfg.subject(), polls)))
+			return
+		}
+		// Clamped, so the final sleep lands exactly on the deadline rather than
+		// past it.
+		time.Sleep(min(cfg.interval, remaining))
+	}
+}
+
+// Holds asserts that probe is true now and remains true for the whole window.
+//
+// Use it for negative and steady-state claims — a deleted record staying
+// deleted, a vote count staying at one after a duplicate event — where an
+// eventually-check would pass on its first poll and never notice the system
+// changing its mind a moment later.
+//
+// The whole window is observed: the last poll is at `start + window`, so
+// Holds(t, 250*time.Millisecond, …) really does watch for 250ms. Stopping an
+// interval early would silently shorten every stays-true assertion in the
+// suite by up to 100ms, which is exactly the span a resurrection-by-replay
+// takes to arrive.
+func Holds(t TestingT, window time.Duration, probe Probe, opts ...WaitOption) {
+	t.Helper()
+	cfg := newWaitConfig(opts)
+
+	start := time.Now()
+	end := start.Add(window)
+	polls := 0
+
+	for {
+		polls++
+		done, err := probe()
+		elapsed := time.Since(start)
+		if err != nil {
+			t.Fatalf("%s", cfg.render(fmt.Sprintf(
+				"holding %s: probe failed after %s of a %s window (poll %d): %v",
+				cfg.subject(), elapsed.Round(time.Millisecond), window, polls, err)))
+			return
+		}
+		if !done {
+			if polls == 1 {
+				t.Fatalf("%s", cfg.render(fmt.Sprintf(
+					"%s was already false at the start of the %s window",
+					cfg.subject(), window)))
+				return
+			}
+			t.Fatalf("%s", cfg.render(fmt.Sprintf(
+				"%s stopped holding after %s of a %s window (poll %d)",
+				cfg.subject(), elapsed.Round(time.Millisecond), window, polls)))
+			return
+		}
+
+		remaining := time.Until(end)
+		if remaining <= 0 {
+			return
+		}
+		// Clamped: a window shorter than one poll interval is still observed
+		// end to end, with a probe at 0 and a probe at the window's close.
+		time.Sleep(min(cfg.interval, remaining))
+	}
+}
diff --git a/tests/testkit/wait_test.go b/tests/testkit/wait_test.go
new file mode 100644
index 0000000..5eabe0e
--- /dev/null
+++ b/tests/testkit/wait_test.go
@@ -0,0 +1,219 @@
+package testkit
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWaitFor_ReturnsOnceTheProbeIsTrue(t *testing.T) {
+	var polls atomic.Int32
+
+	start := time.Now()
+	WaitFor(t, 5*time.Second, func() (bool, error) {
+		return polls.Add(1) >= 3, nil
+	}, WithPollInterval(20*time.Millisecond))
+	elapsed := time.Since(start)
+
+	assert.EqualValues(t, 3, polls.Load(), "should stop polling as soon as the probe is true")
+	// Two intervals of waiting between three polls: enough to prove it really
+	// polled rather than getting lucky on the first call.
+	assert.GreaterOrEqual(t, elapsed, 40*time.Millisecond)
+	assert.Less(t, elapsed, 2*time.Second, "should not have waited out the timeout")
+}
+
+func TestWaitFor_ProbeTrueOnFirstCallDoesNotWait(t *testing.T) {
+	start := time.Now()
+	WaitFor(t, 5*time.Second, func() (bool, error) { return true, nil })
+	assert.Less(t, time.Since(start), 50*time.Millisecond,
+		"an already-true condition must not cost a poll interval")
+}
+
+func TestWaitFor_TimesOutWithTheLastObservation(t *testing.T) {
+	f := &fakeT{}
+	observed := "http 404: post not indexed"
+
+	var polls atomic.Int32
+	start := time.Now()
+	runIsolated(func() {
+		WaitFor(f, 300*time.Millisecond, func() (bool, error) {
+			polls.Add(1)
+			return false, nil
+		},
+			WithPollInterval(50*time.Millisecond),
+			WithDescription("the post to appear in the feed"),
+			WithDiagnostics(func() string { return observed }))
+	})
+	elapsed := time.Since(start)
+
+	require.True(t, f.failed(), "a probe that never becomes true must fail the test")
+	msg := f.message()
+	assert.Contains(t, msg, "limit 300ms")
+	assert.Contains(t, msg, "the post to appear in the feed")
+	assert.Contains(t, msg, observed, "the last observation must reach the failure message")
+	assert.Contains(t, msg, fmt.Sprintf("%d polls", polls.Load()))
+	assert.GreaterOrEqual(t, elapsed, 300*time.Millisecond,
+		"the full timeout must elapse before giving up, not timeout-minus-one-interval")
+	assert.Less(t, elapsed, 2*time.Second)
+}
+
+func TestWaitFor_SucceedsWhenTheConditionArrivesJustBeforeTheDeadline(t *testing.T) {
+	// The off-by-one-interval bug, stated as a test. With a 300ms timeout and a
+	// 100ms interval, a condition that becomes true at ~280ms is inside the
+	// stated timeout; a loop that stops polling one interval early would report
+	// it as never satisfied, and the failure message would be a lie.
+	start := time.Now()
+	WaitFor(t, 300*time.Millisecond, func() (bool, error) {
+		return time.Since(start) >= 280*time.Millisecond, nil
+	}, WithPollInterval(100*time.Millisecond))
+
+	assert.GreaterOrEqual(t, time.Since(start), 280*time.Millisecond)
+}
+
+func TestWaitFor_ProbeErrorIsTerminal(t *testing.T) {
+	f := &fakeT{}
+	boom := errors.New("401 unauthorized")
+
+	var polls atomic.Int32
+	start := time.Now()
+	runIsolated(func() {
+		WaitFor(f, 10*time.Second, func() (bool, error) {
+			polls.Add(1)
+			return false, boom
+		}, WithDescription("the vote count to settle"))
+	})
+	elapsed := time.Since(start)
+
+	require.True(t, f.failed())
+	assert.EqualValues(t, 1, polls.Load(), "a terminal error must not be retried")
+	// The point of terminal errors: an unauthorised request fails in
+	// milliseconds with the status attached instead of burning the whole
+	// timeout and reporting an opaque "never became true".
+	assert.Less(t, elapsed, time.Second, "must fail fast rather than wait out the timeout")
+	msg := f.message()
+	assert.Contains(t, msg, "401 unauthorized")
+	assert.Contains(t, msg, "the vote count to settle")
+}
+
+func TestWaitFor_DiagnosticsAreNotEvaluatedOnSuccess(t *testing.T) {
+	var called atomic.Bool
+	WaitFor(t, time.Second, func() (bool, error) { return true, nil },
+		WithDiagnostics(func() string {
+			called.Store(true)
+			return "should never be rendered"
+		}))
+	assert.False(t, called.Load(),
+		"diagnostics may be expensive (an extra HTTP call); they belong on the failure path only")
+}
+
+func TestWaitFor_DefaultDescription(t *testing.T) {
+	f := &fakeT{}
+	runIsolated(func() {
+		WaitFor(f, 50*time.Millisecond, func() (bool, error) { return false, nil },
+			WithPollInterval(20*time.Millisecond))
+	})
+	require.True(t, f.failed())
+	assert.Contains(t, f.message(), "waiting for condition")
+}
+
+func TestHolds_PassesForAStableWindow(t *testing.T) {
+	var polls atomic.Int32
+
+	start := time.Now()
+	Holds(t, 250*time.Millisecond, func() (bool, error) {
+		polls.Add(1)
+		return true, nil
+	}, WithPollInterval(50*time.Millisecond))
+	elapsed := time.Since(start)
+
+	// The WHOLE window, not window-minus-one-interval. An earlier version
+	// stopped an interval early and this assertion tolerated it, which meant
+	// every stays-true assertion in the suite was quietly up to 100ms shorter
+	// than it claimed — about the span a resurrection-by-replay takes to land.
+	assert.GreaterOrEqual(t, elapsed, 250*time.Millisecond, "must observe the whole window")
+	assert.Greater(t, polls.Load(), int32(1), "must poll repeatedly, not just once")
+}
+
+func TestHolds_ObservesAWindowShorterThanThePollInterval(t *testing.T) {
+	var polls atomic.Int32
+
+	start := time.Now()
+	Holds(t, 40*time.Millisecond, func() (bool, error) {
+		polls.Add(1)
+		return true, nil
+	}, WithPollInterval(500*time.Millisecond))
+	elapsed := time.Since(start)
+
+	assert.GreaterOrEqual(t, elapsed, 40*time.Millisecond)
+	assert.Less(t, elapsed, 400*time.Millisecond,
+		"a short window must not be rounded up to a poll interval")
+	assert.EqualValues(t, 2, polls.Load(), "one probe at the start, one at the close")
+}
+
+func TestHolds_FailsWhenTheConditionFlipsMidWindow(t *testing.T) {
+	f := &fakeT{}
+	start := time.Now()
+
+	runIsolated(func() {
+		Holds(f, 2*time.Second, func() (bool, error) {
+			// The resurrection case: true for a moment, then not. An
+			// eventually-check would have passed on its first poll.
+			return time.Since(start) < 150*time.Millisecond, nil
+		},
+			WithPollInterval(25*time.Millisecond),
+			WithDescription("the deleted post to stay deleted"))
+	})
+
+	require.True(t, f.failed())
+	msg := f.message()
+	assert.Contains(t, msg, "the deleted post to stay deleted")
+	assert.Contains(t, msg, "stopped holding after")
+	assert.Contains(t, msg, "2s window")
+	assert.Less(t, time.Since(start), time.Second, "should fail at the flip, not at the end of the window")
+}
+
+func TestHolds_FailsWhenFalseAtTheStart(t *testing.T) {
+	f := &fakeT{}
+	runIsolated(func() {
+		Holds(f, time.Second, func() (bool, error) { return false, nil },
+			WithDescription("the row count to stay at 1"))
+	})
+	require.True(t, f.failed())
+	msg := f.message()
+	assert.Contains(t, msg, "already false at the start")
+	assert.NotContains(t, msg, "stopped holding",
+		"a condition that was never true is a different failure from one that stopped being true")
+}
+
+func TestHolds_ProbeErrorIsTerminal(t *testing.T) {
+	f := &fakeT{}
+	var polls atomic.Int32
+	runIsolated(func() {
+		Holds(f, 10*time.Second, func() (bool, error) {
+			polls.Add(1)
+			return true, errors.New("500 internal server error")
+		}, WithPollInterval(20*time.Millisecond))
+	})
+	require.True(t, f.failed())
+	assert.EqualValues(t, 1, polls.Load())
+	assert.Contains(t, f.message(), "500 internal server error")
+}
+
+func TestHolds_RendersDiagnostics(t *testing.T) {
+	f := &fakeT{}
+	runIsolated(func() {
+		Holds(f, 100*time.Millisecond, func() (bool, error) { return false, nil },
+			WithDiagnostics(func() string { return "dead letters: 3\ncursor: 172" }))
+	})
+	require.True(t, f.failed())
+	msg := f.message()
+	assert.Contains(t, msg, "diagnostics:")
+	assert.Contains(t, msg, "dead letters: 3")
+	assert.True(t, strings.Contains(msg, "cursor: 172"))
+}