From 89080f28a5dcbdc4843c3975a4d60323bbc2b41c Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 6 Oct 2026 23:56:40 -0500 Subject: [PATCH] fix(dot): order mailbox receipts and retain fence acknowledgements Serialize mailbox writes and cursor reads with the binding lock. Permit paused connections to receive and acknowledge fences while task tools remain denied. Co-Authored-By: Paperclip --- doc/openai-dot-runner.md | 4 ++ server/src/__tests__/dot-runner.test.ts | 59 +++++++++++++++++++++++- server/src/services/dot-runner-broker.ts | 24 ++++++++-- server/src/services/public-mcp/oauth.ts | 4 +- 4 files changed, 83 insertions(+), 8 deletions(-) diff --git a/doc/openai-dot-runner.md b/doc/openai-dot-runner.md index 23b42d8511a..1443bb3f7db 100644 --- a/doc/openai-dot-runner.md +++ b/doc/openai-dot-runner.md @@ -73,6 +73,10 @@ enqueue another run or move the same request to another task. A Dot runs one assignment at a time; competing tasks stay queued regardless of the agent's configured concurrency. During setup the connection panel continues checking for a verified event subscription without requiring a manual refresh. +Mailbox writers and cursor reads serialize through the binding row so a cursor +cannot skip a tool result whose transaction commits later. A paused Dot can +receive fence notices and acknowledge them, but cannot read tasks or execute +tools; those narrow inbox reads leave its task cursor unchanged. ## Limits and recovery diff --git a/server/src/__tests__/dot-runner.test.ts b/server/src/__tests__/dot-runner.test.ts index d455f4d8dd5..e60cf5ef56c 100644 --- a/server/src/__tests__/dot-runner.test.ts +++ b/server/src/__tests__/dot-runner.test.ts @@ -9,7 +9,7 @@ import { and, eq, sql } from "drizzle-orm"; import express, { type Request } from "express"; import request from "supertest"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, companyMemberships, createDb, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations, +import { agents, authUsers, companies, companyMemberships, createDb, dotAgentBindings, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations, heartbeatRuns, agentWakeupRequests, issues, nativeRunResults, nativeRunFinalizations, completionContracts, mcpEventDeliveries, workspaceOperations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createPublicMcpOAuth, DEVICE_GRANT } from "../services/public-mcp/oauth.js"; @@ -18,7 +18,7 @@ import { publicMcpIngressRoutes, publicMcpManagementRoutes } from "../routes/pub import { createPublicMcpEvents, type EventFetch } from "../services/public-mcp/events.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { canonicalNativeRuntimeContextDigest, type StrictCompletionContractInput } from "../vendor/paperclip-runner/index.js"; -import { dotRunnerBroker } from "../services/dot-runner-broker.js"; +import { createDotRunnerMcpTools, dotRunnerBroker } from "../services/dot-runner-broker.js"; import { prepareNativeHeartbeatRun } from "../services/native-runtime/prepare-native-run.js"; import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js"; import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; @@ -169,6 +169,61 @@ describe("durable Dot Runner integration", () => { } }, 30000); + it("serializes tool-result writes and cursor reads with the binding mailbox lock", async () => { + const f = await fixture(); + const { run, assignment } = await offeredWork(f); + const requestId = randomUUID(); + await db.insert(dotRunnerOperations).values({ companyId: f.company.id, assignmentId: assignment.id, requestId, + digest: "test", command: { action: "tool" }, status: "admitted" }); + let release!: () => void; + let held!: () => void; + const locked = new Promise(resolve => { held = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const firstEvent = randomUUID(); + const transaction = db.transaction(async tx => { + await tx.select().from(dotAgentBindings).where(eq(dotAgentBindings.id, f.snapshot.bindingId)).for("update"); + await tx.insert(dotMailboxItems).values({ companyId: f.company.id, bindingId: f.snapshot.bindingId, + bindingGeneration: f.snapshot.bindingGeneration, kind: "operation_result", sourceEventId: firstEvent, references: {} }); + held(); await gate; + }); + await locked; + const port = f.broker.port({ binding: { companyId: f.company.id, agentId: f.agent.id, runId: run.id }, provider: { binding: f.snapshot } }); + let written = false, read = false; + const secondEvent = randomUUID(); + const writer = port.settle({ sourceEventId: secondEvent, payload: { requestId, binding: { ...f.snapshot, runId: run.id, + normalizedSessionId: assignment.normalizedSessionId, turnId: assignment.turnId, assignmentRevision: assignment.revision }, outcome: { status: "completed", result: {} } } } as Parameters[0]).then(() => { written = true; }); + const reader = f.broker.mailbox(f.principal).then(value => { read = true; return value; }); + try { + await new Promise(resolve => setTimeout(resolve, 30)); + expect(written).toBe(false); expect(read).toBe(false); + } finally { release(); } + await Promise.all([transaction, writer]); + const initial = await reader; + const next = await f.broker.mailbox(f.principal, initial.nextCursor); + expect([...initial.items, ...next.items].map(i => i.sourceEventId)).toEqual(expect.arrayContaining([firstEvent, secondEvent])); + await f.events.unsubscribe(f.principal, f.subscription); await f.events.stop(); + }, 30000); + + it("allows paused Dot fence delivery and acknowledgements without task authority", async () => { + const f = await fixture(); + const { assignment } = await offeredWork(f); + await db.update(dotRunnerAssignments).set({ status: "fenced" }).where(eq(dotRunnerAssignments.id, assignment.id)); + await db.insert(dotMailboxItems).values({ companyId: f.company.id, bindingId: f.snapshot.bindingId, + bindingGeneration: f.snapshot.bindingGeneration, assignmentId: assignment.id, kind: "authority_revoked", sourceEventId: randomUUID(), references: { assignmentId: assignment.id } }); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id)); + const principal = await f.oauth.authenticate(f.tokens.access_token); + const tools = createDotRunnerMcpTools(db); + await f.events.tick(); + expect(f.received.at(-1)?.data.kind).toBe("authority_revoked"); + const inbox = await tools.callTool(principal, "paperclip_dot_inbox", { after: 0 }); + expect(inbox).toMatchObject({ result: { nextCursor: 0, items: [{ kind: "authority_revoked" }] } }); + await expect(tools.callTool(principal, "paperclip_dot_tasks", {})).rejects.toThrow("authority"); + await expect(tools.callTool(principal, "paperclip_dot_read", { assignmentId: assignment.id })).rejects.toThrow("authority"); + await expect(tools.callTool(principal, "paperclip_dot_accept", { assignmentId: assignment.id, requestId: randomUUID() })).rejects.toThrow("authority"); + expect(await tools.callTool(principal, "paperclip_dot_control_ack", { assignmentId: assignment.id, requestId: randomUUID() })).toMatchObject({ result: { status: "acknowledged", externalStopConfirmed: false } }); + await f.events.unsubscribe(principal, f.subscription); await f.events.stop(); + }, 30000); + it("verifies reconnects and wakes the same outstanding assignment after exhausted delivery", async () => { const f = await fixture(); const { assignment } = await offeredWork(f); diff --git a/server/src/services/dot-runner-broker.ts b/server/src/services/dot-runner-broker.ts index fd62fdc320b..efdd2279834 100644 --- a/server/src/services/dot-runner-broker.ts +++ b/server/src/services/dot-runner-broker.ts @@ -148,15 +148,26 @@ function createBroker(db: Db) { acceptByUnixMs: Date.now() + 10 * 60_000, expiresAtUnixMs: Date.now() + 2 * 60 * 60_000 }; }, async authorizeBinding(principal: McpPrincipal, companyId: string, bindingId: string) { - const b = await principalBinding(principal, false); + const b = await principalBinding(principal, false, true); if (b.companyId !== companyId || b.id !== bindingId) throw fail("Binding is outside this connection."); return b; }, async mailbox(principal: McpPrincipal, after = 0) { - const b = await principalBinding(principal, false); - const items = await db.select().from(mailbox).where(and(eq(mailbox.companyId, b.companyId), eq(mailbox.bindingId, b.id), - eq(mailbox.bindingGeneration, b.generation), gt(mailbox.id, after))).orderBy(asc(mailbox.id)).limit(50); - return { bindingId: b.id, generation: b.generation, items, nextCursor: items.at(-1)?.id ?? after }; + const binding = await principalBinding(principal, false, true); + return db.transaction(async tx => { + // Every writer takes this lock before allocating a mailbox ID. Reading + // under it cannot advance a cursor past an earlier uncommitted item. + const [b] = await tx.select().from(bindings).where(and(eq(bindings.id, binding.id), + eq(bindings.generation, binding.generation), isNull(bindings.revokedAt))).for("update"); + if (!b) throw fail("Binding authority is unavailable."); + const [agent] = await tx.select().from(agents).where(eq(agents.id, b.agentId)); + if (!agent || ["terminated", "pending_approval"].includes(agent.status)) throw fail("Agent connection authority is unavailable."); + const paused = agent.status === "paused"; + const items = await tx.select().from(mailbox).where(and(eq(mailbox.companyId, b.companyId), eq(mailbox.bindingId, b.id), + eq(mailbox.bindingGeneration, b.generation), gt(mailbox.id, after), paused ? eq(mailbox.kind, "authority_revoked") : undefined)).orderBy(asc(mailbox.id)).limit(50); + // Paused reads expose fences only and never consume hidden task items. + return { bindingId: b.id, generation: b.generation, items, nextCursor: paused ? after : items.at(-1)?.id ?? after }; + }); }, async read(principal: McpPrincipal, assignmentId: string) { const { assignment } = await authorizeAssignment(principal, assignmentId); @@ -363,6 +374,9 @@ function createBroker(db: Db) { const p = z.object({ requestId: z.uuid(), binding: z.object({ runId: z.literal(runId), bindingId: z.literal(execution.provider.binding.bindingId), bindingGeneration: z.literal(execution.provider.binding.bindingGeneration) }).passthrough(), outcome: z.record(z.string(), z.unknown()) }).parse(event.payload); await db.transaction(async tx => { + // Binding precedes assignment everywhere: serialize mailbox IDs + // with dispatch, readiness and reconnect without reversing locks. + await tx.select({ id: bindings.id }).from(bindings).where(eq(bindings.id, p.binding.bindingId)).for("update"); const [a] = await tx.select().from(assignments).where(and(eq(assignments.runId, runId), eq(assignments.bindingGeneration, p.binding.bindingGeneration))).for("update"); if (!a) throw fail("Runner operation has no assignment."); if (p.binding.companyId !== a.companyId || p.binding.agentId !== a.agentId || p.binding.normalizedSessionId !== a.normalizedSessionId diff --git a/server/src/services/public-mcp/oauth.ts b/server/src/services/public-mcp/oauth.ts index 079bf295577..2c5c8fd90d9 100644 --- a/server/src/services/public-mcp/oauth.ts +++ b/server/src/services/public-mcp/oauth.ts @@ -94,7 +94,9 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { if (membership.membershipRole === "viewer") throw invalidGrant(); if (!grant.agentId) return { type: "none", source: "mcp_oauth", companyId: grant.companyId }; const [agent] = await queryDb.select().from(agents).where(and(eq(agents.id, grant.agentId), eq(agents.companyId, grant.companyId))); - if (!agent || ["paused", "terminated", "pending_approval"].includes(agent.status)) throw invalidGrant(); + // Paused Dot connections retain only the broker's fence inbox and ack + // authority. Each task/tool method independently rejects paused agents. + if (!agent || ["terminated", "pending_approval"].includes(agent.status)) throw invalidGrant(); return { type: "agent", source: "mcp_oauth", agentId: agent.id, companyId: grant.companyId }; } if (grant.purpose !== "personal") throw invalidGrant();