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 <noreply@paperclip.ing>

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
 

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<void>(resolve => { held = resolve; });
+    const gate = new Promise<void>(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<typeof port.settle>[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);

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

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();