From 2ac36a6ca5f36bdb3dd2e5827c8034926cde7a0e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 13 Nov 2026 19:10:50 +0700 Subject: [PATCH] =?UTF-8?q?fix(mcp):=20=D0=B7=D0=B0=D1=89=D0=B8=D1=82?= =?UTF-8?q?=D0=B8=D1=82=D1=8C=20=D0=BE=D0=BF=D0=B5=D1=80=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D0=B8=20=D0=BE=D0=B1=D1=81=D0=BB=D1=83=D0=B6=D0=B8=D0=B2=D0=B0?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=20=D0=BE=D1=82=20=D0=B1=D0=BB=D0=BE=D0=BA?= =?UTF-8?q?=D0=B8=D1=80=D0=BE=D0=B2=D0=BE=D0=BA=20=D1=81=D1=82=D1=80=D0=BE?= =?UTF-8?q?=D0=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../mcp/routing-community-scoped-recovery.ts | 14 +++- backend/src/modules/mcp/runtime.ts | 37 ++++++++-- backend/src/modules/mcp/store.ts | 20 +++-- backend/src/routes/mcp.ts | 30 +++++--- ...server-poll-terminal-archive.owned.test.ts | 73 ++++++++++++++++++- 5 files changed, 145 insertions(+), 29 deletions(-) diff --git a/backend/src/modules/mcp/routing-community-scoped-recovery.ts b/backend/src/modules/mcp/routing-community-scoped-recovery.ts index c944701a..4c52e000 100644 --- a/backend/src/modules/mcp/routing-community-scoped-recovery.ts +++ b/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`)) } diff --git a/backend/src/modules/mcp/runtime.ts b/backend/src/modules/mcp/runtime.ts index f3c7b9ed..d5871986 100644 --- a/backend/src/modules/mcp/runtime.ts +++ b/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 { 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 client.query("UPDATE mcp_plans SET state='expired' WHERE expires_at client.query("DELETE FROM mcp_read_leases WHERE expires_at 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 client.query(`WITH expired AS ( + SELECT id FROM mcp_read_leases WHERE expires_at 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`)) } diff --git a/backend/src/modules/mcp/store.ts b/backend/src/modules/mcp/store.ts index ac5d7e0d..92d65f95 100644 --- a/backend/src/modules/mcp/store.ts +++ b/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(work: (client: PoolClient) => Promise): 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(work: (client: PoolClient) => Promise): Promise { + // 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, actorId: string, action: string, connectionId: string | null, payload: unknown): Promise { 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>, 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("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) diff --git a/backend/src/routes/mcp.ts b/backend/src/routes/mcp.ts index 60666f22..63972d69 100644 --- a/backend/src/routes/mcp.ts +++ b/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 } }) diff --git a/backend/src/services/server-poll-terminal-archive.owned.test.ts b/backend/src/services/server-poll-terminal-archive.owned.test.ts index 5a9fbcc0..5706be67 100644 --- a/backend/src/services/server-poll-terminal-archive.owned.test.ts +++ b/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")