merge: integrate Dot mailbox and fence fixes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/openai-dot-runner.md

@@ -75,6 +75,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";
@@ -201,6 +201,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

@@ -153,15 +153,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);
@@ -368,6 +379,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

@@ -97,7 +97,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();