fix(mcp): защитить операции обслуживания от блокировок строк

backend/src/modules/mcp/routing-community-scoped-recovery.ts

@@ -440,5 +440,17 @@ export function registerScopedRecoveryService() {
 
 /** Restart maintenance is read-only with respect to the network. */
 export async function recoverScopedRecoveryIntents() {
-  await withPartitionWrite(({ client }) => client.query("UPDATE mcp_scoped_recovery_intents r SET state='unknown',finished_at=now() WHERE r.state='intent' AND (r.deadline<=now() OR r.recovery_operation_id IS NOT NULL AND EXISTS (SELECT 1 FROM mcp_operations o WHERE o.id=r.recovery_operation_id AND o.state NOT IN ('pending','running')))"))
+  await withPartitionWrite(({ client }) => client.query(`WITH interrupted AS (
+    SELECT intent.source_operation_id,intent.source_step_index
+    FROM mcp_scoped_recovery_intents intent
+    WHERE intent.state='intent' AND (intent.deadline<=now() OR EXISTS (
+      SELECT 1 FROM mcp_operations operation
+      WHERE operation.id=intent.recovery_operation_id AND operation.state NOT IN ('pending','running')
+    ))
+    ORDER BY intent.deadline,intent.source_operation_id,intent.source_step_index
+    LIMIT 1000 FOR UPDATE SKIP LOCKED
+  )
+  UPDATE mcp_scoped_recovery_intents intent SET state='unknown',finished_at=now()
+  FROM interrupted WHERE intent.source_operation_id=interrupted.source_operation_id
+    AND intent.source_step_index=interrupted.source_step_index`))
 }

backend/src/modules/mcp/runtime.ts

@@ -780,18 +780,39 @@ async function performReservedRecovery(operation: StoredOperation, plan: StoredP
 }
 export async function recoverMcpOperations() {
   // A second backend must not interrupt a runner still holding its session lease.
-  const { rows } = await pool.query("SELECT id,connection_id FROM mcp_operations WHERE state IN ('pending','running') AND created_at<now()-interval '2 minutes' ORDER BY created_at LIMIT 100")
+  const { rows } = await pool.query("SELECT id FROM mcp_operations WHERE state IN ('pending','running') AND created_at<now()-interval '2 minutes' ORDER BY created_at,id LIMIT 100")
   for (const row of rows) {
     await withPartitionWrite(async ({ client }) => {
       const { rows: [lock] } = await client.query("SELECT pg_try_advisory_xact_lock(hashtextextended($1,0)) AS locked", ["mcp-operation:" + row.id])
       if (!lock.locked) return
-      const changed = await client.query("UPDATE mcp_operations SET state='unknown',verification='inconclusive',error_code='PROCESS_INTERRUPTED',finished_at=now() WHERE id=$1 AND state IN ('pending','running') RETURNING plan_id", [row.id])
-      if (!changed.rows.length) return
-      await client.query("UPDATE mcp_plans SET state='completed' WHERE id=$1", [changed.rows[0].plan_id])
-      await audit(client, "mcp-recovery", "operation.interrupted", row.connection_id, { operationId: row.id, retryAllowed: false })
+      // Defer the whole transition when either row is busy.
+      const { rows: [target] } = await client.query(`SELECT o.plan_id,o.connection_id
+        FROM mcp_operations o JOIN mcp_plans p ON p.id=o.plan_id
+        WHERE o.id=$1 AND o.state IN ('pending','running') AND o.created_at<now()-interval '2 minutes'
+        FOR UPDATE OF o,p SKIP LOCKED`, [row.id])
+      if (!target) return
+      await client.query("UPDATE mcp_operations SET state='unknown',verification='inconclusive',error_code='PROCESS_INTERRUPTED',finished_at=now() WHERE id=$1", [row.id])
+      await client.query("UPDATE mcp_plans SET state='completed' WHERE id=$1", [target.plan_id])
+      await audit(client, "mcp-recovery", "operation.interrupted", target.connection_id, { operationId: row.id, retryAllowed: false })
     })
   }
-  await withPartitionWrite(({ client }) => client.query("UPDATE mcp_plans SET state='expired' WHERE expires_at<now() AND state IN ('draft','approved','awaiting-approval')"))
-  await withPartitionWrite(({ client }) => client.query("DELETE FROM mcp_read_leases WHERE expires_at<now()"))
-  await withPartitionWrite(({ client }) => client.query("UPDATE mcp_recovery_intents SET state='unknown',finished_at=now() WHERE state='intent' AND operation_id IN (SELECT id FROM mcp_operations WHERE state NOT IN ('pending','running'))"))
+  await withPartitionWrite(({ client }) => client.query(`WITH expired AS (
+    SELECT id FROM mcp_plans
+    WHERE expires_at<now() AND state IN ('draft','approved','awaiting-approval')
+    ORDER BY expires_at,id LIMIT 1000 FOR UPDATE SKIP LOCKED
+  )
+  UPDATE mcp_plans SET state='expired' FROM expired WHERE mcp_plans.id=expired.id`))
+  await withPartitionWrite(({ client }) => client.query(`WITH expired AS (
+    SELECT id FROM mcp_read_leases WHERE expires_at<now()
+    ORDER BY expires_at,id LIMIT 1000 FOR UPDATE SKIP LOCKED
+  )
+  DELETE FROM mcp_read_leases USING expired WHERE mcp_read_leases.id=expired.id`))
+  await withPartitionWrite(({ client }) => client.query(`WITH interrupted AS (
+    SELECT intent.operation_id FROM mcp_recovery_intents intent
+    JOIN mcp_operations operation ON operation.id=intent.operation_id
+    WHERE intent.state='intent' AND operation.state NOT IN ('pending','running')
+    ORDER BY intent.created_at,intent.operation_id LIMIT 1000 FOR UPDATE OF intent SKIP LOCKED
+  )
+  UPDATE mcp_recovery_intents SET state='unknown',finished_at=now() FROM interrupted
+  WHERE mcp_recovery_intents.operation_id=interrupted.operation_id`))
 }

backend/src/modules/mcp/store.ts

@@ -7,6 +7,7 @@ import {
 } from "@mmapp/contracts/mcp"
 import { pool } from "../../db/index.js"
 import { hasPermission, type AuthUser } from "../../lib/permissions.js"
+import { admitPartitionWriter, withPartitionWrite } from "../../services/partition-write-transaction.js"
 import { accessGeneration, mcpStateDirectory, canonicalMcpUrl, McpError, randomToken, redact, tokenDigest } from "./security.js"
 import { assertPortalClaimsCurrent, lockPortalOwner, readCurrentPrincipalSecuritySource } from "./portal-revocation.js"
 
@@ -54,6 +55,13 @@ export async function transaction<T>(work: (client: PoolClient) => Promise<T>):
   try { await client.query("BEGIN"); const result = await work(client); await client.query("COMMIT"); return result }
   catch (error) { await client.query("ROLLBACK"); throw error } finally { client.release() }
 }
+export function connectionWrite<T>(work: (client: PoolClient) => Promise<T>): Promise<T> {
+  // Keep domain errors from connection validation visible to the HTTP handler.
+  return transaction(async client => {
+    await admitPartitionWriter(client)
+    return work(client)
+  })
+}
 export async function audit(client: Pick<PoolClient, "query">, actorId: string, action: string, connectionId: string | null, payload: unknown): Promise<void> {
   await client.query("INSERT INTO mcp_audit(actor_id,action,connection_id,payload) VALUES($1,$2,$3,$4::jsonb)", [actorId, action, connectionId, JSON.stringify(redact(payload))])
 }
@@ -64,7 +72,7 @@ export async function getMcpSettings(retentionOwner?: PoolClient) {
   return mcpSettingsSchema.parse(row)
 }
 export async function updateMcpSettings(patch: Partial<z.infer<typeof mcpSettingsSchema>>, actor: AuthUser) {
-  return transaction(async client => {
+  return withPartitionWrite(async ({ client }) => {
     const { rows: [before] } = await client.query("SELECT * FROM mcp_settings WHERE id=1 FOR UPDATE")
     await client.query("UPDATE mcp_settings SET enabled=COALESCE($1,enabled),requests_retention_days=COALESCE($2,requests_retention_days),audit_retention_days=COALESCE($3,audit_retention_days),updated_at=now() WHERE id=1", [patch.enabled, patch.requestsRetentionDays, patch.auditRetentionDays])
     await audit(client, actor.id, "settings.update", null, { before, patch })
@@ -106,7 +114,7 @@ export async function createMcpConnection(input: McpConnectionCreate, actor: Aut
   const { name, ...configuration } = input
   const policy = mcpConnectionPolicySchema.parse({ ...configuration, id, version: 1, status: "active" })
   const token = randomToken("bearer")
-  return transaction(async client => {
+  return connectionWrite(async client => {
     await lockPortalOwner(client, actor.id)
     await assertPortalClaimsCurrent(actor, client)
     const { rows: [row] } = await client.query<ConnectionRow>("INSERT INTO mcp_connections(id,name,owner_id,owner_permissions,policy,generation,owner_valid_until,owner_claims_issued_at) VALUES($1,$2,$3,$4::jsonb,$5::jsonb,$6,$7,$8) RETURNING *", [id, name, actor.id, JSON.stringify(actor.permissions), JSON.stringify(policy), generation, ownerValidUntil(actor), new Date(actor.issuedAt ?? 0)])
@@ -119,7 +127,7 @@ export async function updateMcpConnection(id: string, input: McpConnectionCreate
   if (input.permissions.some(permission => !hasPermission(actor.permissions, permission))) throw new McpError("PERMISSION_DENIED", "Нельзя делегировать отсутствующие права")
   if (Date.parse(input.expiresAt) > Date.now() + 366 * 86400000) throw new McpError("INVALID_EXPIRATION", "Срок доступа не может превышать год", 400)
   if (input.scope.serverIds.length && (await pool.query("SELECT id FROM servers WHERE id=ANY($1::bigint[])", [input.scope.serverIds])).rows.length !== input.scope.serverIds.length) throw new McpError("SERVER_DENIED", "Выбран неизвестный сервер", 400)
-  return transaction(async client => {
+  return connectionWrite(async client => {
     await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [id])
     const before = await getConnection(id, client)
     if (before.owner_id !== actor.id) throw new McpError("PERMISSION_DENIED", "Политику обновляет её владелец после подтверждения своих прав")
@@ -133,7 +141,7 @@ export async function updateMcpConnection(id: string, input: McpConnectionCreate
   })
 }
 export async function setConnectionStatus(id: string, status: "active" | "paused" | "revoked", actor: AuthUser) {
-  return transaction(async client => {
+  return connectionWrite(async client => {
     await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [id])
     const row = await getConnection(id, client)
     if (row.policy.status === "revoked" && status !== "revoked") throw new McpError("CONNECTION_REVOKED", "Отозванное подключение нельзя активировать", 409)
@@ -145,7 +153,7 @@ export async function setConnectionStatus(id: string, status: "active" | "paused
   })
 }
 export async function revalidateMcpOwner(id: string, actor: AuthUser) {
-  return transaction(async client => {
+  return connectionWrite(async client => {
     await lockPortalOwner(client, actor.id)
     await assertPortalClaimsCurrent(actor, client)
     await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [id])
@@ -162,7 +170,7 @@ export async function revalidateMcpOwner(id: string, actor: AuthUser) {
 }
 export async function rotateConnectionToken(id: string, actor: AuthUser) {
   const token = randomToken("bearer")
-  return transaction(async client => {
+  return connectionWrite(async client => {
     await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [id])
     const row = await getConnection(id, client)
     if (row.policy.status === "revoked") throw new McpError("CONNECTION_REVOKED", "Доступ отозван", 409)

backend/src/routes/mcp.ts

@@ -16,7 +16,7 @@ import { listActions } from "../modules/mcp/actions.js"
 import { pool } from "../db/index.js"
 import { hasPermission, type AuthUser } from "../lib/permissions.js"
 import { accessGeneration, decryptSecret, encryptSecret, McpError, randomToken, redact, tokenDigest } from "../modules/mcp/security.js"
-import { createMcpConnection, listMcpConnections, getMcpSettings, updateMcpSettings, getConnection, assertConnectionVisible, rotateConnectionToken, setConnectionStatus, updateMcpConnection, transaction, audit } from "../modules/mcp/store.js"
+import { createMcpConnection, listMcpConnections, getMcpSettings, updateMcpSettings, getConnection, assertConnectionVisible, rotateConnectionToken, setConnectionStatus, updateMcpConnection, transaction, connectionWrite, audit } from "../modules/mcp/store.js"
 import { approvePlan, loadPlan, publicPlan, readMcpPolicyPreviewInput } from "../modules/mcp/runtime.js"
 import { applyPlan, invokeMcpTool } from "../modules/mcp/runtime.js"
 import { findOwnerPlanReplayTombstone } from "../modules/mcp/operation-replay-tombstones.js"
@@ -298,12 +298,14 @@ export function createMcpRoutes(checkClaims = assertPortalClaimsCurrent, options
   app.delete("/mcp/protections/:id", async request => { const id = idOf(request.params); await transaction(async client => { await client.query("DELETE FROM mcp_protections WHERE id=$1", [id]); await audit(client, actorOf(request.authUser).id, "protection.delete", null, { id }) }); return { ok: true } })
   app.post("/mcp/connections/:id/grants", async request => {
     const connectionId = idOf(request.params)
-    const input = mcpTemporaryGrantSchema.parse(request.body), connection = await getConnection(connectionId), actor = actorOf(request.authUser)
-    assertConnectionVisible(connection, actor)
-    if (Date.parse(input.expiresAt) <= Date.now() || connection.policy.status !== "active" || connection.generation !== await accessGeneration()) throw new McpError("GRANT_INACTIVE", "Подключение или временный доступ недействительны", 400)
-    if (input.scope.operationIds.some(id => !connection.policy.scope.operationIds.includes(id)) || input.scope.serverIds.some(id => !connection.policy.scope.serverIds.includes(id)) || input.scope.allowGlobal && !connection.policy.scope.allowGlobal || Date.parse(input.expiresAt) > Math.min(Date.parse(connection.policy.expiresAt), connection.owner_valid_until.getTime())) throw new McpError("PERMISSION_DENIED", "Временный доступ не может расширять основной доступ и срок подтверждённых прав владельца")
+    const input = mcpTemporaryGrantSchema.parse(request.body), actor = actorOf(request.authUser)
     const credentialId = randomUUID(), token = randomToken("bearer")
-    await transaction(async client => {
+    await connectionWrite(async client => {
+      await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [connectionId])
+      const connection = await getConnection(connectionId, client)
+      assertConnectionVisible(connection, actor)
+      if (Date.parse(input.expiresAt) <= Date.now() || connection.policy.status !== "active" || connection.generation !== await accessGeneration()) throw new McpError("GRANT_INACTIVE", "Подключение или временный доступ недействительны", 400)
+      if (input.scope.operationIds.some(id => !connection.policy.scope.operationIds.includes(id)) || input.scope.serverIds.some(id => !connection.policy.scope.serverIds.includes(id)) || input.scope.allowGlobal && !connection.policy.scope.allowGlobal || Date.parse(input.expiresAt) > Math.min(Date.parse(connection.policy.expiresAt), connection.owner_valid_until.getTime())) throw new McpError("PERMISSION_DENIED", "Временный доступ не может расширять основной доступ и срок подтверждённых прав владельца")
       await client.query("INSERT INTO mcp_credentials(id,connection_id,digest,kind,expires_at) VALUES($1,$2,$3,'bearer',$4)", [credentialId, connectionId, tokenDigest(token), input.expiresAt])
       await client.query("INSERT INTO mcp_grants(id,connection_id,payload) VALUES($1,$2,$3::jsonb)", [randomUUID(), connectionId, JSON.stringify({ ...input, credentialId })])
       await audit(client, actor.id, "grant.create", connectionId, input)
@@ -316,13 +318,17 @@ export function createMcpRoutes(checkClaims = assertPortalClaimsCurrent, options
   })
   app.delete("/mcp/grants/:id", async request => {
     const id = idOf(request.params), actor = actorOf(request.authUser)
-    await transaction(async client => {
-      const grant = (await client.query("SELECT * FROM mcp_grants WHERE id=$1 FOR UPDATE", [id])).rows[0]
+    await connectionWrite(async client => {
+      const { rows: [parent] } = await client.query("SELECT connection_id FROM mcp_grants WHERE id=$1", [id])
+      if (!parent) throw new McpError("NOT_FOUND", "Допуск не найден", 404)
+      await client.query("SELECT id FROM mcp_connections WHERE id=$1 FOR UPDATE", [parent.connection_id])
+      const grant = (await client.query("SELECT payload FROM mcp_grants WHERE id=$1 AND connection_id=$2", [id, parent.connection_id])).rows[0]
       if (!grant) throw new McpError("NOT_FOUND", "Допуск не найден", 404)
-      assertConnectionVisible(await getConnection(grant.connection_id, client), actor)
-      await client.query("UPDATE mcp_grants SET revoked_at=now() WHERE id=$1", [id])
-      await client.query("UPDATE mcp_credentials SET revoked_at=now() WHERE id=$1", [grant.payload.credentialId])
-      await audit(client, actor.id, "grant.revoke", grant.connection_id, { grantId: id })
+      assertConnectionVisible(await getConnection(parent.connection_id, client), actor)
+      await client.query("UPDATE mcp_credentials SET revoked_at=now() WHERE id=$1 AND connection_id=$2", [grant.payload.credentialId, parent.connection_id])
+      const updated = await client.query("UPDATE mcp_grants SET revoked_at=now() WHERE id=$1 AND connection_id=$2", [id, parent.connection_id])
+      if (!updated.rowCount) throw new McpError("NOT_FOUND", "Допуск не найден", 404)
+      await audit(client, actor.id, "grant.revoke", parent.connection_id, { grantId: id })
     })
     return { ok: true }
   })

backend/src/services/server-poll-terminal-archive.owned.test.ts

@@ -209,9 +209,57 @@ async function main() {
     }
     let commitAckLossAccepted = false
     if (historyVariant) {
+      const lockedPlan = randomUUID(), freePlan = randomUUID(), interruptedPlan = randomUUID()
+      const terminalPlan = randomUUID()
+      const fixturePlans = [lockedPlan, freePlan, interruptedPlan, terminalPlan]
+      await pool.query(`INSERT INTO mcp_plans(id,connection_id,operation_id,operation_version,policy_version,generation,payload,evidence,state,expires_at)
+        SELECT fixture.id,source.connection_id,source.operation_id,source.operation_version,source.policy_version,
+          source.generation,'{}'::jsonb,'{}'::jsonb,'draft',now()-interval '1 minute'
+        FROM mcp_plans source CROSS JOIN unnest($2::uuid[]) AS fixture(id) WHERE source.id=$1`, [a.plan_id, fixturePlans])
+      await pool.query("UPDATE mcp_plans SET state='executing' WHERE id=$1", [interruptedPlan])
+      await pool.query("UPDATE mcp_plans SET state='completed' WHERE id=$1", [terminalPlan])
+      const operation = randomUUID(), terminalOperation = randomUUID(), readLease = randomUUID(), recoveryProfile = randomUUID()
+      await pool.query(`INSERT INTO mcp_operations(id,plan_id,connection_id,idempotency_key,state,created_at)
+        SELECT $1,id,connection_id,$1::uuid::text,'running',now()-interval '3 minutes'
+        FROM mcp_plans WHERE id=$2`, [operation, interruptedPlan])
+      await pool.query(`INSERT INTO mcp_operations(id,plan_id,connection_id,idempotency_key,state,verification,finished_at)
+        SELECT $1,id,connection_id,$1::uuid::text,'applied','passed',now()
+        FROM mcp_plans WHERE id=$2`, [terminalOperation, terminalPlan])
+      await pool.query(`INSERT INTO mcp_read_leases(id,connection_id,expires_at)
+        SELECT $1,connection_id,now()-interval '1 minute' FROM mcp_plans WHERE id=$2`, [readLease, interruptedPlan])
+      await pool.query(`INSERT INTO mcp_routing_profiles(id,connection_id,config,enabled,grant_expires_at,credential_id,generation,policy_version)
+        SELECT $1,p.connection_id,'{}',false,now(),$2,p.generation,p.policy_version
+        FROM mcp_plans p WHERE id=$3`, [recoveryProfile, binding.principal.credentialId, interruptedPlan])
+      await pool.query("INSERT INTO mcp_recovery_intents(operation_id,state,authority_profile_id) VALUES($1,'intent',$2)", [terminalOperation, recoveryProfile])
+      const maintenanceState = async () => (await pool.query(`SELECT
+        (SELECT state FROM mcp_operations WHERE id=$1) AS operation,
+        EXISTS(SELECT 1 FROM mcp_read_leases WHERE id=$2) AS lease,
+        (SELECT state FROM mcp_recovery_intents WHERE operation_id=$3) AS intent,
+        (SELECT count(*)::int FROM mcp_audit WHERE action='operation.interrupted' AND payload->>'operationId'=$1::uuid::text) AS audit`,
+        [operation, readLease, terminalOperation])).rows[0]
+      const planStates = async () => Object.fromEntries((await pool.query(
+        "SELECT id,state FROM mcp_plans WHERE id=ANY($1::uuid[])", [fixturePlans])).rows.map(row => [row.id, row.state]))
+      const archiveCount = async () => (await pool.query(
+        "SELECT count(*)::int AS n FROM server_poll_archives WHERE receipt_id=$1", [receiptId])).rows[0].n
+      const blocker = await pool.connect()
+      try {
+        await blocker.query("BEGIN")
+        assert.equal((await blocker.query("SELECT id FROM mcp_plans WHERE id=$1 FOR UPDATE", [lockedPlan])).rowCount, 1)
+        assert.equal((await blocker.query("SELECT id FROM mcp_operations WHERE id=$1 FOR UPDATE", [operation])).rowCount, 1)
+        assert.equal((await blocker.query("SELECT id FROM mcp_read_leases WHERE id=$1 FOR UPDATE", [readLease])).rowCount, 1)
+        assert.equal((await blocker.query("SELECT operation_id FROM mcp_recovery_intents WHERE operation_id=$1 FOR UPDATE", [terminalOperation])).rowCount, 1)
+        await tickMcpWorkers()
+        assert.deepEqual(await planStates(), { [lockedPlan]: "draft", [freePlan]: "expired", [interruptedPlan]: "executing", [terminalPlan]: "completed" })
+        assert.deepEqual(await maintenanceState(), { operation: "running", lease: true, intent: "intent", audit: 0 })
+        assert.equal(await archiveCount(), 1)
+      } finally {
+        try { await blocker.query("ROLLBACK") } finally { blocker.release() }
+      }
       await tickMcpWorkers()
-      assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_poll_archives WHERE receipt_id=$1", [receiptId])).rows[0].n, 1)
-      groups.push("actual default worker tick after natural original expiry performs terminal compaction with source connection paused/server disabled")
+      assert.deepEqual(await planStates(), { [lockedPlan]: "expired", [freePlan]: "expired", [interruptedPlan]: "completed", [terminalPlan]: "completed" })
+      assert.deepEqual(await maintenanceState(), { operation: "unknown", lease: false, intent: "unknown", audit: 1 })
+      assert.equal(await archiveCount(), 1)
+      groups.push("default worker archives a naturally expired POLL source despite four locked maintenance rows and processes deferred work without duplicate archival on the next tick")
     } else if (lostAckVariant) {
       let commits = 0
       const commands: string[] = [], releases: boolean[] = []
@@ -235,6 +283,27 @@ async function main() {
       commitAckLossAccepted = true
       groups.push("after natural original expiry, real claim COMMIT ACK is known and real archive COMMIT succeeds but loses caller ACK; generic unknown destroys backend without rollback/retry")
     } else {
+      const { PARTITION_WRITER_ADVISORY_KEY } = await import("./partition-write-transaction.js")
+      const blockedKeys = [
+        { text: "SELECT pg_advisory_xact_lock($1::bigint)", value: PARTITION_WRITER_ADVISORY_KEY, scanned: 0 },
+        ...["mcp-operation:" + a.execution_id, "server-poll-execution-v1:mcp:" + a.execution_id,
+          "server-poll-receipt-v1:" + receiptId].map(value => ({
+          text: "SELECT pg_advisory_xact_lock(hashtextextended($1,0))", value, scanned: 1,
+        })),
+      ]
+      for (const { text, value, scanned } of blockedKeys) {
+        const blocker = await pool.connect()
+        try {
+          await blocker.query("BEGIN")
+          await blocker.query(text, [value])
+          assert.deepEqual(await compactCompletedServerPollStorage(),
+            { scanned, compacted: 0, skipped: scanned, automaticReplay: false })
+        } finally {
+          try { await blocker.query("ROLLBACK") } finally { blocker.release() }
+        }
+        assert.equal(await captureOriginal(receiptId), beforeSource, "Lock deferral cannot consume the claim")
+      }
+      groups.push("exclusive partition writer and each exact operation, execution and receipt lock defer before claim")
       const compacted = await compactCompletedServerPollStorage()
       assert.deepEqual(compacted, { scanned: 1, compacted: 1, skipped: 0, automaticReplay: false })
       groups.push("natural original plan/admission/activity expiry plus60s passes; production two-transaction one-shot compaction archives one actual MCP source")