diff --git a/.changeset/shared-sandbox-fencing.md b/.changeset/shared-sandbox-fencing.md
new file mode 100644
index 00000000..b1bbb8ba
--- /dev/null
+++ b/.changeset/shared-sandbox-fencing.md
@@ -0,0 +1,5 @@
+---
+"sql-fs-api": patch
+---
+
+Centralize Postgres sandbox fencing while preserving transaction context, lock order, and statement counts. Add regression coverage for stale writes and concurrent transactions, and fix verification commands that could fail under shell pipefail.
diff --git a/.claude/skills/verify/SKILL.md b/.claude/skills/verify/SKILL.md
index e0aa2dce..14b0f814 100644
--- a/.claude/skills/verify/SKILL.md
+++ b/.claude/skills/verify/SKILL.md
@@ -45,7 +45,7 @@ $S up --no-build # reuse the current dist/
 
 `up` refuses while a stack is recorded for this checkout. Reuse it (`$S doctor`) or `$S down` first. After a failed `up`, run `$S down` too, because the containers may be running.
 
-Server environment overrides: `SERVER_ENV="SQL_FS_WORKERS_ENABLED=false SQL_FS_WORKER_OUTPUT_BYTES=1048576" $S restart`. Overrides persist across later restarts until you pass `SERVER_ENV= $S restart`. Values are split on spaces, so use compact JSON with no spaces. `server.log` and `doctor` show only the override names; the values stay in the owner-only `state.env`.
+Server environment overrides: `SERVER_ENV="SQL_FS_WORKERS_ENABLED=false SQL_FS_WORKER_OUTPUT_BYTES=1048576" $S restart`. Overrides persist across later restarts until you pass `SERVER_ENV="" $S restart`. Values are split on spaces, so use compact JSON with no spaces. `server.log` and `doctor` show only the override names; the values stay in the owner-only `state.env`.
 
 ## Doctor
 
diff --git a/.claude/skills/verify/features/exec.md b/.claude/skills/verify/features/exec.md
index 5b7c1b18..2ecc8d75 100644
--- a/.claude/skills/verify/features/exec.md
+++ b/.claude/skills/verify/features/exec.md
@@ -26,7 +26,7 @@ Preconditions:
 - **SSE stream.** Run `$S api POST /v1/sandboxes/$ID/exec --json '{"script":"printf stream-proof"}'`. The output contains `event: stdout` with `"data":"stream-proof"` and exactly one `event: exit` with `"exitCode":0`.
 - **Batch.** Run `$S api POST /v1/sandboxes/$ID/exec-sync-batch --json '{"scripts":[{"id":"a","script":"echo one"},{"id":"b","script":"echo two"}]}' | jq -c '[.results[]|[.id,.exitCode,.stdout]]'`. The output is `[["a",0,"one\n"],["b",0,"two\n"]]`.
 - **Timeout prefix.** Run `EXPECT=408 $S api POST /v1/sandboxes/$ID/exec-sync --json '{"script":"echo p > /home/user/p.txt; sleep 2; echo s > /home/user/s.txt","timeoutMs":100}'`. The `code` is `EXEC_TIMEOUT`. Then `$S api GET /v1/sandboxes/$ID/files/home/user/p.txt` prints `p` and `EXPECT=404 $S api GET /v1/sandboxes/$ID/files/home/user/s.txt` succeeds.
-- **Worker modes.** Repeat the sync run after `SERVER_ENV="SQL_FS_WORKERS_ENABLED=false" $S restart --no-build`, then restore with `SERVER_ENV= $S restart --no-build`. Both modes must give identical results.
+- **Worker modes.** Repeat the sync run after `SERVER_ENV="SQL_FS_WORKERS_ENABLED=false" $S restart --no-build`, then restore with `SERVER_ENV="" $S restart --no-build`. Both modes must give identical results.
 
 ## Gotchas
 
diff --git a/.claude/skills/verify/scripts/journeys.sh b/.claude/skills/verify/scripts/journeys.sh
index ecd2045d..d6d8d32c 100755
--- a/.claude/skills/verify/scripts/journeys.sh
+++ b/.claude/skills/verify/scripts/journeys.sh
@@ -60,10 +60,10 @@ journey_exec() {
 	eq "$(EXPECT=408 api POST "/v1/sandboxes/$id/exec-sync" --json '{"script":"echo p > /home/user/p.txt; sleep 2; echo s > /home/user/s.txt","timeoutMs":100}' | jq -r .code)" EXEC_TIMEOUT "timeout -> 408"
 	eq "$(api GET "/v1/sandboxes/$id/files/home/user/p.txt")" p "timeout kept prefix"
 	eq "$(absent "/v1/sandboxes/$id/files/home/user/s.txt")" absent "timeout dropped suffix"
-	trap 'SERVER_ENV= $S restart --no-build >/dev/null' EXIT
+	trap 'SERVER_ENV="" $S restart --no-build >/dev/null' EXIT
 	SERVER_ENV="SQL_FS_WORKERS_ENABLED=false" $S restart --no-build >/dev/null
 	eq "$(api POST "/v1/sandboxes/$id/exec-sync" --json '{"script":"cat /home/user/a.txt"}' | jq -c '{exitCode,stdout}')" '{"exitCode":0,"stdout":"hi\n"}' "workers disabled: same result"
-	SERVER_ENV= $S restart --no-build >/dev/null
+	SERVER_ENV="" $S restart --no-build >/dev/null
 	trap - EXIT
 	drop "$id"
 }
diff --git a/.claude/skills/verify/scripts/stack.sh b/.claude/skills/verify/scripts/stack.sh
index c943dc4e..15dca459 100755
--- a/.claude/skills/verify/scripts/stack.sh
+++ b/.claude/skills/verify/scripts/stack.sh
@@ -97,7 +97,8 @@ cmd_up() {
 	command -v docker >/dev/null && docker info >/dev/null 2>&1 || die "Docker daemon is not running"
 	[[ -d node_modules ]] || die "run pnpm setup:dev first"
 	deps_current || die "node_modules does not match pnpm-lock.yaml; run: pnpm install --frozen-lockfile"
-	local id="stack-$(date +%s)-$$"
+	local id
+	id="stack-$(date +%s)-$$"
 	DIR="$runs/$id"
 	PROJECT="sqlfs-$id"
 	if docker compose version >/dev/null 2>&1; then COMPOSE="docker compose"; else COMPOSE="docker-compose"; fi
@@ -149,8 +150,10 @@ cmd_doctor() {
 	[[ "$code" == 200 ]] && ok "minted token accepted (GET /v1/sandboxes 200)" || fail "GET /v1/sandboxes returned $code"
 	deps_current && ok "node_modules matches pnpm-lock.yaml" ||
 		fail "node_modules does not match pnpm-lock.yaml -> run: pnpm install --frozen-lockfile && $0 restart"
-	local newer
-	newer=$(find src \( -name '*.ts' -o -name '*.sql' \) -newer dist/api/server.js -not -path '*/tests/*' 2>/dev/null | head -3)
+	local sources newer
+	[[ -f dist/api/server.js ]] || { fail "dist/api/server.js is missing -> run: $0 restart"; return 1; }
+	sources=$(find src \( -name '*.ts' -o -name '*.sql' \) -newer dist/api/server.js -not -path '*/tests/*' 2>/dev/null)
+	newer=$(head -3 <<<"$sources")
 	if [[ -n "$newer" ]]; then
 		fail "dist is stale; source changed after build: $(echo "$newer" | tr '\n' ' ') -> run: $0 restart"
 	elif [[ dist/api/server.js -nt "$DIR/state.env" ]]; then
@@ -210,9 +213,10 @@ suite_names() {
 cmd_suite() {
 	load_state
 	(($# >= 1)) || die "usage: $0 suite NAME [args]   (NAME: integration or a suite entry point in scripts/verify/)"
-	local name=$1
+	local name=$1 names
 	shift
-	suite_names | grep -qx "$name" || die "$name is not a suite entry point (have: $(suite_names | tr '\n' ' '))"
+	names=$(suite_names)
+	grep -qx "$name" <<<"$names" || die "$name is not a suite entry point (have: $(tr '\n' ' ' <<<"$names"))"
 	local -a suite_env=("VERIFY_ENV_FILE=none" "VERIFY_EVIDENCE_DIR=$DIR/suites" "VERIFY_PYTHON=$repo/clients/python/.venv/bin/python")
 	mkdir -p "$DIR/suites"
 	if [[ "$name" == integration ]]; then
diff --git a/docs/DEVELOPER.md b/docs/DEVELOPER.md
index 5db15a4b..3ebcc3ba 100644
--- a/docs/DEVELOPER.md
+++ b/docs/DEVELOPER.md
@@ -146,7 +146,7 @@ The session runtime opens a script scope only if the filesystem implements `IScr
 
 ## Ownership checks
 
-`pnpm check:architecture` enforces the worker module's named adapter imports and the script-scope completion owner. The allowed imports live in `scripts/check-architecture-boundaries.mjs`, with fixture tests beside the checker. Add a new adapter only when it has a real external responsibility; keep dispatch history, reservation state, and outcome classification inside worker ownership. The shared lease-loss context lives in `src/execution/owner-cancellation.ts` so filesystem fencing does not import worker internals.
+`pnpm check:architecture` enforces the worker module's named adapter imports, the script-scope completion owner, and the sandbox fence policy owner in `src/sql-fs/dialects/postgres/sandbox.ts`. The allowed imports live in `scripts/check-architecture-boundaries.mjs`, with fixture tests beside the checker. Add a new adapter only when it has a real external responsibility; keep dispatch history, reservation state, and outcome classification inside worker ownership. The shared lease-loss context lives in `src/execution/owner-cancellation.ts` so filesystem fencing does not import worker internals.
 
 ## Worker owners run scripts off the main thread
 
diff --git a/scripts/check-architecture-boundaries.mjs b/scripts/check-architecture-boundaries.mjs
index 55d83a80..8c02af45 100644
--- a/scripts/check-architecture-boundaries.mjs
+++ b/scripts/check-architecture-boundaries.mjs
@@ -5,7 +5,7 @@ import { fileURLToPath } from "node:url";
 import * as ts from "typescript/unstable/sync";
 import * as ast from "typescript/unstable/ast";
 
-const { API, SymbolFlags } = ts;
+const { API, SignatureKind, SymbolFlags, TypeFlags } = ts;
 const { SyntaxKind } = ast;
 
 const workerImportAllowlist = new Map([
@@ -44,6 +44,40 @@ for (const importer of [
 
 const scopeOperations = new Set(["beginScope", "endScope", "abortScope"]);
 
+const fencePolicyConsumers = new Set([
+	"src/sql-fs/dialects/postgres/namespace.ts",
+	"src/sql-fs/dialects/postgres/ingest.ts",
+]);
+const fencePolicyStatements = [/\bapp\.sandbox_epoch\b/i, /\bversion\s*=\s*(?:\w+\.)?version\s*\+/i];
+
+function forbiddenSqlApiAccess(node, checker) {
+	let receiver;
+	let name;
+	if (ast.isPropertyAccessExpression(node)) {
+		receiver = node.expression;
+		name = node.name.text;
+	} else if (ast.isElementAccessExpression(node) && ast.isStringLiteral(node.argumentExpression)) {
+		receiver = node.expression;
+		name = node.argumentExpression.text;
+	} else if (ast.isBindingElement(node) && ast.isObjectBindingPattern(node.parent)) {
+		receiver = node.parent;
+		const property = node.propertyName ?? node.name;
+		if (ast.isIdentifier(property) || ast.isStringLiteral(property)) name = property.text;
+	}
+	if (name !== "unsafe" && name !== "file") return false;
+	const type = checker.getTypeAtLocation(receiver);
+	return (
+		type === undefined ||
+		(type.flags & (TypeFlags.Any | TypeFlags.Unknown)) !== 0 ||
+		checker.getSignaturesOfType(type, SignatureKind.Call).length > 0
+	);
+}
+
+function templateLiteralText(template) {
+	if (ast.isNoSubstitutionTemplateLiteral(template)) return template.text;
+	return [template.head.text, ...template.templateSpans.map((span) => span.literal.text)].join(" ${} ");
+}
+
 function relative(root, filename) {
 	return path.relative(root, filename).split(path.sep).join("/");
 }
@@ -414,6 +448,34 @@ export function checkArchitectureBoundaries(root) {
 					}
 				}
 
+				if (fencePolicyConsumers.has(sourcePath)) {
+					if (forbiddenSqlApiAccess(node, checker)) {
+						diagnostics.push(
+							diagnostic(
+								absoluteRoot,
+								sourceFile,
+								node,
+								"FENCE_POLICY_RAW_SQL",
+								"tx.unsafe and tx.file bypass fence-policy inspection; use tagged SQL and safe value builders",
+							),
+						);
+					}
+					if (
+						ast.isTaggedTemplateExpression(node) &&
+						fencePolicyStatements.some((pattern) => pattern.test(templateLiteralText(node.template)))
+					) {
+						diagnostics.push(
+							diagnostic(
+								absoluteRoot,
+								sourceFile,
+								node,
+								"FENCE_POLICY_OWNER",
+								"epoch acceptance, epoch publication, and sandboxes.version advancement belong to the fence policy in sandbox.ts; use fenceAndAdvance or compositeFenceAndAdvance",
+							),
+						);
+					}
+				}
+
 				if (ast.isCallExpression(node)) {
 					const expression = unwrapParentheses(node.expression);
 					if (
diff --git a/scripts/check-architecture-boundaries.test.mjs b/scripts/check-architecture-boundaries.test.mjs
index 723ee6c3..63f7b528 100644
--- a/scripts/check-architecture-boundaries.test.mjs
+++ b/scripts/check-architecture-boundaries.test.mjs
@@ -188,3 +188,112 @@ it("permits literal non-worker imports and rejects indirect scope method calls",
 		["SCRIPT_SCOPE_COMPLETION_OWNER", "SCRIPT_SCOPE_COMPLETION_OWNER", "SCRIPT_SCOPE_COMPLETION_OWNER"],
 	);
 });
+
+const sqlTag = "declare const tx: (strings: TemplateStringsArray, ...values: unknown[]) => unknown;\n";
+
+it("rejects fence policy copied into namespace and ingest SQL", () => {
+	const root = project({
+		"src/sql-fs/dialects/postgres/namespace.ts": `${sqlTag}export function copy(id: string, epoch: string | null) {
+	tx\`WITH ctx AS (SELECT s.version FROM sandboxes s WHERE s.id = \${id}
+		AND s.version = COALESCE(\${epoch}::bigint, NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint, s.version))
+		SELECT 1 FROM ctx\`;
+	tx\`UPDATE sandboxes s
+		SET version = s.version + 1 WHERE s.id = \${id}\`;
+}
+`,
+		"src/sql-fs/dialects/postgres/ingest.ts": `${sqlTag}export function publish(version: string) {
+	tx\`SELECT set_config('app.sandbox_epoch', \${version}, true)\`;
+}
+`,
+	});
+	const diagnostics = checkArchitectureBoundaries(root).filter(({ code }) => code === "FENCE_POLICY_OWNER");
+	assert.deepEqual(
+		diagnostics.map(({ file, line }) => [file, line]),
+		[
+			["src/sql-fs/dialects/postgres/ingest.ts", 3],
+			["src/sql-fs/dialects/postgres/namespace.ts", 3],
+			["src/sql-fs/dialects/postgres/namespace.ts", 6],
+		],
+	);
+});
+
+it("permits epoch consumers, version reads, and the fence policy owner", () => {
+	const root = project({
+		"src/sql-fs/dialects/postgres/sandbox.ts": `${sqlTag}export function fence(id: string) {
+	return tx\`ctx AS (SELECT s.version FROM sandboxes s WHERE s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint),
+		advanced AS (UPDATE sandboxes s SET version = s.version + 1 FROM ctx WHERE s.id = \${id} RETURNING s.version)\`;
+}
+`,
+		"src/sql-fs/dialects/postgres/namespace.ts": `${sqlTag}// app.sandbox_epoch is published by the fence policy
+const prose = "UPDATE sandboxes SET version = version + 1";
+export function consume(id: string, fence: unknown) {
+	tx\`WITH \${fence}, inserted AS (INSERT INTO inodes (sandbox_id) SELECT \${id} FROM epoch RETURNING id)
+		SELECT (SELECT count(*) FROM epoch)::int AS fenced, (SELECT id FROM inserted) AS id
+		FROM sandboxes WHERE id = \${id}\`;
+	tx\`SELECT version FROM sandboxes WHERE id = \${id}\`;
+	tx\`UPDATE sandboxes SET name = \${prose} WHERE id = \${id}\`;
+	tx\`UPDATE sandboxes SET name = \${prose} WHERE version = \${id}\`;
+}
+`,
+	});
+	assert.deepEqual(checkArchitectureBoundaries(root), []);
+});
+
+const sqlCapabilities = `declare const tx: {
+	(strings: TemplateStringsArray, ...values: unknown[]): unknown;
+	(values: unknown[] | Record<string, unknown>): unknown;
+	unsafe(query: string): unknown;
+	file(path: string): unknown;
+};
+`;
+
+it("rejects direct, extracted, destructured, and aliased raw SQL capabilities in both fence consumers", () => {
+	const accesses = [
+		'tx.unsafe("UPDATE sandboxes SET version = version + 1")',
+		'tx["unsafe"]("SELECT 1")',
+		'const run = tx.unsafe; run("SELECT 1")',
+		'const run = tx["unsafe"]; run("SELECT 1")',
+		'const { unsafe: run } = tx; run("SELECT 1")',
+		'const { "unsafe": run } = tx; run("SELECT 1")',
+		'const db = tx; db.unsafe("SELECT 1")',
+		'tx.file("query.sql")',
+		'tx["file"]("query.sql")',
+		'const run = tx.file; run("query.sql")',
+		'const run = tx["file"]; run("query.sql")',
+		'const { file: run } = tx; run("query.sql")',
+		'const db = tx; db.file("query.sql")',
+	];
+	const body = accesses.map((access) => `{ ${access}; }`).join("\n");
+	const files = ["namespace", "ingest"].map((name) => `src/sql-fs/dialects/postgres/${name}.ts`);
+	const root = project(Object.fromEntries(files.map((file) => [file, sqlCapabilities + body])));
+	const diagnostics = checkArchitectureBoundaries(root);
+	assert.equal(diagnostics.length, files.length * accesses.length);
+	for (const file of files) {
+		assert.deepEqual(
+			diagnostics.filter((item) => item.file === file).map(({ code, line }) => ({ code, line })),
+			accesses.map((_, index) => ({ code: "FENCE_POLICY_RAW_SQL", line: index + 7 })),
+		);
+	}
+});
+
+it("permits safe SQL builders, unrelated file data, and raw SQL in the sandbox lifecycle owner", () => {
+	const root = project({
+		"src/sql-fs/dialects/postgres/namespace.ts": `${sqlCapabilities}
+const db = tx;
+const prose = "UPDATE sandboxes SET version = version + 1";
+db([1, 2]);
+db({ name: prose });
+db\`SELECT \${prose}\`;
+const entry = { file: "notes.txt", unsafe: false };
+void entry.file;
+void entry["file"];
+const { file, unsafe } = entry;
+`,
+		"src/sql-fs/dialects/postgres/sandbox.ts": `${sqlCapabilities}
+tx.unsafe("UPDATE sandboxes SET version = version + 1");
+const { file: run } = tx;
+run("lifecycle.sql");
+`,
+	});
+	assert.deepEqual(checkArchitectureBoundaries(root), []);
+});
diff --git a/src/sql-fs/dialects/postgres/README.md b/src/sql-fs/dialects/postgres/README.md
index 749de14d..888a08cf 100644
--- a/src/sql-fs/dialects/postgres/README.md
+++ b/src/sql-fs/dialects/postgres/README.md
@@ -13,6 +13,21 @@ Group contracts in [types.ts](types.ts) derive from SqlDialect. Postgres-specifi
 
 The root blob commit and batch commit remain independent single statements outside the caller's namespace transaction. Ingest invokes its blob commit after directory and overwrite checks, before inserting file inodes. Cache backfill remains asynchronous.
 
-Writer-lock statements precede fenced CTEs so their snapshots follow any lock wait. Fenced CTE mutations reference and gate their writes on the epoch CTE; lower-level mutations run directly in the caller transaction. Both mvComposite statements and its two version advances remain together. Sandbox bootstrap inserts remain part of sandbox lifecycle rather than ordinary mkdir operations.
+Writer-lock statements precede fenced CTEs so READ COMMITTED statement snapshots follow any lock wait. REPEATABLE READ retains an established transaction snapshot. Fenced CTE mutations reference and gate their writes on the epoch CTE; lower-level mutations run directly in the caller transaction. Both mvComposite statements and its two version advances remain together. Sandbox bootstrap inserts remain part of sandbox lifecycle rather than ordinary mkdir operations.
+
+## Fencing
+
+| To change | Edit | Verify with |
+| --- | --- | --- |
+| How a namespace or ingest write accepts an epoch, advances the version, and publishes the transaction-local epoch | The private `fenceCtes` policy in [sandbox.ts](sandbox.ts). Lifecycle methods in the same file, such as retirement, context locking, and runtime updates, advance or publish directly. | [fence-policy](../../tests/integration/fence-policy.integration.test.ts), [fencing](../../tests/integration/fencing.integration.test.ts), and [epoch coverage](../../tests/integration/epoch-fence-coverage.integration.test.ts) integration suites |
+| A namespace operation | Its complete statement in [namespace.ts](namespace.ts) | The same suites, plus [move overwrite](../../tests/integration/mvcomposite-overwrite.integration.test.ts) for moves |
+
+`fenceAndAdvance` and `compositeFenceAndAdvance` return lazy SQL fragments that callers place in their own `WITH` clause. Keep them synchronous. An async builder would await the fragment, and awaiting a postgres.js query executes it. `fenceAndAdvance` projects `s.version` for callers that already set context, which covers lower-level namespace mutations and ingest. `compositeFenceAndAdvance` also sets `app.sandbox_id` and takes the reentrant advisory lock inline. Composite callers still run `takeWriterLock` first.
+
+Run the integration suites on the disposable verify stack, whose `sqlfs_app` role is neither superuser nor BYPASSRLS. A rejected write that rolls back looks the same as one that never ran, so the fence-policy and epoch coverage suites include cases that catch `ESTALE` and commit before checking state.
+
+The fence-policy suite pins the advisory-lock wait verdicts under READ COMMITTED and an established REPEATABLE READ snapshot; its `it.each` table lists them.
+
+`pnpm check:architecture` rejects `app.sandbox_epoch` references and `sandboxes.version` increments in tagged SQL in namespace.ts and ingest.ts. It also rejects `unsafe` and `file` raw SQL access in those two consumers. The check does not parse arbitrary SQL strings or track computed names, reflection, or capabilities hidden behind noncallable types. Sandbox lifecycle operations remain owned by sandbox.ts.
 
 GC uses the transaction and isolation selected by its caller. It does not create a nested transaction. Namespace and sandbox operations use the caller transaction. Ingest keeps namespace writes there, but commits its blob batch independently through the root pool.
diff --git a/src/sql-fs/dialects/postgres/namespace.ts b/src/sql-fs/dialects/postgres/namespace.ts
index f85b7e9a..f47b4f65 100644
--- a/src/sql-fs/dialects/postgres/namespace.ts
+++ b/src/sql-fs/dialects/postgres/namespace.ts
@@ -1,7 +1,7 @@
 import { createEnoent, createEstale, translateSqlError } from "../../errors.js";
 import type { CreateInodeOpts, DirentRow, InodeKind, InodeRow, PathCacheEntry, UpdateInodeOpts } from "../../types.js";
 import type { PostgresBlobs } from "./blobs.js";
-import { epochParam, fenceAndAdvance, takeWriterLock } from "./sandbox.js";
+import { compositeFenceAndAdvance, fenceAndAdvance, takeWriterLock } from "./sandbox.js";
 import type { NamespaceOperations, PgTx } from "./types.js";
 
 export class PostgresNamespace implements NamespaceOperations {
@@ -16,35 +16,7 @@ export class PostgresNamespace implements NamespaceOperations {
 	): Promise<bigint> {
 		await takeWriterLock(tx, sandboxId);
 		const rows = await tx<{ fenced: number; id: string | null }[]>`
-			WITH ctx AS (
-				SELECT set_config('app.sandbox_id', ${sandboxId}, true),
-				       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
-				       s.version AS epoch
-				FROM sandboxes s
-				WHERE s.id = ${sandboxId}
-				  AND (
-					s.version = COALESCE(
-						${epochParam(expectedEpoch)}::bigint,
-						NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint,
-						s.version
-					)
-					OR (
-						${epochParam(expectedEpoch)}::bigint IS NOT NULL
-						AND s.version > ${epochParam(expectedEpoch)}::bigint
-						AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint
-					)
-				  )
-			),
-			advanced AS (
-				UPDATE sandboxes s
-				SET version = s.version + 1
-				FROM ctx
-				WHERE s.id = ${sandboxId}
-				RETURNING s.version
-			),
-			epoch AS (
-				SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced
-			),
+			WITH ${compositeFenceAndAdvance(tx, sandboxId, expectedEpoch)},
 			new_inode AS (
 				INSERT INTO inodes (sandbox_id, kind, mode, size)
 				SELECT ${sandboxId}, 2, ${mode}, 0 FROM epoch
@@ -74,35 +46,7 @@ export class PostgresNamespace implements NamespaceOperations {
 	): Promise<bigint> {
 		await takeWriterLock(tx, sandboxId);
 		const rows = await tx<{ fenced: number; removed_inode_id: string | null }[]>`
-			WITH ctx AS (
-				SELECT set_config('app.sandbox_id', ${sandboxId}, true),
-				       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
-				       s.version AS epoch
-				FROM sandboxes s
-				WHERE s.id = ${sandboxId}
-				  AND (
-					s.version = COALESCE(
-						${epochParam(expectedEpoch)}::bigint,
-						NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint,
-						s.version
-					)
-					OR (
-						${epochParam(expectedEpoch)}::bigint IS NOT NULL
-						AND s.version > ${epochParam(expectedEpoch)}::bigint
-						AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint
-					)
-				  )
-			),
-			advanced AS (
-				UPDATE sandboxes s
-				SET version = s.version + 1
-				FROM ctx
-				WHERE s.id = ${sandboxId}
-				RETURNING s.version
-			),
-			epoch AS (
-				SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced
-			),
+			WITH ${compositeFenceAndAdvance(tx, sandboxId, expectedEpoch)},
 			removed_dirent AS (
 				DELETE FROM dirents
 				WHERE parent_inode_id = ${String(parentId)} AND name = ${name}
@@ -151,35 +95,7 @@ export class PostgresNamespace implements NamespaceOperations {
 		// script-tx must not hold the hot-blob `ON CONFLICT DO UPDATE` tuple lock.
 		await takeWriterLock(tx, sandboxId);
 		const rows = await tx<{ fenced: number; new_inode_id: string | null }[]>`
-			WITH ctx AS (
-				SELECT set_config('app.sandbox_id', ${sandboxId}, true),
-				       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
-				       s.version AS epoch
-				FROM sandboxes s
-				WHERE s.id = ${sandboxId}
-				  AND (
-					s.version = COALESCE(
-						${epochParam(expectedEpoch)}::bigint,
-						NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint,
-						s.version
-					)
-					OR (
-						${epochParam(expectedEpoch)}::bigint IS NOT NULL
-						AND s.version > ${epochParam(expectedEpoch)}::bigint
-						AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint
-					)
-				  )
-			),
-			advanced AS (
-				UPDATE sandboxes s
-				SET version = s.version + 1
-				FROM ctx
-				WHERE s.id = ${sandboxId}
-				RETURNING s.version
-			),
-			epoch AS (
-				SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced
-			),
+			WITH ${compositeFenceAndAdvance(tx, sandboxId, expectedEpoch)},
 			new_inode AS (
 				INSERT INTO inodes (sandbox_id, kind, mode, size, content_sha256)
 				SELECT ${sandboxId}, 1, ${mode}, ${size}, ${sha256} FROM epoch
@@ -243,35 +159,7 @@ export class PostgresNamespace implements NamespaceOperations {
 		// refresh in SqlFs absorbs the double bump.
 		await takeWriterLock(tx, sandboxId);
 		const destinationRows = await tx<{ fenced: number }[]>`
-			WITH ctx AS (
-				SELECT set_config('app.sandbox_id', ${sandboxId}, true),
-				       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
-				       s.version AS epoch
-				FROM sandboxes s
-				WHERE s.id = ${sandboxId}
-				  AND (
-					s.version = COALESCE(
-						${epochParam(expectedEpoch)}::bigint,
-						NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint,
-						s.version
-					)
-					OR (
-						${epochParam(expectedEpoch)}::bigint IS NOT NULL
-						AND s.version > ${epochParam(expectedEpoch)}::bigint
-						AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint
-					)
-				  )
-			),
-			advanced AS (
-				UPDATE sandboxes s
-				SET version = s.version + 1
-				FROM ctx
-				WHERE s.id = ${sandboxId}
-				RETURNING s.version
-			),
-			epoch AS (
-				SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced
-			),
+			WITH ${compositeFenceAndAdvance(tx, sandboxId, expectedEpoch)},
 			old_dest AS (
 				DELETE FROM dirents
 				WHERE parent_inode_id = ${String(newParentId)} AND name = ${newName}
@@ -301,35 +189,7 @@ export class PostgresNamespace implements NamespaceOperations {
 		// Re-take for snapshot ordering; re-entrant within this tx.
 		await takeWriterLock(tx, sandboxId);
 		const rows = await tx<{ fenced: number; inode_id: string | null }[]>`
-			WITH ctx AS (
-				SELECT set_config('app.sandbox_id', ${sandboxId}, true),
-				       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
-				       s.version AS epoch
-				FROM sandboxes s
-				WHERE s.id = ${sandboxId}
-				  AND (
-					s.version = COALESCE(
-						${epochParam(expectedEpoch)}::bigint,
-						NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint,
-						s.version
-					)
-					OR (
-						${epochParam(expectedEpoch)}::bigint IS NOT NULL
-						AND s.version > ${epochParam(expectedEpoch)}::bigint
-						AND s.version = NULLIF(current_setting('app.sandbox_epoch', true), '')::bigint
-					)
-				  )
-			),
-			advanced AS (
-				UPDATE sandboxes s
-				SET version = s.version + 1
-				FROM ctx
-				WHERE s.id = ${sandboxId}
-				RETURNING s.version
-			),
-			epoch AS (
-				SELECT set_config('app.sandbox_epoch', version::text, true) FROM advanced
-			),
+			WITH ${compositeFenceAndAdvance(tx, sandboxId, expectedEpoch)},
 			moved AS (
 				UPDATE dirents
 				SET parent_inode_id = ${String(newParentId)}, name = ${newName}
diff --git a/src/sql-fs/dialects/postgres/sandbox.ts b/src/sql-fs/dialects/postgres/sandbox.ts
index 621bcaae..510ed04a 100644
--- a/src/sql-fs/dialects/postgres/sandbox.ts
+++ b/src/sql-fs/dialects/postgres/sandbox.ts
@@ -3,9 +3,7 @@ import { createEnoent, translateSqlError } from "../../errors.js";
 import type { SandboxGeneration, SandboxListEntry, SandboxMeta } from "../../types.js";
 import type { PgTx, SandboxOperations } from "./types.js";
 
-export function epochParam(expectedEpoch?: bigint): string | null {
-	return expectedEpoch === undefined ? null : String(expectedEpoch);
-}
+type FenceContext = "existing" | "composite";
 
 export async function takeWriterLock(tx: PgTx, sandboxId: string): Promise<void> {
 	await tx`SELECT pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0))`;
@@ -16,10 +14,33 @@ export function fenceAndAdvance(
 	sandboxId: string,
 	expectedEpoch?: bigint,
 ): postgres.PendingQuery<postgres.Row[]> {
-	const epoch = epochParam(expectedEpoch);
+	return fenceCtes(tx, sandboxId, expectedEpoch, "existing");
+}
+
+export function compositeFenceAndAdvance(
+	tx: PgTx,
+	sandboxId: string,
+	expectedEpoch?: bigint,
+): postgres.PendingQuery<postgres.Row[]> {
+	return fenceCtes(tx, sandboxId, expectedEpoch, "composite");
+}
+
+function fenceCtes(
+	tx: PgTx,
+	sandboxId: string,
+	expectedEpoch: bigint | undefined,
+	context: FenceContext,
+): postgres.PendingQuery<postgres.Row[]> {
+	const epoch = expectedEpoch === undefined ? null : String(expectedEpoch);
+	const projection =
+		context === "existing"
+			? tx`s.version`
+			: tx`set_config('app.sandbox_id', ${sandboxId}, true),
+			       pg_advisory_xact_lock(hashtextextended(${sandboxId}, 0)),
+			       s.version AS epoch`;
 	return tx`
 			ctx AS (
-				SELECT s.version
+				SELECT ${projection}
 				FROM sandboxes s
 				WHERE s.id = ${sandboxId}
 				  AND (
diff --git a/src/sql-fs/tests/integration/fence-policy.integration.test.ts b/src/sql-fs/tests/integration/fence-policy.integration.test.ts
new file mode 100644
index 00000000..e7abf4fb
--- /dev/null
+++ b/src/sql-fs/tests/integration/fence-policy.integration.test.ts
@@ -0,0 +1,462 @@
+import { randomBytes, randomUUID } from "node:crypto";
+import { setTimeout as pause } from "node:timers/promises";
+import postgres from "postgres";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { bulkIngest } from "../../dialects/postgres/ingest.js";
+import { PostgresNamespace } from "../../dialects/postgres/namespace.js";
+import {
+	compositeFenceAndAdvance,
+	fenceAndAdvance,
+	PostgresSandbox,
+	takeWriterLock,
+} from "../../dialects/postgres/sandbox.js";
+import type { PgTx } from "../../dialects/postgres/types.js";
+import { requireMigratedSchema } from "./helpers/schema-preconditions.js";
+
+const SKIP = !process.env.DATABASE_URL;
+
+type Seeded = { id: string; rootInodeId: bigint };
+type Entry = { parent: string; name: string; inode: string };
+type Inode = { id: string; nlink: number; mode: number; sha: string | null };
+type Snapshot = { version: string; entries: Entry[] | null; inodes: Inode[] | null };
+type Isolation = "read committed" | "repeatable read";
+type Contenders = { holderPid: number; waiterPid: number };
+type LockObservation = { blocked: boolean; matching_lock: boolean };
+
+describe.skipIf(SKIP)("sandbox fence policy", () => {
+	let recording: string[] | undefined;
+	const sql = postgres(process.env.DATABASE_URL ?? "postgres://unused", {
+		prepare: false,
+		max: 4,
+		debug(_connection, query) {
+			recording?.push(query.replace(/\s+/g, " ").trim());
+		},
+	});
+	const sandbox = new PostgresSandbox();
+	let backfills = 0;
+	const namespace = new PostgresNamespace({
+		backfill() {
+			backfills++;
+		},
+	});
+	const ids: string[] = [];
+	const sha = new Uint8Array(randomBytes(32));
+	const data = new Uint8Array([65]);
+
+	beforeAll(async () => {
+		await requireMigratedSchema(process.env.DATABASE_URL!);
+		const [role] = await sql<{ rolsuper: boolean; rolbypassrls: boolean }[]>`
+			SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user
+		`;
+		expect(role, "run against an ordinary application role such as sqlfs_app; RLS proofs need it").toEqual({
+			rolsuper: false,
+			rolbypassrls: false,
+		});
+		await sql`INSERT INTO blobs (sha256, data, size) VALUES (${sha}, ${data}, 1) ON CONFLICT (sha256) DO NOTHING`;
+	});
+
+	afterAll(async () => {
+		recording = undefined;
+		for (const id of ids) {
+			await sql.begin(async (tx) => {
+				await sandbox.deleteSandbox(tx, id);
+				await tx`DELETE FROM sandbox_epochs WHERE sandbox_id = ${id}`;
+			});
+		}
+		await sql`
+			DELETE FROM blobs
+			WHERE sha256 = ${sha} AND NOT EXISTS (SELECT 1 FROM inodes WHERE content_sha256 = ${sha})
+		`;
+		await sql.end();
+	});
+
+	async function create(label: string): Promise<Seeded> {
+		const id = `fence-policy-${label}-${randomUUID()}`;
+		ids.push(id);
+		const created = await sql.begin((tx) => sandbox.createSandbox(tx, id));
+		return { id, rootInodeId: created.rootInodeId };
+	}
+
+	function snapshot(s: Seeded): Promise<Snapshot> {
+		return sql.begin(async (tx) => {
+			await sandbox.setSandboxContext(tx, s.id);
+			const [row] = await tx<Snapshot[]>`
+				SELECT version::text,
+					(SELECT jsonb_agg(jsonb_build_object('parent', parent_inode_id::text, 'name', name, 'inode', inode_id::text)
+						ORDER BY parent_inode_id, name) FROM dirents WHERE sandbox_id = ${s.id}) AS entries,
+					(SELECT jsonb_agg(jsonb_build_object(
+						'id', id::text, 'nlink', nlink, 'mode', mode, 'sha', encode(content_sha256, 'hex')
+					) ORDER BY id) FROM inodes WHERE sandbox_id = ${s.id}) AS inodes
+				FROM sandboxes WHERE id = ${s.id}
+			`;
+			return row!;
+		});
+	}
+
+	async function executed<T>(
+		run: () => Promise<T>,
+	): Promise<{ statements: string[]; result?: T; error?: unknown; code?: string }> {
+		const statements: string[] = [];
+		recording = statements;
+		try {
+			return { statements, result: await run() };
+		} catch (error) {
+			return { statements, error, code: (error as { code?: string }).code };
+		} finally {
+			recording = undefined;
+		}
+	}
+
+	async function linkAlias(tx: PgTx, s: Seeded, target: string, alias: string): Promise<void> {
+		await sandbox.setSandboxContextWithLock(tx, s.id);
+		const [row] = await tx<{ inode_id: string }[]>`
+			SELECT inode_id FROM dirents WHERE parent_inode_id = ${String(s.rootInodeId)} AND name = ${target}
+		`;
+		const inodeId = BigInt(row!.inode_id);
+		await namespace.incrementNlink(tx, inodeId, s.id);
+		await namespace.insertDirent(tx, s.rootInodeId, alias, inodeId);
+	}
+
+	it("dispatches one lock and one fenced statement per composite, and two pairs per move", async () => {
+		const s = await create("counts");
+		const backfillsBefore = backfills;
+		const counts = await sql.begin(async (tx) => {
+			const mkdir = await executed(() => namespace.mkdirComposite(tx, s.id, s.rootInodeId, "source", 0o755, 0n));
+			const write = await executed(() =>
+				namespace.writeFileComposite(tx, s.id, s.rootInodeId, "file", 0o644, 1, sha, data, 1n),
+			);
+			const destination = await executed(() =>
+				namespace.mkdirComposite(tx, s.id, s.rootInodeId, "destination", 0o755, 2n),
+			);
+			const move = await executed(() =>
+				namespace.mvComposite(tx, s.id, s.rootInodeId, "source", s.rootInodeId, "destination", 3n),
+			);
+			const rm = await executed(() => namespace.rmComposite(tx, s.id, s.rootInodeId, "file", 5n));
+			return { mkdir, write, destination, move, rm };
+		});
+		for (const outcome of Object.values(counts)) expect(outcome.error).toBeUndefined();
+		expect(
+			Object.fromEntries(Object.entries(counts).map(([name, outcome]) => [name, outcome.statements.length])),
+		).toEqual({ mkdir: 2, write: 2, destination: 2, move: 4, rm: 2 });
+		const lock = "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))";
+		expect(counts.move.statements[0]).toBe(lock);
+		expect(counts.move.statements[1]).toMatch(/^WITH ctx AS \( SELECT set_config\('app\.sandbox_id', \$1, true\),/);
+		expect(counts.move.statements[2]).toBe(lock);
+		expect(counts.move.statements[3]).toMatch(/^WITH ctx AS \( SELECT set_config\('app\.sandbox_id', \$1, true\),/);
+		expect(backfills - backfillsBefore).toBe(1);
+		const after = await snapshot(s);
+		expect(after.version).toBe("6");
+	});
+
+	it("leaves every composite's writes unreachable when a caught ESTALE is committed", async () => {
+		const s = await create("stale-commit");
+		await sql.begin(async (tx) => {
+			await namespace.mkdirComposite(tx, s.id, s.rootInodeId, "directory", 0o755, 0n);
+			await namespace.writeFileComposite(tx, s.id, s.rootInodeId, "linked", 0o644, 1, sha, data, 1n);
+		});
+		await sql.begin((tx) => linkAlias(tx, s, "linked", "alias"));
+		const before = await snapshot(s);
+		expect(before.version).toBe("3");
+		const backfillsBefore = backfills;
+
+		const outcomes = await sql.begin(async (tx) => {
+			await sandbox.setSandboxContext(tx, s.id);
+			return [
+				await executed(() => namespace.mkdirComposite(tx, s.id, s.rootInodeId, "stale", 0o755, 2n)),
+				await executed(() => namespace.rmComposite(tx, s.id, s.rootInodeId, "alias", 2n)),
+				await executed(() => namespace.writeFileComposite(tx, s.id, s.rootInodeId, "linked", 0o644, 1, sha, data, 2n)),
+				await executed(() => namespace.mvComposite(tx, s.id, s.rootInodeId, "directory", s.rootInodeId, "alias", 2n)),
+			];
+		});
+
+		expect(outcomes.map((outcome) => outcome.code)).toEqual(["ESTALE", "ESTALE", "ESTALE", "ESTALE"]);
+		expect(outcomes.map((outcome) => outcome.statements.length)).toEqual([2, 2, 2, 2]);
+		expect(backfills).toBe(backfillsBefore);
+		expect(await snapshot(s)).toEqual(before);
+	});
+
+	it("leaves existing-context and ingest writes unreachable when a caught ESTALE is committed", async () => {
+		const s = await create("stale-existing");
+		await sql.begin((tx) => namespace.mkdirComposite(tx, s.id, s.rootInodeId, "directory", 0o755, 0n));
+		const before = await snapshot(s);
+		let batches = 0;
+
+		const outcomes = await sql.begin(async (tx) => {
+			await sandbox.setSandboxContext(tx, s.id);
+			return [
+				await executed(() => namespace.createInode(tx, { sandboxId: s.id, kind: 1, mode: 0o644, size: 0 }, 0n)),
+				await executed(() =>
+					bulkIngest(tx, [{ path: "/ingested.txt", content: data, mode: 0o644 }], s.id, 0n, {
+						async commitBatch() {
+							batches++;
+						},
+					}),
+				),
+			];
+		});
+
+		expect(outcomes.map((outcome) => outcome.code)).toEqual(["ESTALE", "ESTALE"]);
+		expect(outcomes.map((outcome) => outcome.statements.length)).toEqual([1, 1]);
+		expect(batches).toBe(0);
+		expect(await snapshot(s)).toEqual(before);
+	});
+
+	it("advances once per lower-level fenced mutation in its single statement", async () => {
+		const s = await create("existing");
+		const outcome = await sql.begin(async (tx) => {
+			await sandbox.setSandboxContextWithLock(tx, s.id);
+			return executed(() => namespace.createInode(tx, { sandboxId: s.id, kind: 2, mode: 0o755, size: 0 }, 0n));
+		});
+		expect(outcome.error).toBeUndefined();
+		expect(outcome.statements).toHaveLength(1);
+		expect(outcome.statements[0]).toMatch(/^WITH ctx AS \( SELECT s\.version FROM sandboxes s WHERE s\.id = \$1/);
+		expect((await snapshot(s)).version).toBe("1");
+	});
+
+	it("falls back to the transaction-local epoch and scopes RLS when no epoch or context is set", async () => {
+		const s = await create("undefined-epoch");
+		const settings = await sql.begin(async (tx) => {
+			await namespace.mkdirComposite(tx, s.id, s.rootInodeId, "first", 0o755);
+			await namespace.mkdirComposite(tx, s.id, s.rootInodeId, "second", 0o755);
+			const [row] = await tx<{ epoch: string; context: string }[]>`
+				SELECT current_setting('app.sandbox_epoch') AS epoch, current_setting('app.sandbox_id') AS context
+			`;
+			return row!;
+		});
+		expect(settings).toEqual({ epoch: "2", context: s.id });
+		expect((await snapshot(s)).version).toBe("2");
+	});
+
+	it("refuses a composite for another sandbox and keeps the active RLS context", async () => {
+		const active = await create("active");
+		const target = await create("target");
+		const before = await snapshot(target);
+		const outcome = await sql.begin(async (tx) => {
+			await sandbox.setSandboxContext(tx, active.id);
+			const call = await executed(() =>
+				namespace.mkdirComposite(tx, target.id, target.rootInodeId, "cross", 0o755, 0n),
+			);
+			const [row] = await tx<{ id: string }[]>`SELECT current_setting('app.sandbox_id') AS id`;
+			return { call, context: row!.id };
+		});
+		expect(outcome.call.error).toBeInstanceOf(Error);
+		expect((outcome.call.error as Error).message).toBe("mkdirComposite: INSERT returned no rows");
+		expect(outcome.call.statements).toHaveLength(2);
+		expect(outcome.context).toBe(active.id);
+		expect(await snapshot(target)).toEqual(before);
+	});
+
+	it("keeps link counts when rm and move drop one name of a hardlinked inode", async () => {
+		const s = await create("hardlink");
+		await sql.begin(async (tx) => {
+			await namespace.writeFileComposite(tx, s.id, s.rootInodeId, "linked", 0o644, 1, sha, data, 0n);
+			await namespace.writeFileComposite(tx, s.id, s.rootInodeId, "other", 0o644, 1, sha, data, 1n);
+		});
+		await sql.begin((tx) => linkAlias(tx, s, "linked", "alias"));
+		const inodeOf = (snap: Snapshot, name: string): string => snap.entries!.find((entry) => entry.name === name)!.inode;
+		const nlinkOf = (snap: Snapshot, id: string): number | undefined =>
+			snap.inodes!.find((inode) => inode.id === id)?.nlink;
+		const seeded = await snapshot(s);
+		const shared = inodeOf(seeded, "linked");
+		const other = inodeOf(seeded, "other");
+		expect(nlinkOf(seeded, shared)).toBe(2);
+
+		await sql.begin((tx) => namespace.rmComposite(tx, s.id, s.rootInodeId, "alias", 3n));
+		const afterRm = await snapshot(s);
+		expect(nlinkOf(afterRm, shared)).toBe(1);
+
+		await sql.begin((tx) => linkAlias(tx, s, "linked", "alias"));
+		await sql.begin((tx) => namespace.mvComposite(tx, s.id, s.rootInodeId, "other", s.rootInodeId, "alias", 5n));
+		const afterMove = await snapshot(s);
+		expect(nlinkOf(afterMove, shared)).toBe(1);
+		expect(inodeOf(afterMove, "alias")).toBe(other);
+		expect(inodeOf(afterMove, "linked")).toBe(shared);
+		expect(afterMove.version).toBe("7");
+
+		await sql.begin((tx) => namespace.mvComposite(tx, s.id, s.rootInodeId, "alias", s.rootInodeId, "linked", 7n));
+		const afterReplace = await snapshot(s);
+		expect(nlinkOf(afterReplace, shared)).toBeUndefined();
+		expect(inodeOf(afterReplace, "linked")).toBe(other);
+		expect(afterReplace.version).toBe("9");
+	});
+
+	async function observeBlocker({ holderPid, waiterPid }: Contenders): Promise<void> {
+		const deadline = Date.now() + 10_000;
+		let observed: LockObservation | undefined;
+		do {
+			observed = await sql.begin(async (tx) => {
+				await tx`SET LOCAL statement_timeout = '1s'`;
+				const [row] = await tx<LockObservation[]>`
+					SELECT ${holderPid} = ANY(pg_blocking_pids(${waiterPid})) AS blocked,
+						EXISTS (
+							SELECT 1 FROM pg_locks holder JOIN pg_locks waiter
+								ON holder.database = waiter.database AND holder.classid = waiter.classid
+								AND holder.objid = waiter.objid AND holder.objsubid = waiter.objsubid
+							WHERE holder.pid = ${holderPid} AND waiter.pid = ${waiterPid}
+								AND holder.locktype = 'advisory' AND waiter.locktype = 'advisory'
+								AND holder.granted AND NOT waiter.granted
+						) AS matching_lock
+				`;
+				return row;
+			});
+			if (observed?.blocked && observed.matching_lock) return;
+			await pause(25);
+		} while (Date.now() < deadline);
+		throw new Error(`advisory lock wait not observed: ${JSON.stringify({ holderPid, waiterPid, observed })}`);
+	}
+
+	async function runContendedFence(
+		s: Seeded,
+		isolation: Isolation,
+		expectedEpoch: bigint,
+		observe: (contenders: Contenders) => Promise<void> = observeBlocker,
+	): Promise<PromiseSettledResult<void>> {
+		const release = deferred<void>();
+		const holderReady = deferred<number>();
+		const waiterReady = deferred<number>();
+		const tasks: Promise<PromiseSettledResult<void>>[] = [];
+		try {
+			const holder = settled(
+				sql.begin(async (tx) => {
+					await tx`SET LOCAL statement_timeout = '10s'`;
+					await takeWriterLock(tx, s.id);
+					const [row] = await tx<{ pid: number }[]>`SELECT pg_backend_pid() AS pid`;
+					expect(row).toBeDefined();
+					holderReady.resolve(row!.pid);
+					await release.promise;
+					await namespace.mkdirComposite(tx, s.id, s.rootInodeId, "holder", 0o755, 0n);
+				}),
+			);
+			tasks.push(holder);
+			const holderPid = await readyPid(holderReady.promise, holder);
+			const waiter = settled(
+				sql.begin(`isolation level ${isolation}`, async (tx) => {
+					await tx`SET LOCAL statement_timeout = '10s'`;
+					await sandbox.setSandboxContext(tx, s.id);
+					const [row] = await tx<{ pid: number; version: string; isolation: string }[]>`
+						SELECT pg_backend_pid() AS pid, version::text,
+							current_setting('transaction_isolation') AS isolation FROM sandboxes WHERE id = ${s.id}
+					`;
+					expect(row).toMatchObject({ version: "0", isolation });
+					waiterReady.resolve(row!.pid);
+					await namespace.mkdirComposite(tx, s.id, s.rootInodeId, "waiter", 0o755, expectedEpoch);
+				}),
+			);
+			tasks.push(waiter);
+			const waiterPid = await readyPid(waiterReady.promise, waiter);
+			await observe({ holderPid, waiterPid });
+			release.resolve();
+			const holderOutcome = await holder;
+			if (holderOutcome.status === "rejected") throw holderOutcome.reason;
+			return await waiter;
+		} finally {
+			release.resolve();
+			await Promise.all(tasks);
+		}
+	}
+
+	it.each([
+		{ isolation: "read committed", pin: 0n, verdict: "ESTALE", version: "1" },
+		{ isolation: "read committed", pin: 1n, verdict: "accepted", version: "2" },
+		{ isolation: "repeatable read", pin: 0n, verdict: "40001", version: "1" },
+		{ isolation: "repeatable read", pin: 1n, verdict: "ESTALE", version: "1" },
+	] satisfies { isolation: Isolation; pin: bigint; verdict: string; version: string }[])(
+		"applies $verdict under $isolation with snapshot 0 and pin $pin after an observed lock wait",
+		async ({ isolation, pin, verdict, version }) => {
+			const s = await create(`lock-${isolation}-${pin}`);
+			const outcome = await runContendedFence(s, isolation, pin);
+			if (verdict === "accepted") {
+				expect(outcome.status).toBe("fulfilled");
+			} else {
+				expect(outcome).toMatchObject({ status: "rejected", reason: { code: verdict } });
+			}
+			const after = await snapshot(s);
+			expect(after.version).toBe(version);
+			const names = after.entries!.map((entry) => entry.name);
+			expect(names).toContain("holder");
+			expect(names.includes("waiter")).toBe(verdict === "accepted");
+		},
+	);
+
+	it("drains both transactions after an observer failure and makes their connections reusable", async () => {
+		const s = await create("observer-failure");
+		const failure = new Error("injected observer failure");
+		const pids: number[] = [];
+		await expect(
+			runContendedFence(s, "read committed", 0n, async (contenders) => {
+				await observeBlocker(contenders);
+				pids.push(contenders.holderPid, contenders.waiterPid);
+				throw failure;
+			}),
+		).rejects.toBe(failure);
+		const connections = await Promise.all(Array.from({ length: 4 }, () => sql.reserve()));
+		try {
+			const reused = await Promise.all(
+				connections.map(async (connection) => {
+					const [row] = await connection<{ pid: number; advisory_locks: number }[]>`
+						SELECT pg_backend_pid() AS pid,
+							(SELECT count(*)::int FROM pg_locks WHERE pid = pg_backend_pid() AND locktype = 'advisory') AS advisory_locks
+					`;
+					return row;
+				}),
+			);
+			expect(pids).toHaveLength(2);
+			for (const pid of pids) expect(reused).toContainEqual({ pid, advisory_locks: 0 });
+		} finally {
+			for (const connection of connections) connection.release();
+		}
+		const after = await snapshot(s);
+		expect(after.version).toBe("1");
+		const names = after.entries!.map((entry) => entry.name);
+		expect(names).toContain("holder");
+		expect(names).not.toContain("waiter");
+	});
+
+	it.each([
+		["existing", fenceAndAdvance],
+		["composite", compositeFenceAndAdvance],
+	])("builds the %s fence fragment without dispatching a statement", async (label, build) => {
+		const s = await create(`lazy-${label}`);
+		const outcome = await sql.begin(async (tx) => {
+			const statements: string[] = [];
+			recording = statements;
+			try {
+				const fragment = build(tx, s.id, 0n);
+				await Promise.resolve();
+				await new Promise((resolve) => setImmediate(resolve));
+				const afterConstruction = statements.length;
+				await takeWriterLock(tx, s.id);
+				const [row] = await tx<{ fenced: number }[]>`WITH ${fragment} SELECT count(*)::int AS fenced FROM epoch`;
+				return { afterConstruction, afterStatement: statements.length, fenced: row!.fenced };
+			} finally {
+				recording = undefined;
+			}
+		});
+		expect(outcome).toEqual({ afterConstruction: 0, afterStatement: 2, fenced: 1 });
+	});
+});
+
+function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } {
+	let resolve!: (value: T) => void;
+	const promise = new Promise<T>((done) => {
+		resolve = done;
+	});
+	return { promise, resolve };
+}
+
+function settled(task: Promise<void>): Promise<PromiseSettledResult<void>> {
+	return task.then(
+		(value) => ({ status: "fulfilled", value }),
+		(reason: unknown) => ({ status: "rejected", reason }),
+	);
+}
+
+function readyPid(pid: Promise<number>, task: Promise<PromiseSettledResult<void>>): Promise<number> {
+	return Promise.race([
+		pid,
+		task.then((outcome) => {
+			if (outcome.status === "rejected") throw outcome.reason;
+			throw new Error("contender completed before reporting its backend PID");
+		}),
+	]);
+}
diff --git a/src/sql-fs/tests/postgres.fencing.test.ts b/src/sql-fs/tests/postgres.fencing.test.ts
index 30b33683..a522329a 100644
--- a/src/sql-fs/tests/postgres.fencing.test.ts
+++ b/src/sql-fs/tests/postgres.fencing.test.ts
@@ -2,17 +2,37 @@ import { describe, expect, it } from "vitest";
 import { PostgresDialect } from "../dialects/postgres.js";
 
 type RecordedSqlCall = { sql: string; values: readonly unknown[] };
+type Rows = unknown[] | ((sql: string) => unknown[]);
 
-function recordingTx(rows: unknown[] = [{ id: "42", inode_id: "42", new_inode_id: "42", removed_inode_id: "42" }]): {
-	tx: ((strings: TemplateStringsArray, ...values: unknown[]) => Promise<unknown[]>) & object;
+function recordingTx(rows: Rows = [{ id: "42", inode_id: "42", new_inode_id: "42", removed_inode_id: "42" }]): {
+	tx: never;
 	calls: RecordedSqlCall[];
 } {
 	const calls: RecordedSqlCall[] = [];
-	const tx = ((strings: TemplateStringsArray, ...values: unknown[]) => {
-		calls.push({ sql: strings.join("?"), values });
-		return Promise.resolve(rows);
-	}) as ((strings: TemplateStringsArray, ...values: unknown[]) => Promise<unknown[]>) & object;
-	return { tx, calls };
+	const recorded = new WeakMap<Promise<unknown[]>, RecordedSqlCall>();
+	const tx = (strings: TemplateStringsArray, ...values: unknown[]): Promise<unknown[]> => {
+		let sql = strings[0]!;
+		const flat: unknown[] = [];
+		values.forEach((value, index) => {
+			const fragment = value instanceof Promise ? recorded.get(value) : undefined;
+			if (fragment === undefined) {
+				sql += "?";
+				flat.push(value);
+			} else {
+				sql += fragment.sql;
+				flat.push(...fragment.values);
+				const at = calls.indexOf(fragment);
+				if (at >= 0) calls.splice(at, 1);
+			}
+			sql += strings[index + 1]!;
+		});
+		const call = { sql: sql.replace(/\s+/g, " ").trim(), values: flat };
+		calls.push(call);
+		const query = Promise.resolve(typeof rows === "function" ? rows(call.sql) : rows);
+		recorded.set(query, call);
+		return query;
+	};
+	return { tx: tx as never, calls };
 }
 
 describe("PostgresDialect epoch fencing", () => {
@@ -20,7 +40,7 @@ describe("PostgresDialect epoch fencing", () => {
 		const dialect = new PostgresDialect("postgres://stub");
 		const makeCall = (run: (tx: never) => Promise<unknown>) => {
 			const recording = recordingTx();
-			return run(recording.tx as never).then(() => recording);
+			return run(recording.tx).then(() => recording);
 		};
 		const recordings = await Promise.all([
 			makeCall((tx) => dialect.mkdirComposite!(tx, "s1", 1n, "new", 0o755, 7n)),
@@ -30,12 +50,11 @@ describe("PostgresDialect epoch fencing", () => {
 			),
 		]);
 		for (const recording of recordings) {
-			// calls[0] takes the writer lock; calls[1] runs the fenced CTE.
 			expect(recording.calls).toHaveLength(2);
+			expect(recording.calls[0]!.sql).toBe("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))");
 			const sql = recording.calls[1]!.sql;
+			expect(sql).toContain("WITH ctx AS ( SELECT set_config('app.sandbox_id', ?, true)");
 			expect(sql).toContain("pg_advisory_xact_lock");
-			expect(sql).toContain("FROM sandboxes");
-			expect(sql).toContain("version");
 			expect(recording.calls[1]!.values).toContain("7");
 		}
 	});
@@ -43,50 +62,36 @@ describe("PostgresDialect epoch fencing", () => {
 	it("throws the fenced no-row error when the ctx yields nothing", async () => {
 		const dialect = new PostgresDialect("postgres://stub");
 		const empty = recordingTx([]);
-		await expect(dialect.mkdirComposite!(empty.tx as never, "s1", 1n, "new", 0o755, 7n)).rejects.toThrow(
+		await expect(dialect.mkdirComposite!(empty.tx, "s1", 1n, "new", 0o755, 7n)).rejects.toThrow(
 			"mkdirComposite: INSERT returned no rows",
 		);
 		await expect(
-			dialect.writeFileComposite!(
-				empty.tx as never,
-				"s1",
-				1n,
-				"file",
-				0o644,
-				3,
-				new Uint8Array(32),
-				new Uint8Array(3),
-				7n,
-			),
+			dialect.writeFileComposite!(empty.tx, "s1", 1n, "file", 0o644, 3, new Uint8Array(32), new Uint8Array(3), 7n),
 		).rejects.toThrow("writeFileComposite: INSERT returned no rows");
 	});
 
 	it("keeps the move's second statement inside the same epoch fence", async () => {
 		const dialect = new PostgresDialect("postgres://stub");
 		const recording = recordingTx();
-		await dialect.mvComposite!(recording.tx as never, "s1", 1n, "old", 2n, "new", 7n);
+		await dialect.mvComposite!(recording.tx, "s1", 1n, "old", 2n, "new", 7n);
 		expect(recording.calls).toHaveLength(4);
+		expect(recording.calls[2]!.sql).toBe("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))");
 		expect(recording.calls[3]!.sql).toContain("WITH ctx AS");
 		expect(recording.calls[3]!.sql).toContain("pg_advisory_xact_lock");
-		expect(recording.calls[3]!.sql).toContain("version");
 	});
 
 	it("returns a strictly advanced epoch on recreation and deletion", async () => {
 		const dialect = new PostgresDialect("postgres://stub");
-		const recording = recordingTx();
-		recording.tx = ((strings: TemplateStringsArray, ...values: unknown[]) => {
-			const sql = strings.join("?");
-			recording.calls.push({ sql, values });
-			if (sql.includes("DELETE FROM sandboxes")) return Promise.resolve([{ epoch: "10" }]);
-			if (sql.includes("sandbox_epochs")) return Promise.resolve([{ epoch: "9" }]);
-			if (sql.includes("INSERT INTO sandboxes"))
-				return Promise.resolve([{ created_at: new Date("2026-01-01T00:00:00Z") }]);
-			return Promise.resolve([{ id: "42" }]);
-		}) as typeof recording.tx;
+		const recording = recordingTx((sql) => {
+			if (sql.includes("DELETE FROM sandboxes")) return [{ epoch: "10" }];
+			if (sql.includes("sandbox_epochs")) return [{ epoch: "9" }];
+			if (sql.includes("INSERT INTO sandboxes")) return [{ created_at: new Date("2026-01-01T00:00:00Z") }];
+			return [{ id: "42" }];
+		});
 
-		const created = await dialect.createSandbox(recording.tx as never, "s1");
+		const created = await dialect.createSandbox(recording.tx, "s1");
 		expect(created.epoch).toBe(9n);
-		const deleted = await dialect.deleteSandbox(recording.tx as never, "s1");
+		const deleted = await dialect.deleteSandbox(recording.tx, "s1");
 		expect(deleted).toBe(10n);
 		const deletion = recording.calls.find((call) => call.sql.includes("DELETE FROM sandboxes"));
 		expect(deletion?.sql).toContain("version + 1");